Test compute notifications and run tests on macOS and Windows.

tests/endpoints/base.py

@@ -28,13 +28,14 @@ class Query:
     Helper to make queries against the test server
     """
 
-    def __init__(self, http_client, prefix='', api_version=None):
+    def __init__(self, http_client, ws_client, prefix='', api_version=None):
         """
         :param prefix: Prefix added before path (ex: /compute)
         :param api_version: Version of the API
         """
 
         self._http_client = http_client
+        self._ws_client = ws_client
         self._prefix = prefix
         self._api_version = api_version
 
@@ -53,6 +54,10 @@ class Query:
     def patch(self, path, **kwargs):
         return self._request("PATCH", path, **kwargs)
 
+    def ws(self, path):
+
+        return self._ws_client.websocket_connect(self.get_url(path))
+
     def get_url(self, path):
 
         if self._api_version is None:

tests/endpoints/compute/test_notifications.py

@@ -21,20 +21,18 @@ import json
 from gns3server.compute.notification_manager import NotificationManager
 
 
-# @pytest.mark.asyncio
-# async def test_notification_ws(compute_api, http_client):
-#
-#     ws = await http_client.ws_connect(compute_api.get_url("/notifications/ws"))
-#     answer = await ws.receive()
-#     answer = json.loads(answer.data)
-#
-#     assert answer["action"] == "ping"
-#
-#     NotificationManager.instance().emit("test", {})
-#
-#     answer = await ws.receive()
-#     answer = json.loads(answer.data)
-#     assert answer["action"] == "test"
-#
-#     if not ws.closed:
-#         await ws.close()
+@pytest.mark.asyncio
+async def test_notification_ws(compute_api):
+
+    with compute_api.ws("/notifications/ws") as ws:
+
+        answer = ws.receive_text()
+        answer = json.loads(answer)
+
+        assert answer["action"] == "ping"
+
+        NotificationManager.instance().emit("test", {})
+
+        answer = ws.receive_text()
+        answer = json.loads(answer)
+        assert answer["action"] == "test"