Save computes to database

tests/api/routes/controller/test_computes.py

@@ -15,12 +15,13 @@
 # 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 uuid
 import pytest
 
 from fastapi import FastAPI, status
 from httpx import AsyncClient
 
-from gns3server.controller import Controller
+from gns3server.schemas.computes import Compute
 
 pytestmark = pytest.mark.asyncio
 
@@ -28,234 +29,167 @@ import unittest
 from tests.utils import asyncio_patch
 
 
-async def test_compute_create_without_id(app: FastAPI, client: AsyncClient, controller: Controller) -> None:
+class TestComputeRoutes:
 
-    params = {
-        "protocol": "http",
-        "host": "localhost",
-        "port": 84,
-        "user": "julien",
-        "password": "secure"}
+    async def test_compute_create(self, app: FastAPI, client: AsyncClient) -> None:
 
-    response = await client.post(app.url_path_for("create_compute"), json=params)
-    assert response.status_code == status.HTTP_201_CREATED
-    response_content = response.json()
-    assert response_content["user"] == "julien"
-    assert response_content["compute_id"] is not None
-    assert "password" not in response_content
-    assert len(controller.computes) == 1
-    assert controller.computes[response_content["compute_id"]].host == "localhost"
+        params = {
+            "protocol": "http",
+            "host": "localhost",
+            "port": 84,
+            "user": "julien",
+            "password": "secure"}
+
+        response = await client.post(app.url_path_for("create_compute"), json=params)
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.json()["compute_id"] is not None
+
+        del params["password"]
+        for param, value in params.items():
+            assert response.json()[param] == value
+
+    async def test_compute_create_with_id(self, app: FastAPI, client: AsyncClient) -> None:
+
+        compute_id = str(uuid.uuid4())
+        params = {
+            "compute_id": compute_id,
+            "protocol": "http",
+            "host": "localhost",
+            "port": 84,
+            "user": "julien",
+            "password": "secure"}
+
+        response = await client.post(app.url_path_for("create_compute"), json=params)
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.json()["compute_id"] == compute_id
+
+        del params["password"]
+        for param, value in params.items():
+            assert response.json()[param] == value
+
+    async def test_compute_list(self, app: FastAPI, client: AsyncClient) -> None:
+
+        response = await client.get(app.url_path_for("get_computes"))
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.json()) > 0
+
+    async def test_compute_get(self, app: FastAPI, client: AsyncClient, test_compute: Compute) -> None:
+
+        response = await client.get(app.url_path_for("get_compute", compute_id=test_compute.compute_id))
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["compute_id"] == str(test_compute.compute_id)
+
+    async def test_compute_update(self, app: FastAPI, client: AsyncClient, test_compute: Compute) -> None:
+
+        params = {
+            "protocol": "http",
+            "host": "localhost",
+            "port": 84,
+            "user": "julien",
+            "password": "secure"
+        }
+
+        response = await client.post(app.url_path_for("create_compute"), json=params)
+        assert response.status_code == status.HTTP_201_CREATED
+        compute_id = response.json()["compute_id"]
+
+        params["protocol"] = "https"
+        response = await client.put(app.url_path_for("update_compute", compute_id=compute_id), json=params)
+        assert response.status_code == status.HTTP_200_OK
+
+        del params["password"]
+        for param, value in params.items():
+            assert response.json()[param] == value
+
+    async def test_compute_delete(self, app: FastAPI, client: AsyncClient, test_compute: Compute) -> None:
+
+        response = await client.delete(app.url_path_for("delete_compute", compute_id=test_compute.compute_id))
+        assert response.status_code == status.HTTP_204_NO_CONTENT
 
 
-async def test_compute_create_with_id(app: FastAPI, client: AsyncClient, controller: Controller) -> None:
+class TestComputeFeatures:
 
-    params = {
-        "compute_id": "my_compute_id",
-        "protocol": "http",
-        "host": "localhost",
-        "port": 84,
-        "user": "julien",
-        "password": "secure"}
+    async def test_compute_list_images(self, app: FastAPI, client: AsyncClient) -> None:
 
-    response = await client.post(app.url_path_for("create_compute"), json=params)
-    assert response.status_code == status.HTTP_201_CREATED
-    assert response.json()["user"] == "julien"
-    assert "password" not in response.json()
-    assert len(controller.computes) == 1
-    assert controller.computes["my_compute_id"].host == "localhost"
+        params = {
+            "protocol": "http",
+            "host": "localhost",
+            "port": 84,
+            "user": "julien",
+            "password": "secure"
+        }
 
