integration: Add tests for events

Signed-off-by: hamistao <[email protected]>

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()