Notif forwarded from hypervisor to controller

docs/general.rst

@@ -209,7 +209,9 @@ The available notification are:
     * vm.created
     * vm.started
     * vm.stopped
+    * vm.deleted
     * log.error
+    * log.warning
 
 Previous versions
 =================

gns3server/controller/__init__.py

@@ -45,7 +45,7 @@ class Controller:
         :param kwargs: See the documentation of Hypervisor
         """
         if hypervisor_id not in self._hypervisors:
-            hypervisor = Hypervisor(hypervisor_id=hypervisor_id, **kwargs)
+            hypervisor = Hypervisor(hypervisor_id=hypervisor_id, controller=self, **kwargs)
             self._hypervisors[hypervisor_id] = hypervisor
         return self._hypervisors[hypervisor_id]
 
@@ -109,3 +109,18 @@ class Controller:
         if not hasattr(Controller, '_instance') or Controller._instance is None:
             Controller._instance = Controller()
         return Controller._instance
+
+    def emit(self, action, event, **kwargs):
+        """
+        Send a notification to clients scoped by projects
+        """
+
+        if "project_id" in kwargs:
+            try:
+                project_id = kwargs.pop("project_id")
+                self._projects[project_id].emit(action, event, **kwargs)
+            except KeyError:
+                pass
+        else:
+            for project in self._projects.values():
+                project.emit(action, event, **kwargs)

gns3server/controller/hypervisor.py

@@ -38,7 +38,8 @@ class Hypervisor:
     A GNS3 hypervisor.
     """
 
-    def __init__(self, hypervisor_id, protocol="http", host="localhost", port=8000, user=None, password=None):
+    def __init__(self, hypervisor_id, controller=None, protocol="http", host="localhost", port=8000, user=None, password=None):
+        assert controller is not None
         log.info("Create hypervisor %s", hypervisor_id)
         self._id = hypervisor_id
         self._protocol = protocol
@@ -48,15 +49,19 @@ class Hypervisor:
         self._password = None
         self._setAuth(user, password)
         self._connected = False
-        # The remote hypervisor version
-        # TODO: For the moment it's fake we return the controller version
-        self._version = __version__
+        self._controller = controller
+        self._session = aiohttp.ClientSession()
 
         # If the hypervisor is local but the hypervisor id is local
         # it's a configuration issue
         if hypervisor_id == "local" and Config.instance().get_section_config("Server")["local"] is False:
             raise HypervisorError("The local hypervisor is started without --local")
 
+        asyncio.async(self._connect())
+
+    def __del__(self):
+        self._session.close()
+
     def _setAuth(self, user, password):
         """
         Set authentication parameters
@@ -110,6 +115,15 @@ class Hypervisor:
 
     @asyncio.coroutine
     def httpQuery(self, method, path, data=None):
+        if not self._connected:
+            yield from self._connect()
+        return (yield from self._runHttpQuery(method, path, data=data))
+
+    @asyncio.coroutine
+    def _connect(self):
+        """
+        Check if remote server is accessible
+        """
         if not self._connected:
             response = yield from self._runHttpQuery("GET", "/version")
             if "version" not in response.json:
@@ -117,45 +131,64 @@ class Hypervisor:
             if parse_version(__version__)[:2] != parse_version(response.json["version"])[:2]:
                 raise aiohttp.web.HTTPConflict(text="The server {} versions are not compatible {} != {}".format(self._id, __version__, response.json["version"]))
 
-        self._connected = True
-        return (yield from self._runHttpQuery(method, path, data=data))
+            self._notifications = asyncio.async(self._connectNotification())
+
+            self._connected = True
+
+    @asyncio.coroutine
+    def _connectNotification(self):
+        """
+        Connect to the notification stream
+        """
+        ws = yield from self._session.ws_connect(self._getUrl("/notifications/ws"), auth=self._auth)
+        while True:
+            response = yield from ws.receive()
+            if response.tp == aiohttp.MsgType.closed or response.tp == aiohttp.MsgType.error:
+                self._connected = False
+                break
+            msg = json.loads(response.data)
+            action = msg.pop("action")
+            event = msg.pop("event")
+            self._controller.emit(action, event, hypervisor_id=self.id, **msg)
+
+    def _getUrl(self, path):
+        return "{}://{}:{}/v2/hypervisor{}".format(self._protocol, self._host, self._port, path)
 
     @asyncio.coroutine
     def _runHttpQuery(self, method, path, data=None):
         with aiohttp.Timeout(10):
-            with aiohttp.ClientSession() as session:
-                url = "{}://{}:{}/v2/hypervisor{}".format(self._protocol, self._host, self._port, path)
-                headers = {'content-type': 'application/json'}
-                if data:
-                    if hasattr(data, '__json__'):
-                        data = data.__json__()
-                    data = json.dumps(data)
-                response = yield from session.request(method, url, headers=headers, data=data, auth=self._auth)
-                body = yield from response.read()
-                if body:
-                    body = body.decode()
+            url = self._getUrl(path)
+            headers = {'content-type': 'application/json'}
+            if data:
+                if hasattr(data, '__json__'):
+                    data = data.__json__()
+                data = json.dumps(data)
+            response = yield from self._session.request(method, url, headers=headers, data=data, auth=self._auth)
+            body = yield from response.read()
+            if body:
+                body = body.decode()
 
-                if response.status >= 300:
-                    if response.status == 400:
-                        raise aiohttp.web.HTTPBadRequest(text="Bad request {} {}".format(url, body))
-                    elif response.status == 401:
-                        raise aiohttp.web.HTTPUnauthorized(text="Invalid authentication for hypervisor {}".format(self.id))
-                    elif response.status == 403:
-                        raise aiohttp.web.HTTPForbidden(text="Forbidden {} {}".format(url, body))
-                    elif response.status == 404:
-                        raise aiohttp.web.HTTPNotFound(text="{} not found on hypervisor".format(url))
-                    elif response.status == 409:
-                        raise aiohttp.web.HTTPConflict(text="Conflict {} {}".format(url, body))
-                    else:
-                        raise NotImplemented("{} status code is not supported".format(e.status))
-                if body and len(body):
-                    try:
-                        response.json = json.loads(body)
-                    except json.JSONDecodeError:
-                        raise aiohttp.web.HTTPConflict(text="The server {} is not a GNS3 server".format(self._id))
-                if response.json is None:
-                    response.json = {}
-                return response
+            if response.status >= 300:
+                if response.status == 400:
+                    raise aiohttp.web.HTTPBadRequest(text="Bad request {} {}".format(url, body))
+                elif response.status == 401:
+                    raise aiohttp.web.HTTPUnauthorized(text="Invalid authentication for hypervisor {}".format(self.id))
+                elif response.status == 403:
+                    raise aiohttp.web.HTTPForbidden(text="Forbidden {} {}".format(url, body))
+                elif response.status == 404:
+                    raise aiohttp.web.HTTPNotFound(text="{} not found on hypervisor".format(url))
+                elif response.status == 409:
+                    raise aiohttp.web.HTTPConflict(text="Conflict {} {}".format(url, body))
+                else:
+                    raise NotImplementedError("{} status code is not supported".format(response.status))
+            if body and len(body):
+                try:
+                    response.json = json.loads(body)
+                except json.JSONDecodeError:
+                    raise aiohttp.web.HTTPConflict(text="The server {} is not a GNS3 server".format(self._id))
+            if response.json is None:
+                response.json = {}
+            return response
 
     @asyncio.coroutine
     def post(self, path, data={}):

gns3server/controller/link.py

@@ -42,14 +42,14 @@ class Link:
         """
         Create the link
         """
-        raise NotImplemented
+        raise NotImplementedError
 
     @asyncio.coroutine
     def delete(self):
         """
         Delete the link
         """
-        raise NotImplemented
+        raise NotImplementedError
 
     @property
     def id(self):

gns3server/hypervisor/base_vm.py

@@ -218,6 +218,7 @@ class BaseVM:
         log.info("{module}: {name} [{id}] created".format(module=self.manager.module_name,
                                                           name=self.name,
                                                           id=self.id))
+        self._project.emit("vm.created", self)
 
     @asyncio.coroutine
     def delete(self):
@@ -225,6 +226,7 @@ class BaseVM:
         Delete the VM (including all its files).
         """
 
+        self._project.emit("vm.deleted", self)
         directory = self.project.vm_working_directory(self)
         if os.path.exists(directory):
             try:

tests/controller/test_controller.py

@@ -77,7 +77,7 @@ def test_removeProject(controller, async_run):
 def test_addProject_with_hypervisor(controller, async_run):
     uuid1 = str(uuid.uuid4())
 
-    hypervisor = Hypervisor("test1")
+    hypervisor = Hypervisor("test1", controller=MagicMock())
     hypervisor.post = MagicMock()
     controller._hypervisors = {"test1": hypervisor}
 
@@ -92,3 +92,48 @@ def test_getProject(controller, async_run):
     assert controller.getProject(uuid1) == project
     with pytest.raises(aiohttp.web.HTTPNotFound):
         assert controller.getProject("dsdssd")
+
+
+def test_emit(controller, async_run):
+    project1 = MagicMock()
+    uuid1 = str(uuid.uuid4())
+    controller._projects[uuid1] = project1
+
+    project2 = MagicMock()
+    uuid2 = str(uuid.uuid4())
+    controller._projects[uuid2] = project2
+
+    # Notif without project should be send to all projects
+    controller.emit("test", {})
+    assert project1.emit.called
+    assert project2.emit.called
+
+
+def test_emit_to_project(controller, async_run):
+    project1 = MagicMock()
+    uuid1 = str(uuid.uuid4())
+    controller._projects[uuid1] = project1
+
+    project2 = MagicMock()
+    uuid2 = str(uuid.uuid4())
+    controller._projects[uuid2] = project2
+
+    # Notif with project should be send to this project
+    controller.emit("test", {}, project_id=uuid1)
+    project1.emit.assert_called_with('test', {})
+    assert not project2.emit.called
+
+
+def test_emit_to_project_not_exists(controller, async_run):
+    project1 = MagicMock()
+    uuid1 = str(uuid.uuid4())
+    controller._projects[uuid1] = project1
+
+    project2 = MagicMock()
+    uuid2 = str(uuid.uuid4())
+    controller._projects[uuid2] = project2
+
+    # Notif with project should be send to this project
+    controller.emit("test", {}, project_id="4444444")
+    assert not project1.emit.called
+    assert not project2.emit.called

tests/controller/test_hypervisor.py

@@ -19,6 +19,7 @@
 import pytest
 import json
 import aiohttp
+import asyncio
 from unittest.mock import patch, MagicMock
 
 from gns3server.controller.project import Project
@@ -29,7 +30,7 @@ from tests.utils import asyncio_patch, AsyncioMagicMock
 
 @pytest.fixture
 def hypervisor():
-    hypervisor = Hypervisor("my_hypervisor_id", protocol="https", host="example.com", port=84)
+    hypervisor = Hypervisor("my_hypervisor_id", protocol="https", host="example.com", port=84, controller=MagicMock())
     hypervisor._connected = True
     return hypervisor
 
@@ -46,10 +47,10 @@ def test_hypervisor_local(hypervisor):
 
     with patch("gns3server.config.Config.get_section_config", return_value={"local": False}):
         with pytest.raises(HypervisorError):
-            s = Hypervisor("local")
+            s = Hypervisor("local", controller=MagicMock())
 
     with patch("gns3server.config.Config.get_section_config", return_value={"local": True}):
-        s = Hypervisor("test")
+        s = Hypervisor("test", controller=MagicMock())
 
 
 def test_hypervisor_httpQuery(hypervisor, async_run):
@@ -139,6 +140,35 @@ def test_hypervisor_httpQuery_project(hypervisor, async_run):
         mock.assert_called_with("POST", "https://example.com:84/v2/hypervisor/projects", data=json.dumps(project.__json__()), headers={'content-type': 'application/json'}, auth=None)
 
 
+def test_connectNotification(hypervisor, async_run):
+    ws_mock = AsyncioMagicMock()
+
+    call = 0
+
+    @asyncio.coroutine
+    def receive():
+        nonlocal call
+        call += 1
+        if call == 1:
+            response = MagicMock()
+            response.data = '{"action": "test", "event": {"a": 1}, "project_id": "42"}'
+            response.tp = aiohttp.MsgType.text
+            return response
+        else:
+            response = MagicMock()
+            response.tp = aiohttp.MsgType.closed
+            return response
+
+    hypervisor._controller = MagicMock()
+    hypervisor._session = AsyncioMagicMock(return_value=ws_mock)
+    hypervisor._session.ws_connect = AsyncioMagicMock(return_value=ws_mock)
+    ws_mock.receive = receive
+    async_run(hypervisor._connectNotification())
+
+    hypervisor._controller.emit.assert_called_with('test', {'a': 1}, hypervisor_id=hypervisor.id, project_id='42')
+    assert hypervisor._connected is False
+
+
 def test_json(hypervisor):
     hypervisor.user = "test"
     assert hypervisor.__json__() == {

tests/controller/test_link.py

@@ -16,6 +16,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import pytest
+from unittest.mock import MagicMock
 
 from gns3server.controller.link import Link
 from gns3server.controller.vm import VM
@@ -30,7 +31,7 @@ def project():
 
 @pytest.fixture
 def hypervisor():
-    return Hypervisor("example.com")
+    return Hypervisor("example.com", controller=MagicMock())
 
 
 def test_addVM(async_run, project, hypervisor):