+        response = await client.post(app.url_path_for("create_compute"), json=params)
+        assert response.status_code == status.HTTP_201_CREATED
+        compute_id = response.json()["compute_id"]
 
-async def test_compute_get(app: FastAPI, client: AsyncClient, controller: Controller) -> None:
+        with asyncio_patch("gns3server.controller.compute.Compute.images", return_value=[{"filename": "linux.qcow2"}, {"filename": "asav.qcow2"}]) as mock:
+            response = await client.get(app.url_path_for("delete_compute", compute_id=compute_id) + "/qemu/images")
+            assert response.json() == [{"filename": "linux.qcow2"}, {"filename": "asav.qcow2"}]
+            mock.assert_called_with("qemu")
 
-    params = {
-        "compute_id": "my_compute_id",
-        "protocol": "http",
-        "host": "localhost",
-        "port": 84,
-        "user": "julien",
-        "password": "secure"
-    }
+    async def test_compute_list_vms(self, app: FastAPI, client: AsyncClient) -> None:
 
-    response = await client.post(app.url_path_for("create_compute"), json=params)
-    assert response.status_code == status.HTTP_201_CREATED
+        params = {
+            "protocol": "http",
+            "host": "localhost",
+            "port": 84,
+            "user": "julien",
+            "password": "secure"
+        }
+        response = await client.post(app.url_path_for("get_computes"), json=params)
+        assert response.status_code == status.HTTP_201_CREATED
+        compute_id = response.json()["compute_id"]
 
-    response = await client.get(app.url_path_for("update_compute", compute_id="my_compute_id"))
-    assert response.status_code == status.HTTP_200_OK
+        with asyncio_patch("gns3server.controller.compute.Compute.forward", return_value=[]) as mock:
+            response = await client.get(app.url_path_for("get_compute", compute_id=compute_id) + "/virtualbox/vms")
+            mock.assert_called_with("GET", "virtualbox", "vms")
+            assert response.json() == []
 
+    async def test_compute_create_img(self, app: FastAPI, client: AsyncClient) -> None:
 
-async def test_compute_update(app: FastAPI, client: AsyncClient) -> None:
+        params = {
+            "protocol": "http",
+            "host": "localhost",
+            "port": 84,
+            "user": "julien",
+            "password": "secure"
+        }
 
-    params = {
-        "compute_id": "my_compute_id",
-        "protocol": "http",
-        "host": "localhost",
-        "port": 84,
-        "user": "julien",
-        "password": "secure"
-    }
+        response = await client.post(app.url_path_for("get_computes"), json=params)
+        assert response.status_code == status.HTTP_201_CREATED
+        compute_id = response.json()["compute_id"]
 
-    response = await client.post(app.url_path_for("create_compute"), json=params)
-    assert response.status_code == status.HTTP_201_CREATED
-    response = await client.get(app.url_path_for("get_compute", compute_id="my_compute_id"))
-    assert response.status_code == status.HTTP_200_OK
-    assert response.json()["protocol"] == "http"
+        params = {"path": "/test"}
+        with asyncio_patch("gns3server.controller.compute.Compute.forward", return_value=[]) as mock:
+            response = await client.post(app.url_path_for("get_compute", compute_id=compute_id) + "/qemu/img", json=params)
+            assert response.json() == []
+            mock.assert_called_with("POST", "qemu", "img", data=unittest.mock.ANY)
 
