integration: Add tests for events

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

integration/test_client.py

@@ -13,10 +13,15 @@
 #    under the License.
 
 import os
+import threading
+
+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 +62,76 @@ 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 = None
+
+        def get_event(): # Receive one event
+            global message
+            message = events_ws_client.recv()
+
+        receiver = threading.Thread(target=get_event)
+        receiver.start()
+
+        self.client.instances.all() # provoke an event
+
+        receiver.join()
+
+        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")
+
+        message = None
+
+        def get_event(): # Receive one event
+            global message
+            message = events_ws_client.recv()
+
+        receiver = threading.Thread(target=get_event)
+        receiver.start()
+
+        self.client.instances.all() # provoke an event
+
+        receiver.join()
+
+        self.assertEqual(type(message), dict)
+        self.assertTrue("timestamp" in message.keys())
+        self.assertTrue("metadata" in message.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()