Notification feed for the hypervisor

tests/handlers/api/hypervisor/test_notification.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python
+#
+# Copyright (C) 2016 GNS3 Technologies Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import json
+
+from gns3server.hypervisor.notification_manager import NotificationManager
+
+
+def test_notification_ws(http_hypervisor, async_run):
+    ws = http_hypervisor.websocket("/notifications/ws")
+    answer = async_run(ws.receive())
+    answer = json.loads(answer.data)
+    assert answer["action"] == "ping"
+
+    NotificationManager.instance().emit("test", {})
+
+    answer = async_run(ws.receive())
+    answer = json.loads(answer.data)
+    assert answer["action"] == "test"
+
+    async_run(http_hypervisor.close())

tests/handlers/api/hypervisor/test_project.py

@@ -199,47 +199,6 @@ def test_close_project_invalid_uuid(http_hypervisor):
     assert response.status == 404
 
 
-def test_notification(http_hypervisor, project, loop):
-    @asyncio.coroutine
-    def go(future):
-        response = yield from aiohttp.request("GET", http_hypervisor.get_url("/projects/{project_id}/notifications".format(project_id=project.id)))
-        response.body = yield from response.content.read(200)
-        project.emit("vm.created", {"a": "b"})
-        response.body += yield from response.content.read(50)
-        response.close()
-        future.set_result(response)
-
-    future = asyncio.Future()
-    asyncio.async(go(future))
-    response = loop.run_until_complete(future)
-    assert response.status == 200
-    assert b'"action": "ping"' in response.body
-    assert b'"cpu_usage_percent"' in response.body
-    assert b'{"action": "vm.created", "event": {"a": "b"}}\n' in response.body
-
-
-def test_notification_invalid_id(http_hypervisor):
-    response = http_hypervisor.get("/projects/{project_id}/notifications".format(project_id=uuid.uuid4()))
-    assert response.status == 404
-
-
-def test_list_files(http_hypervisor, project):
-    files = [
-        {
-            "path": "test.txt",
-            "md5sum": "ad0234829205b9033196ba818f7a872b"
-        },
-        {
-            "path": "vm-1/dynamips/test.bin",
-            "md5sum": "098f6bcd4621d373cade4e832627b4f6"
-        }
-    ]
-    with asyncio_patch("gns3server.hypervisor.project.Project.list_files", return_value=files) as mock:
-        response = http_hypervisor.get("/projects/{project_id}/files".format(project_id=project.id), example=True)
-        assert response.status == 200
-        assert response.json == files
-
-
 def test_get_file(http_hypervisor, tmpdir):
 
     with patch("gns3server.config.Config.get_section_config", return_value={"project_directory": str(tmpdir)}):