-    params["protocol"] = "https"
-    response = await client.put(app.url_path_for("update_compute", compute_id="my_compute_id"), json=params)
-
-    assert response.status_code == status.HTTP_200_OK
-    assert response.json()["protocol"] == "https"
-
-
-async def test_compute_list(app: FastAPI, client: AsyncClient, controller: Controller) -> None:
-
-    params = {
-        "compute_id": "my_compute_id",
-        "protocol": "http",
-        "host": "localhost",
-        "port": 84,
-        "user": "julien",
-        "password": "secure",
-        "name": "My super server"
-    }
-
-    response = await client.post(app.url_path_for("create_compute"), json=params)
-    assert response.status_code == status.HTTP_201_CREATED
-    assert response.json()["user"] == "julien"
-    assert "password" not in response.json()
-
-    response = await client.get(app.url_path_for("get_computes"))
-    for compute in response.json():
-        if compute['compute_id'] != 'local':
-            assert compute == {
-                'compute_id': 'my_compute_id',
-                'connected': False,
-                'host': 'localhost',
-                'port': 84,
-                'protocol': 'http',
-                'user': 'julien',
-                'name': 'My super server',
-                'cpu_usage_percent': 0.0,
-                'memory_usage_percent': 0.0,
-                'disk_usage_percent': 0.0,
-                'last_error': None,
-                'capabilities': {
-                    'version': '',
-                    'platform': '',
-                    'cpus': 0,
-                    'memory': 0,
-                    'disk_size': 0,
-                    'node_types': []
-                }
-            }
-
-
-async def test_compute_delete(app: FastAPI, client: AsyncClient, controller: Controller) -> None:
-
-    params = {
-        "compute_id": "my_compute_id",
-        "protocol": "http",
-        "host": "localhost",
-        "port": 84,
-        "user": "julien",
-        "password": "secure"
-    }
-
-    response = await client.post(app.url_path_for("create_compute"), json=params)
-    assert response.status_code == status.HTTP_201_CREATED
-
-    response = await client.get(app.url_path_for("get_computes"))
-    assert len(response.json()) == 1
-
-    response = await client.delete(app.url_path_for("delete_compute", compute_id="my_compute_id"))
-    assert response.status_code == status.HTTP_204_NO_CONTENT
-
-    response = await client.get(app.url_path_for("get_computes"))
-    assert len(response.json()) == 0
-
-
-async def test_compute_list_images(app: FastAPI, client: AsyncClient) -> None:
-
-    params = {
-        "compute_id": "my_compute_id",
-        "protocol": "http",
-        "host": "localhost",
-        "port": 84,
-        "user": "julien",
-        "password": "secure"
-    }
-    response = await client.post(app.url_path_for("create_compute"), json=params)
-    assert response.status_code == status.HTTP_201_CREATED
-
-    with asyncio_patch("gns3server.controller.compute.Compute.images", return_value=[{"filename": "linux.qcow2"}, {"filename": "asav.qcow2"}]) as mock:
-        response = await client.get(app.url_path_for("delete_compute", compute_id="my_compute_id") + "/qemu/images")
-        assert response.json() == [{"filename": "linux.qcow2"}, {"filename": "asav.qcow2"}]
-        mock.assert_called_with("qemu")
-
-
-async def test_compute_list_vms(app: FastAPI, client: AsyncClient) -> None:
-
-    params = {
-        "compute_id": "my_compute",
-        "protocol": "http",
-        "host": "localhost",
-        "port": 84,
-        "user": "julien",
-        "password": "secure"
-    }
-    response = await client.post(app.url_path_for("get_computes"), json=params)
-    assert response.status_code == status.HTTP_201_CREATED
-
-    with asyncio_patch("gns3server.controller.compute.Compute.forward", return_value=[]) as mock:
-        response = await client.get(app.url_path_for("get_compute", compute_id="my_compute_id") + "/virtualbox/vms")
-        mock.assert_called_with("GET", "virtualbox", "vms")
-        assert response.json() == []
-
-
-async def test_compute_create_img(app: FastAPI, client: AsyncClient) -> None:
-
-    params = {
-        "compute_id": "my_compute",
-        "protocol": "http",
-        "host": "localhost",
-        "port": 84,
-        "user": "julien",
-        "password": "secure"
-    }
-
-    response = await client.post(app.url_path_for("get_computes"), json=params)
-    assert response.status_code == status.HTTP_201_CREATED
-
-    params = {"path": "/test"}
-    with asyncio_patch("gns3server.controller.compute.Compute.forward", return_value=[]) as mock:
-        response = await client.post(app.url_path_for("get_compute", compute_id="my_compute_id") + "/qemu/img", json=params)
-        assert response.json() == []
-        mock.assert_called_with("POST", "qemu", "img", data=unittest.mock.ANY)
-
-
-async def test_compute_autoidlepc(app: FastAPI, client: AsyncClient) -> None:
-
-    params = {
-        "compute_id": "my_compute_id",
-        "protocol": "http",
-        "host": "localhost",
-        "port": 84,
-        "user": "julien",
-        "password": "secure"
-    }
-
-    await client.post(app.url_path_for("get_computes"), json=params)
-
-    params = {
-        "platform": "c7200",
-        "image": "test.bin",
-        "ram": 512
-    }
-
-    with asyncio_patch("gns3server.controller.Controller.autoidlepc", return_value={"idlepc": "0x606de20c"}) as mock:
-        response = await client.post(app.url_path_for("get_compute", compute_id="my_compute_id") + "/auto_idlepc", json=params)
-    assert mock.called
-    assert response.status_code == status.HTTP_200_OK
+    # async def test_compute_autoidlepc(self, app: FastAPI, client: AsyncClient) -> None:
+    #
+    #     params = {
+    #         "protocol": "http",
+    #         "host": "localhost",
+    #         "port": 84,
+    #         "user": "julien",
+    #         "password": "secure"
+    #     }
+    #
+    #     response = await client.post(app.url_path_for("get_computes"), json=params)
+    #     assert response.status_code == status.HTTP_201_CREATED
+    #     compute_id = response.json()["compute_id"]
+    #
+    #     params = {
+    #         "platform": "c7200",
+    #         "image": "test.bin",
+    #         "ram": 512
+    #     }
+    #
+    #     with asyncio_patch("gns3server.controller.Controller.autoidlepc", return_value={"idlepc": "0x606de20c"}) as mock:
+    #         response = await client.post(app.url_path_for("autoidlepc", compute_id=compute_id) + "/auto_idlepc", json=params)
+    #         assert mock.called
+    #         assert response.status_code == status.HTTP_200_OK
 
 
 # FIXME

