arno/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server synced 2024-11-19 06:48:10 +00:00
Code Issues Releases Wiki Activity

Merge pull request #1999 from GNS3/compute-authentication

Browse Source 
Secure controller to compute communication
This commit is contained in:
Jeremy Grossmann 2021-11-18 18:48:04 +10:30 committed by GitHub
commit 488146e802
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
24 changed files with 483 additions and 416 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
gns3server/api/routes/compute/__init__.py
View File

@ -15,13 +15,15 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
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",
)

0
gns3server/api/routes/compute/dependencies/__init__.py Normal file
View File

37
gns3server/api/routes/compute/dependencies/authentication.py Normal file
View File

@ -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 <http://www.gnu.org/licenses/>.
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"},
)

32
gns3server/api/routes/compute/notifications.py
View File

@ -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:

8
gns3server/controller/__init__.py
View File

@ -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):

3
gns3server/controller/compute.py
View File

@ -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
)

13
gns3server/schemas/config.py
View File

@ -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):

15
gns3server/server.py
View File

@ -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):

1
gns3server/services/authentication.py
View File

@ -35,6 +35,7 @@ DEFAULT_JWT_SECRET_KEY = "efd08eccec3bd0a1be2e086670e5efa90969c68d07e072d7354a76
class AuthService:
def hash_password(self, password: str) -> str:
return pwd_context.hash(password)

8
tests/api/routes/compute/test_capabilities.py
View File

@ -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__,

44
tests/api/routes/compute/test_cloud_nodes.py
View File

@ -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",

22
tests/api/routes/compute/test_compute.py
View File

@ -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

64
tests/api/routes/compute/test_docker_nodes.py
View File

@ -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

26
tests/api/routes/compute/test_dynamips_nodes.py
View File

@ -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

110
tests/api/routes/compute/test_iou_nodes.py
View File

@ -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

46
tests/api/routes/compute/test_nat_nodes.py
View File

@ -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

76
tests/api/routes/compute/test_projects.py
View File

@ -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

148
tests/api/routes/compute/test_qemu_nodes.py
View File

@ -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

60
tests/api/routes/compute/test_virtualbox_nodes.py
View File

@ -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

60
tests/api/routes/compute/test_vmware_nodes.py
View File

@ -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

74
tests/api/routes/compute/test_vpcs_nodes.py
View File

@ -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

21
tests/api/routes/controller/test_computes.py
View File

@ -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'

14
tests/conftest.py
View File

@ -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

11
tests/test_config.py
View File

@ -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):