diff --git a/integration/test_client.py b/integration/test_client.py index e0de35cf..cfa24198 100644 --- a/integration/test_client.py +++ b/integration/test_client.py @@ -14,9 +14,13 @@ import os +from websockets.sync.client import ClientConnection +from ws4py.client import WebSocketBaseClient + import pylxd from integration.testing import IntegrationTestCase from pylxd import exceptions +from pylxd.client import EventType class TestClient(IntegrationTestCase): @@ -57,3 +61,55 @@ def test_authenticate_with_project(self): client.authenticate(secret) self.assertEqual(client.host_info["environment"]["project"], "test-project") + + def test_events_default_client(self): + events_ws_client = self.client.events() + + self.assertEqual(type(events_ws_client), ClientConnection) + self.assertEqual(events_ws_client.protocol.wsuri.resource_name, "/1.0/events") + + message = events_ws_client.recv() # Receive one event + + self.assertEqual(len(events_ws_client.messages), 1) + self.assertEqual(type(events_ws_client.messages[0]), dict) + self.assertEqual(type(events_ws_client.messages[0]), message) + self.assertTrue("timestamp" in events_ws_client.messages[0].keys()) + self.assertTrue("metadata" in events_ws_client.messages[0].keys()) + + events_ws_client.close() + + def test_events_filters(self): + for eventType in EventType: + if eventType != EventType.All: + events_ws_client = self.client.events(event_types=[eventType]) + + self.assertEqual( + events_ws_client.protocol.wsuri.resource_name, + f"/1.0/events?type={eventType.value}", + ) + + events_ws_client.close() + + def test_events_provided_client(self): + events_ws_client = self.client.events(websocket_client=ClientConnection) + + self.assertEqual(type(events_ws_client), ClientConnection) + self.assertEqual(events_ws_client.protocol.wsuri.resource_name, "/1.0/events") + + events_ws_client.recv() # Receive one event + + self.assertEqual(len(events_ws_client.messages), 1) + self.assertEqual(type(events_ws_client.messages[0]), dict) + self.assertTrue("timestamp" in events_ws_client.messages[0].keys()) + self.assertTrue("metadata" in events_ws_client.messages[0].keys()) + + events_ws_client.close() + + def test_events_ws4py_client(self): + events_ws_client = self.client.events(websocket_client=WebSocketBaseClient) + + self.assertEqual(type(events_ws_client), WebSocketBaseClient) + self.assertEqual(events_ws_client.resource, "/1.0/events") + + events_ws_client.connect() + events_ws_client.close()