diff --git a/gns3server/api/routes/compute/__init__.py b/gns3server/api/routes/compute/__init__.py index 247a2b8f..69857ec8 100644 --- a/gns3server/api/routes/compute/__init__.py +++ b/gns3server/api/routes/compute/__init__.py @@ -15,13 +15,15 @@ # along with this program. If not, see . -from fastapi import FastAPI, Request +from fastapi import FastAPI, Request, Depends from fastapi.responses import JSONResponse from starlette.exceptions import HTTPException as StarletteHTTPException from gns3server.controller.gns3vm.gns3_vm_error import GNS3VMError from gns3server.compute.error import ImageMissingError, NodeError from gns3server.compute.ubridge.ubridge_error import UbridgeError +from .dependencies.authentication import compute_authentication + from gns3server.compute.compute_error import ( ComputeError, ComputeNotFoundError, @@ -49,9 +51,9 @@ from . import virtualbox_nodes from . import vmware_nodes from . import vpcs_nodes - compute_api = FastAPI( title="GNS3 compute API", + dependencies=[Depends(compute_authentication)], description="This page describes the private compute API for GNS3. PLEASE DO NOT USE DIRECTLY!", version="v3", ) diff --git a/gns3server/api/routes/compute/dependencies/__init__.py b/gns3server/api/routes/compute/dependencies/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/gns3server/api/routes/compute/dependencies/authentication.py b/gns3server/api/routes/compute/dependencies/authentication.py new file mode 100644 index 00000000..5efb9927 --- /dev/null +++ b/gns3server/api/routes/compute/dependencies/authentication.py @@ -0,0 +1,37 @@ +# +# Copyright (C) 2021 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 . + +import secrets + +from fastapi import Depends, HTTPException, status +from fastapi.security import HTTPBasic, HTTPBasicCredentials +from gns3server.config import Config +from typing import Optional + +security = HTTPBasic() + + +def compute_authentication(credentials: Optional[HTTPBasicCredentials] = Depends(security)) -> None: + + server_settings = Config.instance().settings.Server + username = secrets.compare_digest(credentials.username, server_settings.compute_username) + password = secrets.compare_digest(credentials.password, server_settings.compute_password.get_secret_value()) + if not (username and password): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid compute username or password", + headers={"WWW-Authenticate": "Basic"}, + ) diff --git a/gns3server/api/routes/compute/notifications.py b/gns3server/api/routes/compute/notifications.py index 99d9ed31..7bc0eb5e 100644 --- a/gns3server/api/routes/compute/notifications.py +++ b/gns3server/api/routes/compute/notifications.py @@ -18,7 +18,11 @@ API routes for compute notifications. """ -from fastapi import APIRouter, WebSocket, WebSocketDisconnect +import base64 +import binascii + +from fastapi import APIRouter, WebSocket, WebSocketDisconnect, status, HTTPException +from fastapi.security.utils import get_authorization_scheme_param from websockets.exceptions import ConnectionClosed, WebSocketException from gns3server.compute.notification_manager import NotificationManager @@ -37,6 +41,32 @@ async def notification_ws(websocket: WebSocket) -> None: """ await websocket.accept() + + # handle basic HTTP authentication + invalid_user_credentials_exc = HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid authentication credentials", + headers={"WWW-Authenticate": "Basic"}, + ) + + try: + authorization = websocket.headers.get("Authorization") + scheme, param = get_authorization_scheme_param(authorization) + if not authorization or scheme.lower() != "basic": + raise invalid_user_credentials_exc + try: + data = base64.b64decode(param).decode("ascii") + except (ValueError, UnicodeDecodeError, binascii.Error): + raise invalid_user_credentials_exc + username, separator, password = data.partition(":") + if not separator: + raise invalid_user_credentials_exc + except invalid_user_credentials_exc as e: + websocket_error = {"action": "log.error", "event": {"message": f"Could not authenticate while connecting to " + f"compute WebSocket: {e.detail}"}} + await websocket.send_json(websocket_error) + return await websocket.close(code=1008) + log.info(f"New client {websocket.client.host}:{websocket.client.port} has connected to compute WebSocket") try: with NotificationManager.instance().queue() as queue: diff --git a/gns3server/controller/__init__.py b/gns3server/controller/__init__.py index 22be7f8d..9629fbfd 100644 --- a/gns3server/controller/__init__.py +++ b/gns3server/controller/__init__.py @@ -97,8 +97,8 @@ class Controller: host=host, console_host=console_host, port=port, - user=server_config.user, - password=server_config.password, + user=server_config.compute_username, + password=server_config.compute_password, force=True, connect=True, ssl_context=self._ssl_context, @@ -154,8 +154,8 @@ class Controller: """ if self._local_server: - self._local_server.user = Config.instance().settings.Server.user - self._local_server.password = Config.instance().settings.Server.password + self._local_server.user = Config.instance().settings.Server.compute_username + self._local_server.password = Config.instance().settings.Server.compute_password async def stop(self): diff --git a/gns3server/controller/compute.py b/gns3server/controller/compute.py index e3c1cca5..4add4e90 100644 --- a/gns3server/controller/compute.py +++ b/gns3server/controller/compute.py @@ -119,6 +119,7 @@ class Compute: """ Set authentication parameters """ + if user is None or len(user.strip()) == 0: self._user = None self._password = None @@ -468,6 +469,8 @@ class Compute: # FIXME: slow down number of compute events self._controller.notification.controller_emit("compute.updated", self.asdict()) else: + if action == "log.error": + log.error(event.pop("message")) await self._controller.notification.dispatch( action, event, project_id=project_id, compute_id=self.id ) diff --git a/gns3server/schemas/config.py b/gns3server/schemas/config.py index ec671823..cbcf1275 100644 --- a/gns3server/schemas/config.py +++ b/gns3server/schemas/config.py @@ -131,9 +131,8 @@ class ServerSettings(BaseModel): udp_start_port_range: int = Field(10000, gt=0, le=65535) udp_end_port_range: int = Field(30000, gt=0, le=65535) ubridge_path: str = "ubridge" - user: str = None - password: SecretStr = None - enable_http_auth: bool = False + compute_username: str = "admin" + compute_password: SecretStr = SecretStr("") default_admin_username: str = "admin" default_admin_password: SecretStr = SecretStr("admin") allowed_interfaces: List[str] = Field(default_factory=list) @@ -164,14 +163,6 @@ class ServerSettings(BaseModel): raise ValueError("vnc_console_end_port_range must be > vnc_console_start_port_range") return v - @validator("enable_http_auth") - def validate_enable_auth(cls, v, values): - - if v is True: - if "user" not in values or not values["user"]: - raise ValueError("HTTP authentication is enabled but user is not configured") - return v - @validator("enable_ssl") def validate_enable_ssl(cls, v, values): diff --git a/gns3server/server.py b/gns3server/server.py index e52f0834..56ea42d7 100644 --- a/gns3server/server.py +++ b/gns3server/server.py @@ -30,6 +30,8 @@ import asyncio import signal import functools import uvicorn +import secrets +import string from gns3server.controller import Controller from gns3server.compute.port_manager import PortManager @@ -38,7 +40,7 @@ from gns3server.version import __version__ from gns3server.config import Config from gns3server.crash_report import CrashReport from gns3server.api.server import app -from pydantic import ValidationError +from pydantic import ValidationError, SecretStr import logging @@ -269,8 +271,15 @@ class Server: if config.Server.local: log.warning("Local mode is enabled. Beware, clients will have full control on your filesystem") - if config.Server.enable_http_auth: - log.info(f"HTTP authentication is enabled with username '{config.Server.user}'") + if not config.Server.compute_password.get_secret_value(): + alphabet = string.ascii_letters + string.digits + string.punctuation + generated_password = ''.join(secrets.choice(alphabet) for _ in range(16)) + config.Server.compute_password = SecretStr(generated_password) + log.warning(f"Compute authentication is enabled with username '{config.Server.compute_username}' and " + f"a randomly generated password. Please set a password in the config file if this compute " + f"is to be used by an external controller") + else: + log.info(f"Compute authentication is enabled with username '{config.Server.compute_username}'") # we only support Python 3 version >= 3.6 if sys.version_info < (3, 6, 0): diff --git a/gns3server/services/authentication.py b/gns3server/services/authentication.py index 021db36e..5fc3796a 100644 --- a/gns3server/services/authentication.py +++ b/gns3server/services/authentication.py @@ -35,6 +35,7 @@ DEFAULT_JWT_SECRET_KEY = "efd08eccec3bd0a1be2e086670e5efa90969c68d07e072d7354a76 class AuthService: + def hash_password(self, password: str) -> str: return pwd_context.hash(password) diff --git a/tests/api/routes/compute/test_capabilities.py b/tests/api/routes/compute/test_capabilities.py index 5b5499af..2c56cfe6 100644 --- a/tests/api/routes/compute/test_capabilities.py +++ b/tests/api/routes/compute/test_capabilities.py @@ -30,9 +30,9 @@ pytestmark = pytest.mark.asyncio @pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows") -async def test_get(app: FastAPI, client: AsyncClient, windows_platform) -> None: +async def test_get(app: FastAPI, compute_client: AsyncClient, windows_platform) -> None: - response = await client.get(app.url_path_for("compute:get_capabilities")) + response = await compute_client.get(app.url_path_for("compute:get_capabilities")) assert response.status_code == status.HTTP_200_OK assert response.json() == {'node_types': ['cloud', 'ethernet_hub', 'ethernet_switch', 'nat', 'vpcs', 'virtualbox', 'dynamips', 'frame_relay_switch', 'atm_switch', 'qemu', 'vmware', 'docker', 'iou'], 'version': __version__, @@ -44,9 +44,9 @@ async def test_get(app: FastAPI, client: AsyncClient, windows_platform) -> None: @pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows") -async def test_get_on_gns3vm(app: FastAPI, client: AsyncClient, on_gns3vm) -> None: +async def test_get_on_gns3vm(app: FastAPI, compute_client: AsyncClient, on_gns3vm) -> None: - response = await client.get(app.url_path_for("compute:get_capabilities")) + response = await compute_client.get(app.url_path_for("compute:get_capabilities")) assert response.status_code == status.HTTP_200_OK assert response.json() == {'node_types': ['cloud', 'ethernet_hub', 'ethernet_switch', 'nat', 'vpcs', 'virtualbox', 'dynamips', 'frame_relay_switch', 'atm_switch', 'qemu', 'vmware', 'docker', 'iou'], 'version': __version__, diff --git a/tests/api/routes/compute/test_cloud_nodes.py b/tests/api/routes/compute/test_cloud_nodes.py index ba7995b6..6ea9f3cc 100644 --- a/tests/api/routes/compute/test_cloud_nodes.py +++ b/tests/api/routes/compute/test_cloud_nodes.py @@ -28,35 +28,35 @@ pytestmark = pytest.mark.asyncio @pytest.fixture(scope="function") -async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, on_gns3vm) -> dict: +async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project, on_gns3vm) -> dict: with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._start_ubridge"): - response = await client.post(app.url_path_for("compute:create_cloud", project_id=compute_project.id), + response = await compute_client.post(app.url_path_for("compute:create_cloud", project_id=compute_project.id), json={"name": "Cloud 1"}) assert response.status_code == status.HTTP_201_CREATED return response.json() -async def test_cloud_create(app: FastAPI, client: AsyncClient, compute_project: Project) -> None: +async def test_cloud_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None: with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._start_ubridge"): - response = await client.post(app.url_path_for("compute:create_cloud", project_id=compute_project.id), + response = await compute_client.post(app.url_path_for("compute:create_cloud", project_id=compute_project.id), json={"name": "Cloud 1"}) assert response.status_code == 201 assert response.json()["name"] == "Cloud 1" assert response.json()["project_id"] == compute_project.id -async def test_get_cloud(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None: +async def test_get_cloud(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None: - response = await client.get(app.url_path_for("compute:get_cloud", project_id=vm["project_id"], node_id=vm["node_id"])) + response = await compute_client.get(app.url_path_for("compute:get_cloud", project_id=vm["project_id"], node_id=vm["node_id"])) assert response.status_code == status.HTTP_200_OK assert response.json()["name"] == "Cloud 1" assert response.json()["project_id"] == compute_project.id assert response.json()["status"] == "started" -async def test_cloud_nio_create_udp(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None: +async def test_cloud_nio_create_udp(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None: params = {"type": "nio_udp", "lport": 4242, @@ -68,12 +68,12 @@ async def test_cloud_nio_create_udp(app: FastAPI, client: AsyncClient, compute_p node_id=vm["node_id"], adapter_number="0", port_number="0") - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_udp" -async def test_cloud_nio_update_udp(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None: +async def test_cloud_nio_update_udp(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None: params = {"type": "nio_udp", "lport": 4242, @@ -85,7 +85,7 @@ async def test_cloud_nio_update_udp(app: FastAPI, client: AsyncClient, compute_p node_id=vm["node_id"], adapter_number="0", port_number="0") - await client.post(url, json=params) + await compute_client.post(url, json=params) params["filters"] = {} url = app.url_path_for("compute:create_cloud_nio", @@ -93,12 +93,12 @@ async def test_cloud_nio_update_udp(app: FastAPI, client: AsyncClient, compute_p node_id=vm["node_id"], adapter_number="0", port_number="0") - response = await client.put(url, json=params) + response = await compute_client.put(url, json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_udp" -async def test_cloud_delete_nio(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None: +async def test_cloud_delete_nio(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None: params = {"type": "nio_udp", "lport": 4242, @@ -110,7 +110,7 @@ async def test_cloud_delete_nio(app: FastAPI, client: AsyncClient, compute_proje node_id=vm["node_id"], adapter_number="0", port_number="0") - await client.post(url, json=params) + await compute_client.post(url, json=params) url = app.url_path_for("compute:delete_cloud_nio", project_id=vm["project_id"], @@ -118,25 +118,25 @@ async def test_cloud_delete_nio(app: FastAPI, client: AsyncClient, compute_proje adapter_number="0", port_number="0") with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._start_ubridge"): - response = await client.delete(url) + response = await compute_client.delete(url) assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_cloud_delete(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None: +async def test_cloud_delete(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None: - response = await client.delete(app.url_path_for("compute:delete_cloud", project_id=vm["project_id"], node_id=vm["node_id"])) + response = await compute_client.delete(app.url_path_for("compute:delete_cloud", project_id=vm["project_id"], node_id=vm["node_id"])) assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_cloud_update(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_cloud_update(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: - response = await client.put(app.url_path_for("compute:update_cloud", project_id=vm["project_id"], node_id=vm["node_id"]), + response = await compute_client.put(app.url_path_for("compute:update_cloud", project_id=vm["project_id"], node_id=vm["node_id"]), json={"name": "test"}) assert response.status_code == status.HTTP_200_OK assert response.json()["name"] == "test" -async def test_cloud_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_cloud_start_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "capture_file_name": "test.pcap", @@ -144,7 +144,7 @@ async def test_cloud_start_capture(app: FastAPI, client: AsyncClient, vm: dict) } with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud.start_capture") as mock: - response = await client.post(app.url_path_for("compute:start_cloud_capture", + response = await compute_client.post(app.url_path_for("compute:start_cloud_capture", project_id=vm["project_id"], node_id=vm["node_id"], adapter_number="0", @@ -155,10 +155,10 @@ async def test_cloud_start_capture(app: FastAPI, client: AsyncClient, vm: dict) assert "test.pcap" in response.json()["pcap_file_path"] -async def test_cloud_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_cloud_stop_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud.stop_capture") as mock: - response = await client.post(app.url_path_for("compute:stop_cloud_capture", + response = await compute_client.post(app.url_path_for("compute:stop_cloud_capture", project_id=vm["project_id"], node_id=vm["node_id"], adapter_number="0", diff --git a/tests/api/routes/compute/test_compute.py b/tests/api/routes/compute/test_compute.py index 70fa27e1..b940445e 100644 --- a/tests/api/routes/compute/test_compute.py +++ b/tests/api/routes/compute/test_compute.py @@ -27,27 +27,33 @@ from gns3server.compute.project import Project pytestmark = pytest.mark.asyncio -async def test_udp_allocation(app: FastAPI, client: AsyncClient, compute_project: Project) -> None: +async def test_udp_allocation(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None: - response = await client.post(app.url_path_for("compute:allocate_udp_port", project_id=compute_project.id), json={}) + response = await compute_client.post(app.url_path_for("compute:allocate_udp_port", project_id=compute_project.id), json={}) assert response.status_code == status.HTTP_201_CREATED assert response.json()['udp_port'] is not None -async def test_interfaces(app: FastAPI, client: AsyncClient) -> None: +async def test_interfaces(app: FastAPI, compute_client: AsyncClient) -> None: - response = await client.get(app.url_path_for("compute:network_interfaces")) + response = await compute_client.get(app.url_path_for("compute:network_interfaces")) assert response.status_code == status.HTTP_200_OK assert isinstance(response.json(), list) -async def test_version_output(app: FastAPI, client: AsyncClient) -> None: +async def test_version_output(app: FastAPI, compute_client: AsyncClient) -> None: - response = await client.get(app.url_path_for("compute:compute_version")) + response = await compute_client.get(app.url_path_for("compute:compute_version")) assert response.status_code == status.HTTP_200_OK assert response.json() == {'local': True, 'version': __version__} +async def test_compute_authentication(app: FastAPI, compute_client: AsyncClient) -> None: + + response = await compute_client.get(app.url_path_for("compute:compute_version"), auth=("admin", "invalid_password")) + assert response.status_code == status.HTTP_401_UNAUTHORIZED + + # @pytest.mark.asyncio # async def test_debug_output(compute_api): # @@ -55,7 +61,7 @@ async def test_version_output(app: FastAPI, client: AsyncClient) -> None: # assert response.status_code == 200 -async def test_statistics_output(app: FastAPI, client: AsyncClient) -> None: +async def test_statistics_output(app: FastAPI, compute_client: AsyncClient) -> None: - response = await client.get(app.url_path_for("compute:compute_statistics")) + response = await compute_client.get(app.url_path_for("compute:compute_statistics")) assert response.status_code == status.HTTP_200_OK diff --git a/tests/api/routes/compute/test_docker_nodes.py b/tests/api/routes/compute/test_docker_nodes.py index 2d64c097..01aa7e33 100644 --- a/tests/api/routes/compute/test_docker_nodes.py +++ b/tests/api/routes/compute/test_docker_nodes.py @@ -57,22 +57,22 @@ def base_params() -> dict: @pytest.fixture -async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> dict: +async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project, base_params: dict) -> dict: with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "nginx"}]): with asyncio_patch("gns3server.compute.docker.Docker.query", return_value={"Id": "8bd8153ea8f5"}): with asyncio_patch("gns3server.compute.docker.DockerVM._get_container_state", return_value="exited"): - response = await client.post(app.url_path_for("compute:create_docker_node", project_id=compute_project.id), + response = await compute_client.post(app.url_path_for("compute:create_docker_node", project_id=compute_project.id), json=base_params) assert response.status_code == status.HTTP_201_CREATED return response.json() -async def test_docker_create(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> None: +async def test_docker_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project, base_params: dict) -> None: with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "nginx"}]): with asyncio_patch("gns3server.compute.docker.Docker.query", return_value={"Id": "8bd8153ea8f5"}): - response = await client.post(app.url_path_for("compute:create_docker_node", project_id=compute_project.id), + response = await compute_client.post(app.url_path_for("compute:create_docker_node", project_id=compute_project.id), json=base_params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["name"] == "PC TEST 1" @@ -85,68 +85,68 @@ async def test_docker_create(app: FastAPI, client: AsyncClient, compute_project: assert response.json()["extra_hosts"] == "test:127.0.0.1" -async def test_docker_start(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_docker_start(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.start", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:start_docker_node", + response = await compute_client.post(app.url_path_for("compute:start_docker_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_docker_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_docker_stop(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.stop", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:stop_docker_node", + response = await compute_client.post(app.url_path_for("compute:stop_docker_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_docker_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_docker_reload(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.restart", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:reload_docker_node", + response = await compute_client.post(app.url_path_for("compute:reload_docker_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_docker_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_docker_delete(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.delete", return_value=True) as mock: - response = await client.delete(app.url_path_for("compute:delete_docker_node", + response = await compute_client.delete(app.url_path_for("compute:delete_docker_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_docker_pause(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_docker_pause(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.pause", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:pause_docker_node", + response = await compute_client.post(app.url_path_for("compute:pause_docker_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_docker_unpause(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_docker_unpause(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.unpause", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:unpause_docker_node", + response = await compute_client.post(app.url_path_for("compute:unpause_docker_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_docker_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_docker_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "type": "nio_udp", @@ -159,12 +159,12 @@ async def test_docker_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict node_id=vm["node_id"], adapter_number="0", port_number="0") - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_udp" -async def test_docker_update_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_docker_update_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "type": "nio_udp", @@ -178,7 +178,7 @@ async def test_docker_update_nio(app: FastAPI, client: AsyncClient, vm: dict) -> node_id=vm["node_id"], adapter_number="0", port_number="0") - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_201_CREATED url = app.url_path_for("compute:update_docker_node_nio", @@ -187,11 +187,11 @@ async def test_docker_update_nio(app: FastAPI, client: AsyncClient, vm: dict) -> adapter_number="0", port_number="0") with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.adapter_update_nio_binding"): - response = await client.put(url, json=params) + response = await compute_client.put(url, json=params) assert response.status_code == status.HTTP_201_CREATED -async def test_docker_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_docker_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: url = app.url_path_for("compute:delete_docker_node_nio", project_id=vm["project_id"], @@ -199,11 +199,11 @@ async def test_docker_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> adapter_number="0", port_number="0") with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.adapter_remove_nio_binding"): - response = await client.delete(url) + response = await compute_client.delete(url) assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_docker_update(app: FastAPI, client: AsyncClient, vm: dict, free_console_port: int) -> None: +async def test_docker_update(app: FastAPI, compute_client: AsyncClient, vm: dict, free_console_port: int) -> None: params = { "name": "test", @@ -214,7 +214,7 @@ async def test_docker_update(app: FastAPI, client: AsyncClient, vm: dict, free_c } with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.update") as mock: - response = await client.put(app.url_path_for("compute:update_docker_node", + response = await compute_client.put(app.url_path_for("compute:update_docker_node", project_id=vm["project_id"], node_id=vm["node_id"]), json=params) @@ -227,7 +227,7 @@ async def test_docker_update(app: FastAPI, client: AsyncClient, vm: dict, free_c assert response.json()["extra_hosts"] == "test:127.0.0.1" -async def test_docker_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_docker_start_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: url = app.url_path_for("compute:start_docker_node_capture", project_id=vm["project_id"], @@ -238,13 +238,13 @@ async def test_docker_start_capture(app: FastAPI, client: AsyncClient, vm: dict) with patch("gns3server.compute.docker.docker_vm.DockerVM.is_running", return_value=True): with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.start_capture") as mock: params = {"capture_file_name": "test.pcap", "data_link_type": "DLT_EN10MB"} - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_200_OK assert mock.called assert "test.pcap" in response.json()["pcap_file_path"] -async def test_docker_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_docker_stop_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: url = app.url_path_for("compute:stop_docker_node_capture", project_id=vm["project_id"], @@ -254,22 +254,22 @@ async def test_docker_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) with patch("gns3server.compute.docker.docker_vm.DockerVM.is_running", return_value=True): with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.stop_capture") as mock: - response = await client.post(url) + response = await compute_client.post(url) assert response.status_code == status.HTTP_204_NO_CONTENT assert mock.called -async def test_docker_duplicate(app: FastAPI, client: AsyncClient, vm: dict, base_params: dict) -> None: +async def test_docker_duplicate(app: FastAPI, compute_client: AsyncClient, vm: dict, base_params: dict) -> None: # create destination node first with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "nginx"}]): with asyncio_patch("gns3server.compute.docker.Docker.query", return_value={"Id": "8bd8153ea8f5"}): - response = await client.post(app.url_path_for("compute:create_docker_node", + response = await compute_client.post(app.url_path_for("compute:create_docker_node", project_id=vm["project_id"]), json=base_params) assert response.status_code == status.HTTP_201_CREATED params = {"destination_node_id": response.json()["node_id"]} - response = await client.post(app.url_path_for("compute:duplicate_docker_node", + response = await compute_client.post(app.url_path_for("compute:duplicate_docker_node", project_id=vm["project_id"], node_id=vm["node_id"]), json=params) assert response.status_code == status.HTTP_201_CREATED diff --git a/tests/api/routes/compute/test_dynamips_nodes.py b/tests/api/routes/compute/test_dynamips_nodes.py index 7c9463ff..904f0dda 100644 --- a/tests/api/routes/compute/test_dynamips_nodes.py +++ b/tests/api/routes/compute/test_dynamips_nodes.py @@ -162,10 +162,10 @@ def fake_file(tmpdir) -> str: return path -async def test_images(app: FastAPI, client: AsyncClient, tmpdir, fake_image: str, fake_file: str) -> None: +async def test_images(app: FastAPI, compute_client: AsyncClient, tmpdir, fake_image: str, fake_file: str) -> None: with patch("gns3server.utils.images.default_images_directory", return_value=str(tmpdir)): - response = await client.get(app.url_path_for("compute:get_dynamips_images")) + response = await compute_client.get(app.url_path_for("compute:get_dynamips_images")) assert response.status_code == status.HTTP_200_OK assert response.json() == [{"filename": "7200.bin", "path": "7200.bin", @@ -173,9 +173,9 @@ async def test_images(app: FastAPI, client: AsyncClient, tmpdir, fake_image: str "md5sum": "b0d5aa897d937aced5a6b1046e8f7e2e"}] -async def test_upload_image(app: FastAPI, client: AsyncClient, images_dir: str) -> None: +async def test_upload_image(app: FastAPI, compute_client: AsyncClient, images_dir: str) -> None: - response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename="test2"), content=b"TEST") + response = await compute_client.post(app.url_path_for("compute:upload_dynamips_image", filename="test2"), content=b"TEST") assert response.status_code == status.HTTP_204_NO_CONTENT with open(os.path.join(images_dir, "IOS", "test2")) as f: @@ -186,36 +186,36 @@ async def test_upload_image(app: FastAPI, client: AsyncClient, images_dir: str) assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf" -async def test_upload_image_forbidden_location(app: FastAPI, client: AsyncClient) -> None: +async def test_upload_image_forbidden_location(app: FastAPI, compute_client: AsyncClient) -> None: file_path = "%2e%2e/hello" - response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename=file_path), content=b"TEST") + response = await compute_client.post(app.url_path_for("compute:upload_dynamips_image", filename=file_path), content=b"TEST") assert response.status_code == status.HTTP_403_FORBIDDEN -async def test_download_image(app: FastAPI, client: AsyncClient, images_dir: str) -> None: +async def test_download_image(app: FastAPI, compute_client: AsyncClient, images_dir: str) -> None: - response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename="test3"), content=b"TEST") + response = await compute_client.post(app.url_path_for("compute:upload_dynamips_image", filename="test3"), content=b"TEST") assert response.status_code == status.HTTP_204_NO_CONTENT - response = await client.get(app.url_path_for("compute:download_dynamips_image", filename="test3")) + response = await compute_client.get(app.url_path_for("compute:download_dynamips_image", filename="test3")) assert response.status_code == status.HTTP_200_OK -async def test_download_image_forbidden(app: FastAPI, client: AsyncClient, tmpdir) -> None: +async def test_download_image_forbidden(app: FastAPI, compute_client: AsyncClient, tmpdir) -> None: file_path = "foo/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/etc/passwd" - response = await client.get(app.url_path_for("compute:download_dynamips_image", filename=file_path)) + response = await compute_client.get(app.url_path_for("compute:download_dynamips_image", filename=file_path)) assert response.status_code == status.HTTP_403_FORBIDDEN @pytest.mark.skipif(not sys.platform.startswith("win") and os.getuid() == 0, reason="Root can delete any image") -async def test_upload_image_permission_denied(app: FastAPI, client: AsyncClient, images_dir: str) -> None: +async def test_upload_image_permission_denied(app: FastAPI, compute_client: AsyncClient, images_dir: str) -> None: os.makedirs(os.path.join(images_dir, "IOS"), exist_ok=True) with open(os.path.join(images_dir, "IOS", "test2.tmp"), "w+") as f: f.write("") os.chmod(os.path.join(images_dir, "IOS", "test2.tmp"), 0) - response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename="test2"), content=b"TEST") + response = await compute_client.post(app.url_path_for("compute:upload_dynamips_image", filename="test2"), content=b"TEST") assert response.status_code == status.HTTP_409_CONFLICT diff --git a/tests/api/routes/compute/test_iou_nodes.py b/tests/api/routes/compute/test_iou_nodes.py index 1a2458a2..135578ce 100644 --- a/tests/api/routes/compute/test_iou_nodes.py +++ b/tests/api/routes/compute/test_iou_nodes.py @@ -51,9 +51,9 @@ def base_params(tmpdir, fake_iou_bin) -> dict: @pytest.fixture -async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> dict: +async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project, base_params: dict) -> dict: - response = await client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=base_params) + response = await compute_client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=base_params) assert response.status_code == status.HTTP_201_CREATED return response.json() @@ -65,9 +65,9 @@ def startup_config_file(compute_project: Project, vm: dict) -> str: return os.path.join(directory, "startup-config.cfg") -async def test_iou_create(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> None: +async def test_iou_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project, base_params: dict) -> None: - response = await client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=base_params) + response = await compute_client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=base_params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["name"] == "PC TEST 1" assert response.json()["project_id"] == compute_project.id @@ -79,7 +79,7 @@ async def test_iou_create(app: FastAPI, client: AsyncClient, compute_project: Pr async def test_iou_create_with_params(app: FastAPI, - client: AsyncClient, + compute_client: AsyncClient, compute_project: Project, base_params: dict) -> None: @@ -92,7 +92,7 @@ async def test_iou_create_with_params(app: FastAPI, params["startup_config_content"] = "hostname test" params["use_default_iou_values"] = False - response = await client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=params) + response = await compute_client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["name"] == "PC TEST 1" assert response.json()["project_id"] == compute_project.id @@ -109,7 +109,7 @@ async def test_iou_create_with_params(app: FastAPI, async def test_iou_create_startup_config_already_exist( app: FastAPI, - client: AsyncClient, + compute_client: AsyncClient, compute_project: Project, base_params: dict) -> None: """We don't erase a startup-config if already exist at project creation""" @@ -123,16 +123,16 @@ async def test_iou_create_startup_config_already_exist( params["node_id"] = node_id params["startup_config_content"] = "hostname test" - response = await client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=params) + response = await compute_client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=params) assert response.status_code == status.HTTP_201_CREATED with open(startup_config_file(compute_project, response.json())) as f: assert f.read() == "echo hello" -async def test_iou_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None: +async def test_iou_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None: - response = await client.get(app.url_path_for("compute:get_iou_node", project_id=vm["project_id"], node_id=vm["node_id"])) + response = await compute_client.get(app.url_path_for("compute:get_iou_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert response.status_code == status.HTTP_200_OK assert response.json()["name"] == "PC TEST 1" assert response.json()["project_id"] == compute_project.id @@ -143,58 +143,58 @@ async def test_iou_get(app: FastAPI, client: AsyncClient, compute_project: Proje assert response.json()["l1_keepalives"] is False -async def test_iou_start(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_iou_start(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.start", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:start_iou_node", + response = await compute_client.post(app.url_path_for("compute:start_iou_node", project_id=vm["project_id"], node_id=vm["node_id"]), json={}) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_iou_start_with_iourc(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_iou_start_with_iourc(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = {"iourc_content": "test"} with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.start", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:start_iou_node", + response = await compute_client.post(app.url_path_for("compute:start_iou_node", project_id=vm["project_id"], node_id=vm["node_id"]), json=params) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_iou_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_iou_stop(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.stop", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:stop_iou_node", + response = await compute_client.post(app.url_path_for("compute:stop_iou_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_iou_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_iou_reload(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.reload", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:reload_iou_node", + response = await compute_client.post(app.url_path_for("compute:reload_iou_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_iou_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_iou_delete(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.iou.IOU.delete_node", return_value=True) as mock: - response = await client.delete(app.url_path_for("compute:delete_iou_node", + response = await compute_client.delete(app.url_path_for("compute:delete_iou_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_iou_update(app: FastAPI, client: AsyncClient, vm: dict, free_console_port: int) -> None: +async def test_iou_update(app: FastAPI, compute_client: AsyncClient, vm: dict, free_console_port: int) -> None: params = { "name": "test", @@ -207,7 +207,7 @@ async def test_iou_update(app: FastAPI, client: AsyncClient, vm: dict, free_cons "use_default_iou_values": True, } - response = await client.put(app.url_path_for("compute:update_iou_node", + response = await compute_client.put(app.url_path_for("compute:update_iou_node", project_id=vm["project_id"], node_id=vm["node_id"]), json=params) assert response.status_code == status.HTTP_200_OK @@ -221,7 +221,7 @@ async def test_iou_update(app: FastAPI, client: AsyncClient, vm: dict, free_cons assert response.json()["use_default_iou_values"] is True -async def test_iou_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_iou_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = {"type": "nio_udp", "lport": 4242, @@ -233,12 +233,12 @@ async def test_iou_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) - node_id=vm["node_id"], adapter_number="1", port_number="0") - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_udp" -async def test_iou_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_iou_nio_update_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = {"type": "nio_udp", "lport": 4242, @@ -251,7 +251,7 @@ async def test_iou_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) - adapter_number="1", port_number="0") - await client.post(url, json=params) + await compute_client.post(url, json=params) params["filters"] = {} url = app.url_path_for("compute:update_iou_node_nio", @@ -259,12 +259,12 @@ async def test_iou_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) - node_id=vm["node_id"], adapter_number="1", port_number="0") - response = await client.put(url, json=params) + response = await compute_client.put(url, json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_udp" -async def test_iou_nio_create_ethernet(app: FastAPI, client: AsyncClient, vm: dict, ethernet_device: str) -> None: +async def test_iou_nio_create_ethernet(app: FastAPI, compute_client: AsyncClient, vm: dict, ethernet_device: str) -> None: params = { "type": "nio_ethernet", @@ -277,14 +277,14 @@ async def test_iou_nio_create_ethernet(app: FastAPI, client: AsyncClient, vm: di adapter_number="1", port_number="0") - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_ethernet" assert response.json()["ethernet_device"] == ethernet_device async def test_iou_nio_create_ethernet_different_port(app: FastAPI, - client: AsyncClient, + compute_client: AsyncClient, vm: dict, ethernet_device: str) -> None: @@ -298,13 +298,13 @@ async def test_iou_nio_create_ethernet_different_port(app: FastAPI, node_id=vm["node_id"], adapter_number="0", port_number="3") - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_ethernet" assert response.json()["ethernet_device"] == ethernet_device -async def test_iou_nio_create_tap(app: FastAPI, client: AsyncClient, vm: dict, ethernet_device: str) -> None: +async def test_iou_nio_create_tap(app: FastAPI, compute_client: AsyncClient, vm: dict, ethernet_device: str) -> None: params = { "type": "nio_tap", @@ -317,12 +317,12 @@ async def test_iou_nio_create_tap(app: FastAPI, client: AsyncClient, vm: dict, e adapter_number="1", port_number="0") with patch("gns3server.compute.base_manager.BaseManager.has_privileged_access", return_value=True): - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_tap" -async def test_iou_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_iou_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "type": "nio_udp", @@ -337,7 +337,7 @@ async def test_iou_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> No adapter_number="1", port_number="0") - await client.post(url, json=params) + await compute_client.post(url, json=params) url = app.url_path_for("compute:delete_iou_node_nio", project_id=vm["project_id"], @@ -345,11 +345,11 @@ async def test_iou_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> No adapter_number="1", port_number="0") - response = await client.delete(url) + response = await compute_client.delete(url) assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_iou_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_iou_start_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "capture_file_name": "test.pcap", @@ -364,13 +364,13 @@ async def test_iou_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> with patch("gns3server.compute.iou.iou_vm.IOUVM.is_running", return_value=True): with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.start_capture") as mock: - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_200_OK assert mock.called assert "test.pcap" in response.json()["pcap_file_path"] -async def test_iou_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_iou_stop_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: url = app.url_path_for("compute:stop_iou_node_capture", project_id=vm["project_id"], @@ -380,7 +380,7 @@ async def test_iou_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> with patch("gns3server.compute.iou.iou_vm.IOUVM.is_running", return_value=True): with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.stop_capture") as mock: - response = await client.post(url) + response = await compute_client.post(url) assert response.status_code == status.HTTP_204_NO_CONTENT assert mock.called @@ -390,21 +390,21 @@ async def test_iou_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> # # with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.get_nio"): # with asyncio_patch("gns3server.compute.iou.IOU.stream_pcap_file"): -# response = await client.get("/projects/{project_id}/iou/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True) +# response = await compute_client.get("/projects/{project_id}/iou/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True) # assert response.status_code == status.HTTP_200_OK -async def test_images(app: FastAPI, client: AsyncClient, fake_iou_bin: str) -> None: +async def test_images(app: FastAPI, compute_client: AsyncClient, fake_iou_bin: str) -> None: - response = await client.get(app.url_path_for("compute:get_iou_images")) + response = await compute_client.get(app.url_path_for("compute:get_iou_images")) assert response.status_code == status.HTTP_200_OK assert response.json() == [{"filename": "iou.bin", "path": "iou.bin", "filesize": 7, "md5sum": "e573e8f5c93c6c00783f20c7a170aa6c"}] -async def test_upload_image(app: FastAPI, client: AsyncClient, tmpdir) -> None: +async def test_upload_image(app: FastAPI, compute_client: AsyncClient, tmpdir) -> None: with patch("gns3server.compute.IOU.get_images_directory", return_value=str(tmpdir)): - response = await client.post(app.url_path_for("compute:upload_iou_image", filename="test2"), content=b"TEST") + response = await compute_client.post(app.url_path_for("compute:upload_iou_image", filename="test2"), content=b"TEST") assert response.status_code == status.HTTP_204_NO_CONTENT with open(str(tmpdir / "test2")) as f: @@ -415,38 +415,38 @@ async def test_upload_image(app: FastAPI, client: AsyncClient, tmpdir) -> None: assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf" -async def test_upload_image_forbidden_location(app: FastAPI, client: AsyncClient) -> None: +async def test_upload_image_forbidden_location(app: FastAPI, compute_client: AsyncClient) -> None: file_path = "%2e%2e/hello" - response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename=file_path), content=b"TEST") + response = await compute_client.post(app.url_path_for("compute:upload_dynamips_image", filename=file_path), content=b"TEST") assert response.status_code == status.HTTP_403_FORBIDDEN -async def test_download_image(app: FastAPI, client: AsyncClient, images_dir: str) -> None: +async def test_download_image(app: FastAPI, compute_client: AsyncClient, images_dir: str) -> None: - response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename="test3"), content=b"TEST") + response = await compute_client.post(app.url_path_for("compute:upload_dynamips_image", filename="test3"), content=b"TEST") assert response.status_code == status.HTTP_204_NO_CONTENT - response = await client.get(app.url_path_for("compute:download_dynamips_image", filename="test3")) + response = await compute_client.get(app.url_path_for("compute:download_dynamips_image", filename="test3")) assert response.status_code == status.HTTP_200_OK -async def test_download_image_forbidden(app: FastAPI, client: AsyncClient, tmpdir) -> None: +async def test_download_image_forbidden(app: FastAPI, compute_client: AsyncClient, tmpdir) -> None: file_path = "foo/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/etc/passwd" - response = await client.get(app.url_path_for("compute:download_iou_image", filename=file_path)) + response = await compute_client.get(app.url_path_for("compute:download_iou_image", filename=file_path)) assert response.status_code == status.HTTP_403_FORBIDDEN -async def test_iou_duplicate(app: FastAPI, client: AsyncClient, vm: dict, base_params: dict) -> None: +async def test_iou_duplicate(app: FastAPI, compute_client: AsyncClient, vm: dict, base_params: dict) -> None: # create destination node first - response = await client.post(app.url_path_for("compute:create_iou_node", project_id=vm["project_id"]), json=base_params) + response = await compute_client.post(app.url_path_for("compute:create_iou_node", project_id=vm["project_id"]), json=base_params) assert response.status_code == status.HTTP_201_CREATED params = {"destination_node_id": response.json()["node_id"]} - response = await client.post(app.url_path_for("compute:duplicate_iou_node", + response = await compute_client.post(app.url_path_for("compute:duplicate_iou_node", project_id=vm["project_id"], node_id=vm["node_id"]), json=params) assert response.status_code == status.HTTP_201_CREATED diff --git a/tests/api/routes/compute/test_nat_nodes.py b/tests/api/routes/compute/test_nat_nodes.py index 3bab3ffe..67913785 100644 --- a/tests/api/routes/compute/test_nat_nodes.py +++ b/tests/api/routes/compute/test_nat_nodes.py @@ -27,35 +27,35 @@ pytestmark = pytest.mark.asyncio @pytest.fixture(scope="function") -async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, ubridge_path: str, on_gns3vm) -> dict: +async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project, ubridge_path: str, on_gns3vm) -> dict: with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat._start_ubridge"): - response = await client.post(app.url_path_for("compute:create_nat_node", project_id=compute_project.id), + response = await compute_client.post(app.url_path_for("compute:create_nat_node", project_id=compute_project.id), json={"name": "Nat 1"}) assert response.status_code == status.HTTP_201_CREATED return response.json() -async def test_nat_create(app: FastAPI, client: AsyncClient, compute_project: Project, on_gns3vm) -> None: +async def test_nat_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project, on_gns3vm) -> None: with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat._start_ubridge"): - response = await client.post(app.url_path_for("compute:create_nat_node", project_id=compute_project.id), + response = await compute_client.post(app.url_path_for("compute:create_nat_node", project_id=compute_project.id), json={"name": "Nat 1"}) assert response.status_code == status.HTTP_201_CREATED assert response.json()["name"] == "Nat 1" assert response.json()["project_id"] == compute_project.id -async def test_nat_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None: +async def test_nat_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None: - response = await client.get(app.url_path_for("compute:get_nat_node", project_id=vm["project_id"], node_id=vm["node_id"])) + response = await compute_client.get(app.url_path_for("compute:get_nat_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert response.status_code == status.HTTP_200_OK assert response.json()["name"] == "Nat 1" assert response.json()["project_id"] == compute_project.id assert response.json()["status"] == "started" -async def test_nat_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_nat_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "type": "nio_udp", @@ -71,12 +71,12 @@ async def test_nat_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) - port_number="0") with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.add_nio"): - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_udp" -async def test_nat_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_nat_nio_update_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "type": "nio_udp", @@ -91,7 +91,7 @@ async def test_nat_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) - adapter_number="0", port_number="0") - await client.post(url, json=params) + await compute_client.post(url, json=params) params["filters"] = {} url = app.url_path_for("compute:update_nat_node_nio", @@ -99,12 +99,12 @@ async def test_nat_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) - node_id=vm["node_id"], adapter_number="0", port_number="0") - response = await client.put(url, json=params) + response = await compute_client.put(url, json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_udp" -async def test_nat_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_nat_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "type": "nio_udp", @@ -119,7 +119,7 @@ async def test_nat_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> No adapter_number="0", port_number="0") with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.add_nio"): - await client.post(url, json=params) + await compute_client.post(url, json=params) url = app.url_path_for("compute:delete_nat_node_nio", project_id=vm["project_id"], @@ -127,30 +127,30 @@ async def test_nat_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> No adapter_number="0", port_number="0") with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.remove_nio") as mock: - response = await client.delete(url) + response = await compute_client.delete(url) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_nat_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_nat_delete(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: - response = await client.delete(app.url_path_for("compute:delete_nat_node", + response = await compute_client.delete(app.url_path_for("compute:delete_nat_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_nat_update(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_nat_update(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: - response = await client.put(app.url_path_for("compute:update_nat_node", + response = await compute_client.put(app.url_path_for("compute:update_nat_node", project_id=vm["project_id"], node_id=vm["node_id"]), json={"name": "test"}) assert response.status_code == status.HTTP_200_OK assert response.json()["name"] == "test" -async def test_nat_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_nat_start_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "capture_file_name": "test.pcap", @@ -163,13 +163,13 @@ async def test_nat_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> adapter_number="0", port_number="0") with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.start_capture") as mock: - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_200_OK assert mock.called assert "test.pcap" in response.json()["pcap_file_path"] -async def test_nat_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_nat_stop_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: url = app.url_path_for("compute:stop_nat_node_capture", project_id=vm["project_id"], @@ -178,7 +178,7 @@ async def test_nat_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> port_number="0") with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.stop_capture") as mock: - response = await client.post(url) + response = await compute_client.post(url) assert response.status_code == status.HTTP_204_NO_CONTENT assert mock.called @@ -188,5 +188,5 @@ async def test_nat_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> # # with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.get_nio"): # with asyncio_patch("gns3server.compute.builtin.Builtin.stream_pcap_file"): -# response = await client.get("/projects/{project_id}/nat/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True) +# response = await compute_client.get("/projects/{project_id}/nat/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True) # assert response.status_code == status.HTTP_200_OK diff --git a/tests/api/routes/compute/test_projects.py b/tests/api/routes/compute/test_projects.py index 45b24f3a..cb95896a 100644 --- a/tests/api/routes/compute/test_projects.py +++ b/tests/api/routes/compute/test_projects.py @@ -42,40 +42,40 @@ def base_params(tmpdir) -> dict: return params -async def test_create_project_with_path(app: FastAPI, client: AsyncClient, base_params: dict) -> None: +async def test_create_project_with_path(app: FastAPI, compute_client: AsyncClient, base_params: dict) -> None: with patch("gns3server.compute.project.Project.is_local", return_value=True): - response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params) + response = await compute_client.post(app.url_path_for("compute:create_compute_project"), json=base_params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["project_id"] == base_params["project_id"] async def test_create_project_with_path_and_empty_variables(app: FastAPI, - client: AsyncClient, + compute_client: AsyncClient, base_params: dict) -> None: base_params["variables"] = None with patch("gns3server.compute.project.Project.is_local", return_value=True): - response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params) + response = await compute_client.post(app.url_path_for("compute:create_compute_project"), json=base_params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["project_id"] == base_params["project_id"] -async def test_create_project_without_dir(app: FastAPI, client: AsyncClient, base_params: dict) -> None: +async def test_create_project_without_dir(app: FastAPI, compute_client: AsyncClient, base_params: dict) -> None: del base_params["path"] - response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params) + response = await compute_client.post(app.url_path_for("compute:create_compute_project"), json=base_params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["project_id"] == base_params["project_id"] assert response.json()["name"] == base_params["name"] -async def test_show_project(app: FastAPI, client: AsyncClient, base_params: dict) -> None: +async def test_show_project(app: FastAPI, compute_client: AsyncClient, base_params: dict) -> None: - response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params) + response = await compute_client.post(app.url_path_for("compute:create_compute_project"), json=base_params) assert response.status_code == status.HTTP_201_CREATED - response = await client.get(app.url_path_for("compute:get_compute_project", project_id=base_params["project_id"])) + response = await compute_client.get(app.url_path_for("compute:get_compute_project", project_id=base_params["project_id"])) #print(response.json().keys()) #assert len(response.json().keys()) == 3 @@ -84,60 +84,60 @@ async def test_show_project(app: FastAPI, client: AsyncClient, base_params: dict assert response.json()["variables"] is None -async def test_show_project_invalid_uuid(app: FastAPI, client: AsyncClient) -> None: +async def test_show_project_invalid_uuid(app: FastAPI, compute_client: AsyncClient) -> None: - response = await client.get(app.url_path_for("compute:get_compute_project", + response = await compute_client.get(app.url_path_for("compute:get_compute_project", project_id="50010203-0405-0607-0809-0a0b0c0d0e42")) assert response.status_code == status.HTTP_404_NOT_FOUND -async def test_list_projects(app: FastAPI, client: AsyncClient) -> dict: +async def test_list_projects(app: FastAPI, compute_client: AsyncClient) -> dict: ProjectManager.instance()._projects = {} params = {"name": "test", "project_id": "51010203-0405-0607-0809-0a0b0c0d0e0f"} - response = await client.post(app.url_path_for("compute:create_compute_project"), json=params) + response = await compute_client.post(app.url_path_for("compute:create_compute_project"), json=params) assert response.status_code == status.HTTP_201_CREATED params = {"name": "test", "project_id": "52010203-0405-0607-0809-0a0b0c0d0e0b"} - response = await client.post(app.url_path_for("compute:create_compute_project"), json=params) + response = await compute_client.post(app.url_path_for("compute:create_compute_project"), json=params) assert response.status_code == status.HTTP_201_CREATED - response = await client.get(app.url_path_for("compute:get_compute_projects")) + response = await compute_client.get(app.url_path_for("compute:get_compute_projects")) assert response.status_code == status.HTTP_200_OK assert len(response.json()) == 2 assert "51010203-0405-0607-0809-0a0b0c0d0e0f" in [p["project_id"] for p in response.json()] -async def test_delete_project(app: FastAPI, client: AsyncClient, compute_project: Project) -> None: +async def test_delete_project(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None: with asyncio_patch("gns3server.compute.project.Project.delete", return_value=True) as mock: - response = await client.delete(app.url_path_for("compute:delete_compute_project", project_id=compute_project.id)) + response = await compute_client.delete(app.url_path_for("compute:delete_compute_project", project_id=compute_project.id)) assert response.status_code == status.HTTP_204_NO_CONTENT assert mock.called -async def test_update_project(app: FastAPI, client: AsyncClient, base_params: dict) -> None: +async def test_update_project(app: FastAPI, compute_client: AsyncClient, base_params: dict) -> None: - response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params) + response = await compute_client.post(app.url_path_for("compute:create_compute_project"), json=base_params) assert response.status_code == status.HTTP_201_CREATED params = {"variables": [{"name": "TEST1", "value": "VAL1"}]} - response = await client.put(app.url_path_for("compute:update_compute_project", project_id=base_params["project_id"]), + response = await compute_client.put(app.url_path_for("compute:update_compute_project", project_id=base_params["project_id"]), json=params) assert response.status_code == status.HTTP_200_OK assert response.json()["variables"] == [{"name": "TEST1", "value": "VAL1"}] -async def test_delete_project_invalid_uuid(app: FastAPI, client: AsyncClient) -> None: +async def test_delete_project_invalid_uuid(app: FastAPI, compute_client: AsyncClient) -> None: - response = await client.delete(app.url_path_for("compute:delete_compute_project", project_id=str(uuid.uuid4()))) + response = await compute_client.delete(app.url_path_for("compute:delete_compute_project", project_id=str(uuid.uuid4()))) assert response.status_code == status.HTTP_404_NOT_FOUND -async def test_close_project(app: FastAPI, client: AsyncClient, compute_project: Project) -> None: +async def test_close_project(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None: with asyncio_patch("gns3server.compute.project.Project.close", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:close_compute_project", project_id=compute_project.id)) + response = await compute_client.post(app.url_path_for("compute:close_compute_project", project_id=compute_project.id)) assert response.status_code == status.HTTP_204_NO_CONTENT assert mock.called @@ -147,18 +147,18 @@ async def test_close_project(app: FastAPI, client: AsyncClient, compute_project: # # ProjectHandler._notifications_listening = {compute_project.id: 2} # with asyncio_patch("gns3server.compute.project.Project.close", return_value=True) as mock: -# response = await client.post("/projects/{project_id}/close".format(project_id=compute_project.id)) +# response = await compute_client.post("/projects/{project_id}/close".format(project_id=compute_project.id)) # assert response.status_code == status.HTTP_204_NO_CONTENT # assert not mock.called -async def test_close_project_invalid_uuid(app: FastAPI, client: AsyncClient) -> None: +async def test_close_project_invalid_uuid(app: FastAPI, compute_client: AsyncClient) -> None: - response = await client.post(app.url_path_for("compute:close_compute_project", project_id=str(uuid.uuid4()))) + response = await compute_client.post(app.url_path_for("compute:close_compute_project", project_id=str(uuid.uuid4()))) assert response.status_code == status.HTTP_404_NOT_FOUND -async def test_get_file(app: FastAPI, client: AsyncClient, config, tmpdir) -> None: +async def test_get_file(app: FastAPI, compute_client: AsyncClient, config, tmpdir) -> None: config.settings.Server.projects_path = str(tmpdir) project = ProjectManager.instance().create_project(project_id="01010203-0405-0607-0809-0a0b0c0d0e0b") @@ -166,25 +166,25 @@ async def test_get_file(app: FastAPI, client: AsyncClient, config, tmpdir) -> No with open(os.path.join(project.path, "hello"), "w+") as f: f.write("world") - response = await client.get(app.url_path_for("compute:get_compute_project_file", project_id=project.id, file_path="hello")) + response = await compute_client.get(app.url_path_for("compute:get_compute_project_file", project_id=project.id, file_path="hello")) assert response.status_code == status.HTTP_200_OK assert response.content == b"world" - response = await client.get(app.url_path_for("compute:get_compute_project_file", project_id=project.id, file_path="false")) + response = await compute_client.get(app.url_path_for("compute:get_compute_project_file", project_id=project.id, file_path="false")) assert response.status_code == status.HTTP_404_NOT_FOUND - response = await client.get(app.url_path_for("compute:get_compute_project_file", + response = await compute_client.get(app.url_path_for("compute:get_compute_project_file", project_id=project.id, file_path="../hello")) assert response.status_code == status.HTTP_404_NOT_FOUND -async def test_get_file_forbidden_location(app: FastAPI, client: AsyncClient, config, tmpdir) -> None: +async def test_get_file_forbidden_location(app: FastAPI, compute_client: AsyncClient, config, tmpdir) -> None: config.settings.Server.projects_path = str(tmpdir) project = ProjectManager.instance().create_project(project_id="01010203-0405-0607-0809-0a0b0c0d0e0b") file_path = "foo/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/etc/passwd" - response = await client.get( + response = await compute_client.get( app.url_path_for( "compute:get_compute_project_file", project_id=project.id, @@ -194,12 +194,12 @@ async def test_get_file_forbidden_location(app: FastAPI, client: AsyncClient, co assert response.status_code == status.HTTP_403_FORBIDDEN -async def test_write_file(app: FastAPI, client: AsyncClient, config, tmpdir) -> None: +async def test_write_file(app: FastAPI, compute_client: AsyncClient, config, tmpdir) -> None: config.settings.Server.projects_path = str(tmpdir) project = ProjectManager.instance().create_project(project_id="01010203-0405-0607-0809-0a0b0c0d0e0b") - response = await client.post(app.url_path_for("compute:write_compute_project_file", + response = await compute_client.post(app.url_path_for("compute:write_compute_project_file", project_id=project.id, file_path="hello"), content=b"world") assert response.status_code == status.HTTP_204_NO_CONTENT @@ -207,19 +207,19 @@ async def test_write_file(app: FastAPI, client: AsyncClient, config, tmpdir) -> with open(os.path.join(project.path, "hello")) as f: assert f.read() == "world" - response = await client.post(app.url_path_for("compute:write_compute_project_file", + response = await compute_client.post(app.url_path_for("compute:write_compute_project_file", project_id=project.id, file_path="../hello")) assert response.status_code == status.HTTP_404_NOT_FOUND -async def test_write_file_forbidden_location(app: FastAPI, client: AsyncClient, config, tmpdir) -> None: +async def test_write_file_forbidden_location(app: FastAPI, compute_client: AsyncClient, config, tmpdir) -> None: config.settings.Server.projects_path = str(tmpdir) project = ProjectManager.instance().create_project(project_id="01010203-0405-0607-0809-0a0b0c0d0e0b") file_path = "%2e%2e/hello" - response = await client.post(app.url_path_for("compute:write_compute_project_file", + response = await compute_client.post(app.url_path_for("compute:write_compute_project_file", project_id=project.id, file_path=file_path), content=b"world") assert response.status_code == status.HTTP_403_FORBIDDEN diff --git a/tests/api/routes/compute/test_qemu_nodes.py b/tests/api/routes/compute/test_qemu_nodes.py index e8910ed4..0a418ac7 100644 --- a/tests/api/routes/compute/test_qemu_nodes.py +++ b/tests/api/routes/compute/test_qemu_nodes.py @@ -64,20 +64,20 @@ def base_params(tmpdir, fake_qemu_bin) -> dict: @pytest.fixture -async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> None: +async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project, base_params: dict) -> None: - response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params) + response = await compute_client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params) assert response.status_code == status.HTTP_201_CREATED return response.json() async def test_qemu_create(app: FastAPI, - client: AsyncClient, + compute_client: AsyncClient, compute_project: Project, base_params: dict, fake_qemu_bin: str) -> None: - response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params) + response = await compute_client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["name"] == "PC TEST 1" assert response.json()["project_id"] == compute_project.id @@ -86,14 +86,14 @@ async def test_qemu_create(app: FastAPI, async def test_qemu_create_platform(app: FastAPI, - client: AsyncClient, + compute_client: AsyncClient, compute_project: Project, base_params: dict, fake_qemu_bin: str): base_params["qemu_path"] = None base_params["platform"] = "x86_64" - response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params) + response = await compute_client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["name"] == "PC TEST 1" assert response.json()["project_id"] == compute_project.id @@ -103,7 +103,7 @@ async def test_qemu_create_platform(app: FastAPI, @pytest.mark.asyncio async def test_qemu_create_with_params(app: FastAPI, - client: AsyncClient, + compute_client: AsyncClient, compute_project: Project, base_params: dict, fake_qemu_vm: str): @@ -111,7 +111,7 @@ async def test_qemu_create_with_params(app: FastAPI, params = base_params params["ram"] = 1024 params["hda_disk_image"] = "linux载.img" - response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params) + response = await compute_client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["name"] == "PC TEST 1" assert response.json()["project_id"] == compute_project.id @@ -122,26 +122,26 @@ async def test_qemu_create_with_params(app: FastAPI, @pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows") async def test_qemu_create_with_project_file(app: FastAPI, - client: AsyncClient, + compute_client: AsyncClient, compute_project: Project, base_params: dict, fake_qemu_vm: str) -> None: - response = await client.post(app.url_path_for("compute:write_compute_project_file", + response = await compute_client.post(app.url_path_for("compute:write_compute_project_file", project_id=compute_project.id, file_path="hello.img"), content=b"world") assert response.status_code == status.HTTP_204_NO_CONTENT params = base_params params["hda_disk_image"] = "hello.img" - response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params) + response = await compute_client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["hda_disk_image"] == "hello.img" assert response.json()["hda_disk_image_md5sum"] == "7d793037a0760186574b0282f2f435e7" -async def test_qemu_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict): +async def test_qemu_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict): - response = await client.get(app.url_path_for("compute:get_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"])) + response = await compute_client.get(app.url_path_for("compute:get_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert response.status_code == status.HTTP_200_OK assert response.json()["name"] == "PC TEST 1" assert response.json()["project_id"] == compute_project.id @@ -151,60 +151,60 @@ async def test_qemu_get(app: FastAPI, client: AsyncClient, compute_project: Proj vm["node_id"]) -async def test_qemu_start(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_qemu_start(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.start", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:start_qemu_node", + response = await compute_client.post(app.url_path_for("compute:start_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_qemu_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_qemu_stop(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.stop", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:stop_qemu_node", + response = await compute_client.post(app.url_path_for("compute:stop_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_qemu_reload(app: FastAPI, client: AsyncClient, vm) -> None: +async def test_qemu_reload(app: FastAPI, compute_client: AsyncClient, vm) -> None: with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.reload", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:reload_qemu_node", + response = await compute_client.post(app.url_path_for("compute:reload_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_qemu_suspend(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_qemu_suspend(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.suspend", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:suspend_qemu_node", + response = await compute_client.post(app.url_path_for("compute:suspend_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_qemu_resume(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_qemu_resume(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.resume", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:resume_qemu_node", + response = await compute_client.post(app.url_path_for("compute:resume_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_qemu_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_qemu_delete(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.qemu.Qemu.delete_node", return_value=True) as mock: - response = await client.delete(app.url_path_for("compute:delete_qemu_node", + response = await compute_client.delete(app.url_path_for("compute:delete_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called @@ -212,7 +212,7 @@ async def test_qemu_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None: async def test_qemu_update(app: FastAPI, - client: AsyncClient, + compute_client: AsyncClient, vm: dict, free_console_port: int, fake_qemu_vm: str) -> None: @@ -224,7 +224,7 @@ async def test_qemu_update(app: FastAPI, "hdb_disk_image": "linux载.img" } - response = await client.put(app.url_path_for("compute:update_qemu_node", + response = await compute_client.put(app.url_path_for("compute:update_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"]), json=params) assert response.status_code == status.HTTP_200_OK @@ -234,7 +234,7 @@ async def test_qemu_update(app: FastAPI, assert response.json()["ram"] == 1024 -async def test_qemu_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_qemu_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "type": "nio_udp", @@ -244,7 +244,7 @@ async def test_qemu_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) } with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.add_ubridge_udp_connection"): - await client.put(app.url_path_for("compute:update_qemu_node", + await compute_client.put(app.url_path_for("compute:update_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"]), json={"adapters": 2}) @@ -253,12 +253,12 @@ async def test_qemu_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) node_id=vm["node_id"], adapter_number="1", port_number="0") - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_udp" -async def test_qemu_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_qemu_nio_update_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "type": "nio_udp", @@ -267,7 +267,7 @@ async def test_qemu_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) "rhost": "127.0.0.1" } - await client.put(app.url_path_for("compute:update_qemu_node", + await compute_client.put(app.url_path_for("compute:update_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"]), json={"adapters": 2}) @@ -277,7 +277,7 @@ async def test_qemu_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) adapter_number="1", port_number="0") - await client.post(url, json=params) + await compute_client.post(url, json=params) params["filters"] = {} @@ -286,12 +286,12 @@ async def test_qemu_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) node_id=vm["node_id"], adapter_number="1", port_number="0") - response = await client.put(url, json=params) + response = await compute_client.put(url, json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_udp" -async def test_qemu_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_qemu_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "type": "nio_udp", @@ -301,7 +301,7 @@ async def test_qemu_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> N } with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM._ubridge_send"): - await client.put(app.url_path_for("compute:update_qemu_node", + await compute_client.put(app.url_path_for("compute:update_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"]), json={"adapters": 2}) @@ -310,30 +310,30 @@ async def test_qemu_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> N node_id=vm["node_id"], adapter_number="1", port_number="0") - await client.post(url, json=params) + await compute_client.post(url, json=params) url = app.url_path_for("compute:delete_qemu_node_nio", project_id=vm["project_id"], node_id=vm["node_id"], adapter_number="1", port_number="0") - response = await client.delete(url) + response = await compute_client.delete(url) assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_qemu_list_binaries(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_qemu_list_binaries(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: ret = [{"path": "/tmp/1", "version": "2.2.0"}, {"path": "/tmp/2", "version": "2.1.0"}] with asyncio_patch("gns3server.compute.qemu.Qemu.binary_list", return_value=ret) as mock: - response = await client.get(app.url_path_for("compute:get_qemu_binaries")) + response = await compute_client.get(app.url_path_for("compute:get_qemu_binaries")) assert mock.called_with(None) assert response.status_code == status.HTTP_200_OK assert response.json() == ret -# async def test_qemu_list_binaries_filter(app: FastAPI, client: AsyncClient, vm: dict) -> None: +# async def test_qemu_list_binaries_filter(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: # # ret = [ # {"path": "/tmp/x86_64", "version": "2.2.0"}, @@ -342,25 +342,25 @@ async def test_qemu_list_binaries(app: FastAPI, client: AsyncClient, vm: dict) - # ] # # with asyncio_patch("gns3server.compute.qemu.Qemu.binary_list", return_value=ret) as mock: -# response = await client.get(app.url_path_for("compute:get_qemu_binaries"), +# response = await compute_client.get(app.url_path_for("compute:get_qemu_binaries"), # json={"archs": ["i386"]}) # assert response.status_code == status.HTTP_200_OK # assert mock.called_with(["i386"]) # assert response.json() == ret -async def test_images(app: FastAPI, client: AsyncClient, fake_qemu_vm) -> None: +async def test_images(app: FastAPI, compute_client: AsyncClient, fake_qemu_vm) -> None: - response = await client.get(app.url_path_for("compute:get_qemu_images")) + response = await compute_client.get(app.url_path_for("compute:get_qemu_images")) assert response.status_code == status.HTTP_200_OK assert {"filename": "linux载.img", "path": "linux载.img", "md5sum": "c4ca4238a0b923820dcc509a6f75849b", "filesize": 1} in response.json() -async def test_upload_image(app: FastAPI, client: AsyncClient, tmpdir: str) -> None: +async def test_upload_image(app: FastAPI, compute_client: AsyncClient, tmpdir: str) -> None: with patch("gns3server.compute.Qemu.get_images_directory", return_value=str(tmpdir)): - response = await client.post(app.url_path_for("compute:upload_qemu_image", + response = await compute_client.post(app.url_path_for("compute:upload_qemu_image", filename="test2使"), content=b"TEST") assert response.status_code == status.HTTP_204_NO_CONTENT @@ -372,11 +372,11 @@ async def test_upload_image(app: FastAPI, client: AsyncClient, tmpdir: str) -> N assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf" -async def test_upload_image_ova(app: FastAPI, client: AsyncClient, tmpdir:str) -> None: +async def test_upload_image_ova(app: FastAPI, compute_client: AsyncClient, tmpdir:str) -> None: with patch("gns3server.compute.Qemu.get_images_directory", return_value=str(tmpdir)): - response = await client.post(app.url_path_for("compute:upload_qemu_image", + response = await compute_client.post(app.url_path_for("compute:upload_qemu_image", filename="test2.ova/test2.vmdk"), content=b"TEST") assert response.status_code == status.HTTP_204_NO_CONTENT @@ -388,42 +388,42 @@ async def test_upload_image_ova(app: FastAPI, client: AsyncClient, tmpdir:str) - assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf" -async def test_upload_image_forbidden_location(app: FastAPI, client: AsyncClient, tmpdir: str) -> None: +async def test_upload_image_forbidden_location(app: FastAPI, compute_client: AsyncClient, tmpdir: str) -> None: - response = await client.post(app.url_path_for("compute:upload_qemu_image", + response = await compute_client.post(app.url_path_for("compute:upload_qemu_image", filename="/qemu/images/../../test2"), content=b"TEST") assert response.status_code == status.HTTP_403_FORBIDDEN -async def test_download_image(app: FastAPI, client: AsyncClient, images_dir: str) -> None: +async def test_download_image(app: FastAPI, compute_client: AsyncClient, images_dir: str) -> None: - response = await client.post(app.url_path_for("compute:upload_qemu_image", filename="test3"), content=b"TEST") + response = await compute_client.post(app.url_path_for("compute:upload_qemu_image", filename="test3"), content=b"TEST") assert response.status_code == status.HTTP_204_NO_CONTENT - response = await client.get(app.url_path_for("compute:download_qemu_image", filename="test3")) + response = await compute_client.get(app.url_path_for("compute:download_qemu_image", filename="test3")) assert response.status_code == status.HTTP_200_OK -async def test_download_image_forbidden_location(app: FastAPI, client: AsyncClient, tmpdir) -> None: +async def test_download_image_forbidden_location(app: FastAPI, compute_client: AsyncClient, tmpdir) -> None: file_path = "foo/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/etc/passwd" - response = await client.get(app.url_path_for("compute:download_qemu_image", filename=file_path)) + response = await compute_client.get(app.url_path_for("compute:download_qemu_image", filename=file_path)) assert response.status_code == status.HTTP_403_FORBIDDEN @pytest.mark.skipif(not sys.platform.startswith("win") and os.getuid() == 0, reason="Root can delete any image") -async def test_upload_image_permission_denied(app: FastAPI, client: AsyncClient, images_dir: str) -> None: +async def test_upload_image_permission_denied(app: FastAPI, compute_client: AsyncClient, images_dir: str) -> None: with open(os.path.join(images_dir, "QEMU", "test2.tmp"), "w+") as f: f.write("") os.chmod(os.path.join(images_dir, "QEMU", "test2.tmp"), 0) - response = await client.post(app.url_path_for("compute:upload_qemu_image", filename="test2"), content=b"TEST") + response = await compute_client.post(app.url_path_for("compute:upload_qemu_image", filename="test2"), content=b"TEST") assert response.status_code == status.HTTP_409_CONFLICT @pytest.mark.asyncio -async def test_create_img_relative(app: FastAPI, client: AsyncClient): +async def test_create_img_relative(app: FastAPI, compute_client: AsyncClient): params = { "qemu_img": "/tmp/qemu-img", @@ -436,11 +436,11 @@ async def test_create_img_relative(app: FastAPI, client: AsyncClient): "size": 100 } with asyncio_patch("gns3server.compute.Qemu.create_disk"): - response = await client.post(app.url_path_for("compute:create_qemu_image"), json=params) + response = await compute_client.post(app.url_path_for("compute:create_qemu_image"), json=params) assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_create_img_absolute_non_local(app: FastAPI, client: AsyncClient, config) -> None: +async def test_create_img_absolute_non_local(app: FastAPI, compute_client: AsyncClient, config) -> None: config.settings.Server.local = False params = { @@ -454,11 +454,11 @@ async def test_create_img_absolute_non_local(app: FastAPI, client: AsyncClient, "size": 100 } with asyncio_patch("gns3server.compute.Qemu.create_disk"): - response = await client.post(app.url_path_for("compute:create_qemu_image"), json=params) + response = await compute_client.post(app.url_path_for("compute:create_qemu_image"), json=params) assert response.status_code == 403 -async def test_create_img_absolute_local(app: FastAPI, client: AsyncClient, config) -> None: +async def test_create_img_absolute_local(app: FastAPI, compute_client: AsyncClient, config) -> None: params = { "qemu_img": "/tmp/qemu-img", @@ -471,36 +471,36 @@ async def test_create_img_absolute_local(app: FastAPI, client: AsyncClient, conf "size": 100 } with asyncio_patch("gns3server.compute.Qemu.create_disk"): - response = await client.post(app.url_path_for("compute:create_qemu_image"), json=params) + response = await compute_client.post(app.url_path_for("compute:create_qemu_image"), json=params) assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_capabilities(app: FastAPI, client: AsyncClient) -> None: +async def test_capabilities(app: FastAPI, compute_client: AsyncClient) -> None: with asyncio_patch("gns3server.compute.Qemu.get_kvm_archs", return_value=["x86_64"]): - response = await client.get(app.url_path_for("compute:get_qemu_capabilities")) + response = await compute_client.get(app.url_path_for("compute:get_qemu_capabilities")) assert response.json()["kvm"] == ["x86_64"] async def test_qemu_duplicate(app: FastAPI, - client: AsyncClient, + compute_client: AsyncClient, compute_project: Project, vm: dict, base_params: dict) -> None: # create destination node first - response = await client.post(app.url_path_for("compute:create_qemu_node", + response = await compute_client.post(app.url_path_for("compute:create_qemu_node", project_id=vm["project_id"]), json=base_params) assert response.status_code == status.HTTP_201_CREATED params = {"destination_node_id": response.json()["node_id"]} - response = await client.post(app.url_path_for("compute:duplicate_qemu_node", + response = await compute_client.post(app.url_path_for("compute:duplicate_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"]), json=params) assert response.status_code == status.HTTP_201_CREATED @pytest.mark.asyncio -async def test_qemu_start_capture(app: FastAPI, client: AsyncClient, vm): +async def test_qemu_start_capture(app: FastAPI, compute_client: AsyncClient, vm): params = { "capture_file_name": "test.pcap", @@ -515,14 +515,14 @@ async def test_qemu_start_capture(app: FastAPI, client: AsyncClient, vm): with patch("gns3server.compute.qemu.qemu_vm.QemuVM.is_running", return_value=True): with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.start_capture") as mock: - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_200_OK assert mock.called assert "test.pcap" in response.json()["pcap_file_path"] @pytest.mark.asyncio -async def test_qemu_stop_capture(app: FastAPI, client: AsyncClient, vm): +async def test_qemu_stop_capture(app: FastAPI, compute_client: AsyncClient, vm): url = app.url_path_for("compute:stop_qemu_node_capture", project_id=vm["project_id"], @@ -532,15 +532,15 @@ async def test_qemu_stop_capture(app: FastAPI, client: AsyncClient, vm): with patch("gns3server.compute.qemu.qemu_vm.QemuVM.is_running", return_value=True): with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.stop_capture") as mock: - response = await client.post(url) + response = await compute_client.post(url) assert response.status_code == status.HTTP_204_NO_CONTENT assert mock.called # @pytest.mark.asyncio -# async def test_qemu_pcap(app: FastAPI, client: AsyncClient, vm, compute_project): +# async def test_qemu_pcap(app: FastAPI, compute_client: AsyncClient, vm, compute_project): # # with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.get_nio"): # with asyncio_patch("gns3server.compute.qemu.Qemu.stream_pcap_file"): -# response = await client.get("/projects/{project_id}/qemu/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True) +# response = await compute_client.get("/projects/{project_id}/qemu/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True) # assert response.status_code == status.HTTP_200_OK diff --git a/tests/api/routes/compute/test_virtualbox_nodes.py b/tests/api/routes/compute/test_virtualbox_nodes.py index 8615e8ec..30ffe139 100644 --- a/tests/api/routes/compute/test_virtualbox_nodes.py +++ b/tests/api/routes/compute/test_virtualbox_nodes.py @@ -28,7 +28,7 @@ pytestmark = pytest.mark.asyncio @pytest.fixture(scope="function") -async def vm(app: FastAPI, client: AsyncClient, compute_project: Project) -> None: +async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None: vboxmanage_path = "/fake/VboxManage" params = { @@ -38,7 +38,7 @@ async def vm(app: FastAPI, client: AsyncClient, compute_project: Project) -> Non } with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.create", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:create_virtualbox_node", project_id=compute_project.id), + response = await compute_client.post(app.url_path_for("compute:create_virtualbox_node", project_id=compute_project.id), json=params) assert mock.called assert response.status_code == status.HTTP_201_CREATED @@ -47,7 +47,7 @@ async def vm(app: FastAPI, client: AsyncClient, compute_project: Project) -> Non return response.json() -async def test_vbox_create(app: FastAPI, client: AsyncClient, compute_project: Project) -> None: +async def test_vbox_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None: params = { "name": "VM1", @@ -56,16 +56,16 @@ async def test_vbox_create(app: FastAPI, client: AsyncClient, compute_project: P } with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.create", return_value=True): - response = await client.post(app.url_path_for("compute:create_virtualbox_node", project_id=compute_project.id), + response = await compute_client.post(app.url_path_for("compute:create_virtualbox_node", project_id=compute_project.id), json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["name"] == "VM1" assert response.json()["project_id"] == compute_project.id -async def test_vbox_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None: +async def test_vbox_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None: - response = await client.get(app.url_path_for("compute:get_virtualbox_node", + response = await compute_client.get(app.url_path_for("compute:get_virtualbox_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert response.status_code == status.HTTP_200_OK @@ -73,58 +73,58 @@ async def test_vbox_get(app: FastAPI, client: AsyncClient, compute_project: Proj assert response.json()["project_id"] == compute_project.id -async def test_vbox_start(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vbox_start(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.start", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:start_virtualbox_node", + response = await compute_client.post(app.url_path_for("compute:start_virtualbox_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vbox_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vbox_stop(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.stop", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:stop_virtualbox_node", + response = await compute_client.post(app.url_path_for("compute:stop_virtualbox_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vbox_suspend(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vbox_suspend(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.suspend", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:suspend_virtualbox_node", + response = await compute_client.post(app.url_path_for("compute:suspend_virtualbox_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vbox_resume(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vbox_resume(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.resume", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:resume_virtualbox_node", + response = await compute_client.post(app.url_path_for("compute:resume_virtualbox_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vbox_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vbox_reload(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.reload", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:reload_virtualbox_node", + response = await compute_client.post(app.url_path_for("compute:reload_virtualbox_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vbox_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vbox_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "type": "nio_udp", @@ -140,7 +140,7 @@ async def test_vbox_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) port_number="0") with asyncio_patch('gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.adapter_add_nio_binding') as mock: - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert mock.called args, kwgars = mock.call_args assert args[0] == 0 @@ -150,7 +150,7 @@ async def test_vbox_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) # @pytest.mark.asyncio -# async def test_vbox_nio_update_udp(app: FastAPI, client: AsyncClient, vm): +# async def test_vbox_nio_update_udp(app: FastAPI, compute_client: AsyncClient, vm): # # params = { # "type": "nio_udp", @@ -162,13 +162,13 @@ async def test_vbox_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) # # with asyncio_patch('gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.ethernet_adapters'): # with asyncio_patch('gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.adapter_remove_nio_binding'): -# response = await client.put("/projects/{project_id}/virtualbox/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params) +# response = await compute_client.put("/projects/{project_id}/virtualbox/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params) # # assert response.status_code == status.HTTP_201_CREATED # assert response.json()["type"] == "nio_udp" -async def test_vbox_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vbox_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: url = app.url_path_for("compute:delete_virtualbox_node_nio", project_id=vm["project_id"], @@ -177,7 +177,7 @@ async def test_vbox_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> N port_number="0") with asyncio_patch('gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.adapter_remove_nio_binding') as mock: - response = await client.delete(url) + response = await compute_client.delete(url) assert mock.called args, kwgars = mock.call_args assert args[0] == 0 @@ -185,14 +185,14 @@ async def test_vbox_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> N @pytest.mark.asyncio -async def test_vbox_update(app: FastAPI, client: AsyncClient, vm, free_console_port): +async def test_vbox_update(app: FastAPI, compute_client: AsyncClient, vm, free_console_port): params = { "name": "test", "console": free_console_port } - response = await client.put(app.url_path_for("compute:update_virtualbox_node", + response = await compute_client.put(app.url_path_for("compute:update_virtualbox_node", project_id=vm["project_id"], node_id=vm["node_id"]), json=params) assert response.status_code == status.HTTP_200_OK @@ -201,7 +201,7 @@ async def test_vbox_update(app: FastAPI, client: AsyncClient, vm, free_console_p @pytest.mark.asyncio -async def test_virtualbox_start_capture(app: FastAPI, client: AsyncClient, vm): +async def test_virtualbox_start_capture(app: FastAPI, compute_client: AsyncClient, vm): params = { "capture_file_name": "test.pcap", @@ -216,14 +216,14 @@ async def test_virtualbox_start_capture(app: FastAPI, client: AsyncClient, vm): with patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.is_running", return_value=True): with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.start_capture") as mock: - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_200_OK assert mock.called assert "test.pcap" in response.json()["pcap_file_path"] @pytest.mark.asyncio -async def test_virtualbox_stop_capture(app: FastAPI, client: AsyncClient, vm): +async def test_virtualbox_stop_capture(app: FastAPI, compute_client: AsyncClient, vm): url = app.url_path_for("compute:stop_virtualbox_node_capture", project_id=vm["project_id"], @@ -233,15 +233,15 @@ async def test_virtualbox_stop_capture(app: FastAPI, client: AsyncClient, vm): with patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.is_running", return_value=True): with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.stop_capture") as mock: - response = await client.post(url) + response = await compute_client.post(url) assert response.status_code == status.HTTP_204_NO_CONTENT assert mock.called # @pytest.mark.asyncio -# async def test_virtualbox_pcap(app: FastAPI, client: AsyncClient, vm, compute_project): +# async def test_virtualbox_pcap(app: FastAPI, compute_client: AsyncClient, vm, compute_project): # # with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.get_nio"): # with asyncio_patch("gns3server.compute.virtualbox.VirtualBox.stream_pcap_file"): -# response = await client.get("/projects/{project_id}/virtualbox/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True) +# response = await compute_client.get("/projects/{project_id}/virtualbox/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True) # assert response.status_code == status.HTTP_200_OK diff --git a/tests/api/routes/compute/test_vmware_nodes.py b/tests/api/routes/compute/test_vmware_nodes.py index d203471c..7ac81ef4 100644 --- a/tests/api/routes/compute/test_vmware_nodes.py +++ b/tests/api/routes/compute/test_vmware_nodes.py @@ -28,7 +28,7 @@ pytestmark = pytest.mark.asyncio @pytest.fixture(scope="function") -async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, vmx_path: str) -> dict: +async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vmx_path: str) -> dict: params = { "name": "VMTEST", @@ -37,7 +37,7 @@ async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, vmx_pa } with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.create", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:create_vmware_node", project_id=compute_project.id), + response = await compute_client.post(app.url_path_for("compute:create_vmware_node", project_id=compute_project.id), json=params) assert mock.called assert response.status_code == status.HTTP_201_CREATED @@ -56,7 +56,7 @@ def vmx_path(tmpdir: str) -> str: return path -async def test_vmware_create(app: FastAPI, client: AsyncClient, compute_project: Project, vmx_path: str) -> None: +async def test_vmware_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vmx_path: str) -> None: params = { "name": "VM1", @@ -65,72 +65,72 @@ async def test_vmware_create(app: FastAPI, client: AsyncClient, compute_project: } with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.create", return_value=True): - response = await client.post(app.url_path_for("compute:create_vmware_node", project_id=compute_project.id), + response = await compute_client.post(app.url_path_for("compute:create_vmware_node", project_id=compute_project.id), json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["name"] == "VM1" assert response.json()["project_id"] == compute_project.id -async def test_vmware_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None: +async def test_vmware_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None: - response = await client.get(app.url_path_for("compute:get_vmware_node", project_id=vm["project_id"], node_id=vm["node_id"])) + response = await compute_client.get(app.url_path_for("compute:get_vmware_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert response.status_code == status.HTTP_200_OK assert response.json()["name"] == "VMTEST" assert response.json()["project_id"] == compute_project.id -async def test_vmware_start(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vmware_start(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.start", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:start_vmware_node", + response = await compute_client.post(app.url_path_for("compute:start_vmware_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vmware_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vmware_stop(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.stop", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:stop_vmware_node", + response = await compute_client.post(app.url_path_for("compute:stop_vmware_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vmware_suspend(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vmware_suspend(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.suspend", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:suspend_vmware_node", + response = await compute_client.post(app.url_path_for("compute:suspend_vmware_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vmware_resume(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vmware_resume(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.resume", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:resume_vmware_node", + response = await compute_client.post(app.url_path_for("compute:resume_vmware_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vmware_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vmware_reload(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.reload", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:reload_vmware_node", + response = await compute_client.post(app.url_path_for("compute:reload_vmware_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vmware_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vmware_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "type": "nio_udp", @@ -146,7 +146,7 @@ async def test_vmware_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict port_number="0") with asyncio_patch('gns3server.compute.vmware.vmware_vm.VMwareVM.adapter_add_nio_binding') as mock: - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert mock.called args, kwgars = mock.call_args assert args[0] == 0 @@ -156,7 +156,7 @@ async def test_vmware_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict # @pytest.mark.asyncio -# async def test_vmware_nio_update_udp(app: FastAPI, client: AsyncClient, vm): +# async def test_vmware_nio_update_udp(app: FastAPI, compute_client: AsyncClient, vm): # # params = { # "type": "nio_udp", @@ -169,12 +169,12 @@ async def test_vmware_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict # with asyncio_patch('gns3server.compute.vmware.vmware_vm.VMwareVM._ubridge_send'): # with asyncio_patch('gns3server.compute.vmware.vmware_vm.VMwareVM.ethernet_adapters'): # with patch('gns3server.compute.vmware.vmware_vm.VMwareVM._get_vnet') as mock: -# response = await client.put("/projects/{project_id}/vmware/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params) +# response = await compute_client.put("/projects/{project_id}/vmware/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params) # assert response.status_code == status.HTTP_201_CREATED # assert response.json()["type"] == "nio_udp" -async def test_vmware_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vmware_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: url = app.url_path_for("compute:delete_vmware_node_nio", project_id=vm["project_id"], @@ -183,7 +183,7 @@ async def test_vmware_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> port_number="0") with asyncio_patch('gns3server.compute.vmware.vmware_vm.VMwareVM.adapter_remove_nio_binding') as mock: - response = await client.delete(url) + response = await compute_client.delete(url) assert mock.called args, kwgars = mock.call_args assert args[0] == 0 @@ -191,14 +191,14 @@ async def test_vmware_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vmware_update(app: FastAPI, client: AsyncClient, vm: dict, free_console_port: int) -> None: +async def test_vmware_update(app: FastAPI, compute_client: AsyncClient, vm: dict, free_console_port: int) -> None: params = { "name": "test", "console": free_console_port } - response = await client.put(app.url_path_for("compute:update_vmware_node", + response = await compute_client.put(app.url_path_for("compute:update_vmware_node", project_id=vm["project_id"], node_id=vm["node_id"]), json=params) assert response.status_code == status.HTTP_200_OK @@ -206,7 +206,7 @@ async def test_vmware_update(app: FastAPI, client: AsyncClient, vm: dict, free_c assert response.json()["console"] == free_console_port -async def test_vmware_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vmware_start_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "capture_file_name": "test.pcap", @@ -222,13 +222,13 @@ async def test_vmware_start_capture(app: FastAPI, client: AsyncClient, vm: dict) with patch("gns3server.compute.vmware.vmware_vm.VMwareVM.is_running", return_value=True): with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.start_capture") as mock: - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_200_OK assert mock.called assert "test.pcap" in response.json()["pcap_file_path"] -async def test_vmware_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vmware_stop_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: url = app.url_path_for("compute:stop_vmware_node_capture", project_id=vm["project_id"], @@ -238,15 +238,15 @@ async def test_vmware_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) with patch("gns3server.compute.vmware.vmware_vm.VMwareVM.is_running", return_value=True): with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.stop_capture") as mock: - response = await client.post(url) + response = await compute_client.post(url) assert response.status_code == status.HTTP_204_NO_CONTENT assert mock.called # @pytest.mark.asyncio -# async def test_vmware_pcap(app: FastAPI, client: AsyncClient, vm, compute_project): +# async def test_vmware_pcap(app: FastAPI, compute_client: AsyncClient, vm, compute_project): # # with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.get_nio"): # with asyncio_patch("gns3server.compute.vmware.VMware.stream_pcap_file"): -# response = await client.get("/projects/{project_id}/vmware/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True) +# response = await compute_client.get("/projects/{project_id}/vmware/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True) # assert response.status_code == status.HTTP_200_OK diff --git a/tests/api/routes/compute/test_vpcs_nodes.py b/tests/api/routes/compute/test_vpcs_nodes.py index c092116d..e49a84bb 100644 --- a/tests/api/routes/compute/test_vpcs_nodes.py +++ b/tests/api/routes/compute/test_vpcs_nodes.py @@ -28,47 +28,47 @@ pytestmark = pytest.mark.asyncio @pytest.fixture -async def vm(app: FastAPI, client: AsyncClient, compute_project: Project) -> None: +async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None: params = {"name": "PC TEST 1"} - response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params) + response = await compute_client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params) assert response.status_code == status.HTTP_201_CREATED return response.json() -async def test_vpcs_create(app: FastAPI, client: AsyncClient, compute_project: Project) -> None: +async def test_vpcs_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None: params = {"name": "PC TEST 1"} - response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params) + response = await compute_client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["name"] == "PC TEST 1" assert response.json()["project_id"] == compute_project.id -async def test_vpcs_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None: +async def test_vpcs_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None: - response = await client.get(app.url_path_for("compute:get_vpcs_node", project_id=vm["project_id"], node_id=vm["node_id"])) + response = await compute_client.get(app.url_path_for("compute:get_vpcs_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert response.status_code == status.HTTP_200_OK assert response.json()["name"] == "PC TEST 1" assert response.json()["project_id"] == compute_project.id assert response.json()["status"] == "stopped" -async def test_vpcs_create_startup_script(app: FastAPI, client: AsyncClient, compute_project: Project) -> None: +async def test_vpcs_create_startup_script(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None: params = { "name": "PC TEST 1", "startup_script": "ip 192.168.1.2\necho TEST" } - response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params) + response = await compute_client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["name"] == "PC TEST 1" assert response.json()["project_id"] == compute_project.id async def test_vpcs_create_port(app: FastAPI, - client: AsyncClient, + compute_client: AsyncClient, compute_project: Project, free_console_port: int) -> None: @@ -77,14 +77,14 @@ async def test_vpcs_create_port(app: FastAPI, "console": free_console_port } - response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params) + response = await compute_client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["name"] == "PC TEST 1" assert response.json()["project_id"] == compute_project.id assert response.json()["console"] == free_console_port -async def test_vpcs_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vpcs_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "type": "nio_udp", @@ -100,12 +100,12 @@ async def test_vpcs_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) port_number="0") with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.add_ubridge_udp_connection"): - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_udp" -async def test_vpcs_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vpcs_nio_update_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "type": "nio_udp", @@ -121,7 +121,7 @@ async def test_vpcs_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) port_number="0") with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.add_ubridge_udp_connection"): - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_201_CREATED params["filters"] = {} @@ -130,12 +130,12 @@ async def test_vpcs_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) node_id=vm["node_id"], adapter_number="0", port_number="0") - response = await client.put(url, json=params) + response = await compute_client.put(url, json=params) assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_udp" -async def test_vpcs_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vpcs_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "type": "nio_udp", @@ -150,73 +150,73 @@ async def test_vpcs_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> N node_id=vm["node_id"], adapter_number="0", port_number="0") - await client.post(url, json=params) + await compute_client.post(url, json=params) url = app.url_path_for("compute:delete_vpcs_node_nio", project_id=vm["project_id"], node_id=vm["node_id"], adapter_number="0", port_number="0") - response = await client.delete(url) + response = await compute_client.delete(url) assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vpcs_start(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vpcs_start(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.start", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:start_vpcs_node", + response = await compute_client.post(app.url_path_for("compute:start_vpcs_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vpcs_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vpcs_stop(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.stop", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:stop_vpcs_node", + response = await compute_client.post(app.url_path_for("compute:stop_vpcs_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vpcs_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vpcs_reload(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.reload", return_value=True) as mock: - response = await client.post(app.url_path_for("compute:reload_vpcs_node", + response = await compute_client.post(app.url_path_for("compute:reload_vpcs_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vpcs_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vpcs_delete(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: with asyncio_patch("gns3server.compute.vpcs.VPCS.delete_node", return_value=True) as mock: - response = await client.delete(app.url_path_for("compute:delete_vpcs_node", + response = await compute_client.delete(app.url_path_for("compute:delete_vpcs_node", project_id=vm["project_id"], node_id=vm["node_id"])) assert mock.called assert response.status_code == status.HTTP_204_NO_CONTENT -async def test_vpcs_duplicate(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None: +async def test_vpcs_duplicate(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None: # create destination node first params = {"name": "PC TEST 1"} - response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params) + response = await compute_client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params) assert response.status_code == status.HTTP_201_CREATED params = {"destination_node_id": response.json()["node_id"]} - response = await client.post(app.url_path_for("compute:duplicate_vpcs_node", + response = await compute_client.post(app.url_path_for("compute:duplicate_vpcs_node", project_id=vm["project_id"], node_id=vm["node_id"]), json=params) assert response.status_code == status.HTTP_201_CREATED -async def test_vpcs_update(app: FastAPI, client: AsyncClient, vm: dict, free_console_port: int) -> None: +async def test_vpcs_update(app: FastAPI, compute_client: AsyncClient, vm: dict, free_console_port: int) -> None: console_port = free_console_port params = { @@ -224,7 +224,7 @@ async def test_vpcs_update(app: FastAPI, client: AsyncClient, vm: dict, free_con "console": console_port } - response = await client.put(app.url_path_for("compute:update_vpcs_node", + response = await compute_client.put(app.url_path_for("compute:update_vpcs_node", project_id=vm["project_id"], node_id=vm["node_id"]), json=params) assert response.status_code == status.HTTP_200_OK @@ -232,7 +232,7 @@ async def test_vpcs_update(app: FastAPI, client: AsyncClient, vm: dict, free_con assert response.json()["console"] == console_port -async def test_vpcs_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vpcs_start_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: params = { "capture_file_name": "test.pcap", @@ -247,13 +247,13 @@ async def test_vpcs_start_capture(app: FastAPI, client: AsyncClient, vm: dict) - with patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.is_running", return_value=True): with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.start_capture") as mock: - response = await client.post(url, json=params) + response = await compute_client.post(url, json=params) assert response.status_code == status.HTTP_200_OK assert mock.called assert "test.pcap" in response.json()["pcap_file_path"] -async def test_vpcs_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None: +async def test_vpcs_stop_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: url = app.url_path_for("compute:stop_vpcs_node_capture", project_id=vm["project_id"], @@ -263,15 +263,15 @@ async def test_vpcs_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> with patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.is_running", return_value=True): with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.stop_capture") as mock: - response = await client.post(url) + response = await compute_client.post(url) assert response.status_code == status.HTTP_204_NO_CONTENT assert mock.called # @pytest.mark.asyncio -# async def test_vpcs_pcap(app: FastAPI, client: AsyncClient, vm, compute_project: Project): +# async def test_vpcs_pcap(app: FastAPI, compute_client: AsyncClient, vm, compute_project: Project): # # with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.get_nio"): # with asyncio_patch("gns3server.compute.vpcs.VPCS.stream_pcap_file"): -# response = await client.get("/projects/{project_id}/vpcs/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True) +# response = await compute_client.get("/projects/{project_id}/vpcs/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True) # assert response.status_code == status.HTTP_200_OK diff --git a/tests/api/routes/controller/test_computes.py b/tests/api/routes/controller/test_computes.py index 290e7461..70134771 100644 --- a/tests/api/routes/controller/test_computes.py +++ b/tests/api/routes/controller/test_computes.py @@ -190,24 +190,3 @@ class TestComputeFeatures: # 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 -# @pytest.mark.asyncio -# async def test_compute_endpoint(controller_api): -# -# params = { -# "compute_id": "my_compute", -# "protocol": "http", -# "host": "localhost", -# "port": 84, -# "user": "julien", -# "password": "secure" -# } -# -# response = await controller_api.post("/computes", params) -# assert response.status_code == 201 -# -# response = await controller_api.get("/computes/endpoint/my_compute/qemu/images") -# assert response.status_code == 200 -# assert response.json['endpoint'] == 'http://localhost:84/v2/compute/qemu/images' diff --git a/tests/conftest.py b/tests/conftest.py index 5cd3fa43..7863cb05 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -6,6 +6,7 @@ import sys import os import uuid import configparser +import base64 from fastapi import FastAPI from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine @@ -180,6 +181,18 @@ async def client(base_client: AsyncClient) -> AsyncClient: return base_client +@pytest.fixture +async def compute_client(base_client: AsyncClient) -> AsyncClient: + + # default compute username is 'admin' + base64_credentials = base64.b64encode(b"admin:").decode("ascii") + base_client.headers = { + **base_client.headers, + "Authorization": f"Basic {base64_credentials}", + } + return base_client + + @pytest.fixture def controller_config_path(tmpdir): @@ -394,7 +407,6 @@ def run_around_tests(monkeypatch, config, port_manager):#port_manager, controlle config.settings.Server.ubridge_path = os.path.join(tmppath, 'bin', 'ubridge') config.settings.Server.local = True - config.settings.Server.enable_http_auth = False # Prevent executions of the VM if we forgot to mock something config.settings.VirtualBox.vboxmanage_path = tmppath diff --git a/tests/test_config.py b/tests/test_config.py index 1f016b91..0b00a0eb 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -99,10 +99,10 @@ def test_reload(tmpdir): def test_server_password_hidden(): - server_settings = {"Server": {"password": "password123"}} + server_settings = {"Server": {"compute_password": "password123"}} config = ServerConfig(**server_settings) - assert str(config.Server.password) == "**********" - assert config.Server.password.get_secret_value() == "password123" + assert str(config.Server.compute_password) == "**********" + assert config.Server.compute_password.get_secret_value() == "password123" @pytest.mark.parametrize( @@ -124,10 +124,7 @@ def test_server_password_hidden(): ({"enable_ssl": True, "certfile": "/path/to/certfile", "certkey": "/path/to/certkey"}, True), ({"enable_ssl": True}, True), ({"enable_ssl": True, "certfile": "/path/to/certfile"}, True), - ({"enable_ssl": True, "certkey": "/path/to/certkey"}, True), - ({"enable_http_auth": True, "user": "user1"}, False), - ({"enable_http_auth": True, "user": ""}, True), - ({"enable_http_auth": True}, True), + ({"enable_ssl": True, "certkey": "/path/to/certkey"}, True) ) ) def test_server_settings(settings: dict, exception_expected: bool):