tests/conftest.py

@@ -4,6 +4,7 @@ import tempfile
 import shutil
 import sys
 import os
+import uuid
 
 from fastapi import FastAPI
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
@@ -16,10 +17,12 @@ from gns3server.config import Config
 from gns3server.compute import MODULES
 from gns3server.compute.port_manager import PortManager
 from gns3server.compute.project_manager import ProjectManager
-from gns3server.db.models import Base, User
+from gns3server.db.models import Base, User, Compute
 from gns3server.db.repositories.users import UsersRepository
+from gns3server.db.repositories.computes import ComputesRepository
 from gns3server.api.routes.controller.dependencies.database import get_db_session
-from gns3server.schemas.users import UserCreate
+from gns3server import schemas
+from gns3server.schemas.computes import Protocol
 from gns3server.services import auth_service
 
 sys._called_from_test = True
@@ -27,7 +30,7 @@ sys.original_platform = sys.platform
 
 
 if sys.platform.startswith("win") and sys.version_info < (3, 8):
-    @pytest.yield_fixture(scope="session")
+    @pytest.fixture(scope="session")
     def event_loop(request):
         """
         Overwrite pytest_asyncio event loop on Windows for Python < 3.8
@@ -43,7 +46,7 @@ if sys.platform.startswith("win") and sys.version_info < (3, 8):
 # https://github.com/pytest-dev/pytest-asyncio/issues/68
 # this event_loop is used by pytest-asyncio, and redefining it
 # is currently the only way of changing the scope of this fixture
-@pytest.yield_fixture(scope="class")
+@pytest.fixture(scope="class")
 def event_loop(request):
 
     loop = asyncio.get_event_loop_policy().new_event_loop()
@@ -54,9 +57,6 @@ def event_loop(request):
 @pytest.fixture(scope="class")
 async def app() -> FastAPI:
 
-    # async with db_engine.begin() as conn:
-    #     await conn.run_sync(Base.metadata.drop_all)
-    #     await conn.run_sync(Base.metadata.create_all)
     from gns3server.api.server import app as gns3app
     yield gns3app
 
@@ -109,7 +109,7 @@ async def client(app: FastAPI, db_session: AsyncSession) -> AsyncClient:
 @pytest.fixture
 async def test_user(db_session: AsyncSession) -> User:
 
-    new_user = UserCreate(
+    new_user = schemas.UserCreate(
         username="user1",
         email="user1@email.com",
         password="user1_password",
@@ -121,6 +121,25 @@ async def test_user(db_session: AsyncSession) -> User:
     return await user_repo.create_user(new_user)
 
 
+@pytest.fixture
+async def test_compute(db_session: AsyncSession) -> Compute:
+
+    new_compute = schemas.ComputeCreate(
+        compute_id=uuid.uuid4(),
+        protocol=Protocol.http,
+        host="localhost",
+        port=4242,
+        user="julien",
+        password="secure"
+    )
+
+    compute_repo = ComputesRepository(db_session)
+    existing_compute = await compute_repo.get_compute(new_compute.compute_id)
+    if existing_compute:
+        return existing_compute
+    return await compute_repo.create_compute(new_compute)
+
+
 @pytest.fixture
 def authorized_client(client: AsyncClient, test_user: User) -> AsyncClient: