Refactor tests and start work on database integration.

This commit is contained in:
grossmj committed 2020-12-02 18:39:08 +10:30
1 parent ae55c0ec9c
commit bf7cf862af
69 files changed
+2955 -1877

No files matched your search

+28 -25
View File
@@ -20,35 +20,38 @@ import sys
import pytest
import psutil
from fastapi import FastAPI, status
from httpx import AsyncClient
from gns3server.version import __version__
from gns3server.utils.path import get_default_project_directory
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
@pytest.mark.asyncio
async def test_get(compute_api, windows_platform):
response = await compute_api.get('/capabilities')
assert response.status_code == 200
assert response.json == {'node_types': ['cloud', 'ethernet_hub', 'ethernet_switch', 'nat', 'vpcs', 'virtualbox', 'dynamips', 'frame_relay_switch', 'atm_switch', 'qemu', 'vmware', 'traceng', 'docker', 'iou'],
'version': __version__,
'platform': sys.platform,
'cpus': psutil.cpu_count(logical=True),
'memory': psutil.virtual_memory().total,
'disk_size': psutil.disk_usage(get_default_project_directory()).total,
}
pytestmark = pytest.mark.asyncio
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
@pytest.mark.asyncio
async def test_get_on_gns3vm(compute_api, on_gns3vm):
async def test_get(app: FastAPI, client: AsyncClient, windows_platform) -> None:
response = await compute_api.get('/capabilities')
assert response.status_code == 200
assert response.json == {'node_types': ['cloud', 'ethernet_hub', 'ethernet_switch', 'nat', 'vpcs', 'virtualbox', 'dynamips', 'frame_relay_switch', 'atm_switch', 'qemu', 'vmware', 'traceng', 'docker', 'iou'],
'version': __version__,
'platform': sys.platform,
'cpus': psutil.cpu_count(logical=True),
'memory': psutil.virtual_memory().total,
'disk_size': psutil.disk_usage(get_default_project_directory()).total,
}
response = await client.get(app.url_path_for("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', 'traceng', 'docker', 'iou'],
'version': __version__,
'platform': sys.platform,
'cpus': psutil.cpu_count(logical=True),
'memory': psutil.virtual_memory().total,
'disk_size': psutil.disk_usage(get_default_project_directory()).total,
}
@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:
response = await client.get(app.url_path_for("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', 'traceng', 'docker', 'iou'],
'version': __version__,
'platform': sys.platform,
'cpus': psutil.cpu_count(logical=True),
'memory': psutil.virtual_memory().total,
'disk_size': psutil.disk_usage(get_default_project_directory()).total,
}
+87 -52
View File
@@ -17,99 +17,126 @@
import pytest
from unittest.mock import patch
from fastapi import FastAPI, status
from httpx import AsyncClient
from tests.utils import asyncio_patch
from gns3server.compute.project import Project
pytestmark = pytest.mark.asyncio
@pytest.fixture(scope="function")
@pytest.mark.asyncio
async def vm(compute_api, compute_project, on_gns3vm):
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, on_gns3vm) -> dict:
with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._start_ubridge"):
response = await compute_api.post("/projects/{project_id}/cloud/nodes".format(project_id=compute_project.id), {"name": "Cloud 1"})
assert response.status_code == 201
return response.json
response = await client.post(app.url_path_for("create_cloud", project_id=compute_project.id),
json={"name": "Cloud 1"})
assert response.status_code == status.HTTP_201_CREATED
return response.json()
@pytest.mark.asyncio
async def test_cloud_create(compute_api, compute_project):
async def test_cloud_create(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._start_ubridge"):
response = await compute_api.post("/projects/{project_id}/cloud/nodes".format(project_id=compute_project.id), {"name": "Cloud 1"})
response = await client.post(app.url_path_for("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
assert response.json()["name"] == "Cloud 1"
assert response.json()["project_id"] == compute_project.id
@pytest.mark.asyncio
async def test_cloud_get(compute_api, compute_project, vm):
async def test_get_cloud(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await compute_api.get("/projects/{project_id}/cloud/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 200
assert response.json["name"] == "Cloud 1"
assert response.json["project_id"] == compute_project.id
assert response.json["status"] == "started"
response = await client.get(app.url_path_for("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"
@pytest.mark.asyncio
async def test_cloud_nio_create_udp(compute_api, vm):
async def test_cloud_nio_create_udp(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
params = {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"}
response = await compute_api.post("/projects/{project_id}/cloud/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
assert response.json["type"] == "nio_udp"
url = app.url_path_for("create_cloud_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
response = await client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
@pytest.mark.asyncio
async def test_cloud_nio_update_udp(compute_api, vm):
async def test_cloud_nio_update_udp(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
params = {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"}
await compute_api.post("/projects/{project_id}/cloud/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
url = app.url_path_for("create_cloud_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
await client.post(url, json=params)
params["filters"] = {}
response = await compute_api.put("/projects/{project_id}/cloud/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201, response.body.decode()
assert response.json["type"] == "nio_udp"
url = app.url_path_for("create_cloud_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
response = await client.put(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
@pytest.mark.asyncio
async def test_cloud_delete_nio(compute_api, vm):
async def test_cloud_delete_nio(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
params = {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"}
await compute_api.post("/projects/{project_id}/cloud/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
url = app.url_path_for("create_cloud_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
await client.post(url, json=params)
url = app.url_path_for("delete_cloud_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._start_ubridge"):
response = await compute_api.delete("/projects/{project_id}/cloud/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204
response = await client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_cloud_delete(compute_api, vm):
async def test_cloud_delete(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await compute_api.delete("/projects/{project_id}/cloud/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204
response = await client.delete(app.url_path_for("delete_cloud", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_cloud_update(compute_api, vm):
async def test_cloud_update(app: FastAPI, client: AsyncClient, vm: dict) -> None:
response = await compute_api.put("/projects/{project_id}/cloud/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]), {"name": "test"})
assert response.status_code == 200
assert response.json["name"] == "test"
response = await client.put(app.url_path_for("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"
@pytest.mark.asyncio
async def test_cloud_start_capture(compute_api, vm):
async def test_cloud_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"capture_file_name": "test.pcap",
@@ -117,18 +144,26 @@ async def test_cloud_start_capture(compute_api, vm):
}
with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud.start_capture") as mock:
response = await compute_api.post("/projects/{project_id}/cloud/nodes/{node_id}/adapters/0/ports/0/capture/start".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 200
response = await client.post(app.url_path_for("start_cloud_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0"),
json=params)
assert response.status_code == status.HTTP_200_OK
assert mock.called
assert "test.pcap" in response.json["pcap_file_path"]
assert "test.pcap" in response.json()["pcap_file_path"]
@pytest.mark.asyncio
async def test_cloud_stop_capture(compute_api, vm):
async def test_cloud_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud.stop_capture") as mock:
response = await compute_api.post("/projects/{project_id}/cloud/nodes/{node_id}/adapters/0/ports/0/capture/stop".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204
response = await client.post(app.url_path_for("stop_cloud_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0"))
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
+21 -21
View File
@@ -18,34 +18,35 @@
import os
import pytest
from fastapi import FastAPI, status
from httpx import AsyncClient
from gns3server.version import __version__
from gns3server.compute.project import Project
pytestmark = pytest.mark.asyncio
@pytest.mark.asyncio
async def test_udp_allocation(compute_api, compute_project):
async def test_udp_allocation(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
response = await compute_api.post('/projects/{}/ports/udp'.format(compute_project.id), {})
assert response.status_code == 201
assert response.json['udp_port'] is not None
response = await client.post(app.url_path_for("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
# Netfifaces is not available on Travis
@pytest.mark.skipif(os.environ.get("TRAVIS", False) is not False, reason="Not supported on Travis")
@pytest.mark.asyncio
async def test_interfaces(compute_api):
async def test_interfaces(app: FastAPI, client: AsyncClient) -> None:
response = await compute_api.get('/network/interfaces')
assert response.status_code == 200
assert isinstance(response.json, list)
response = await client.get(app.url_path_for("network_interfaces"))
assert response.status_code == status.HTTP_200_OK
assert isinstance(response.json(), list)
@pytest.mark.asyncio
async def test_version_output(compute_api, config):
async def test_version_output(app: FastAPI, client: AsyncClient, config) -> None:
config.set("Server", "local", "true")
response = await compute_api.get('/version')
assert response.status_code == 200
assert response.json == {'local': True, 'version': __version__}
response = await client.get(app.url_path_for("compute_version"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == {'local': True, 'version': __version__}
# @pytest.mark.asyncio
@@ -55,8 +56,7 @@ async def test_version_output(compute_api, config):
# assert response.status_code == 200
@pytest.mark.asyncio
async def test_statistics_output(compute_api):
async def test_statistics_output(app: FastAPI, client: AsyncClient) -> None:
response = await compute_api.get('/statistics')
assert response.status_code == 200
response = await client.get(app.url_path_for("compute_statistics"))
assert response.status_code == status.HTTP_200_OK
+125 -83
View File
@@ -17,16 +17,20 @@
import pytest
import sys
import uuid
from fastapi import FastAPI, status
from httpx import AsyncClient
from tests.utils import asyncio_patch
from unittest.mock import patch
pytestmark = pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
from gns3server.compute.project import Project
pytestmark = [pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows"),
pytest.mark.asyncio]
@pytest.fixture
def base_params():
def base_params() -> dict:
"""Return standard parameters"""
params = {
@@ -53,90 +57,96 @@ def base_params():
@pytest.fixture
@pytest.mark.asyncio
async def vm(compute_api, compute_project, base_params):
async def vm(app: FastAPI, 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 compute_api.post("/projects/{project_id}/docker/nodes".format(project_id=compute_project.id), base_params)
assert response.status_code == 201
return response.json
response = await client.post(app.url_path_for("create_docker_node", project_id=compute_project.id),
json=base_params)
assert response.status_code == status.HTTP_201_CREATED
return response.json()
@pytest.mark.asyncio
async def test_docker_create(compute_api, compute_project, base_params):
async def test_docker_create(app: FastAPI, 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 compute_api.post("/projects/{project_id}/docker/nodes".format(project_id=compute_project.id), base_params)
assert response.status_code == 201
assert response.json["name"] == "PC TEST 1"
assert response.json["project_id"] == compute_project.id
assert response.json["container_id"] == "8bd8153ea8f5"
assert response.json["image"] == "nginx:latest"
assert response.json["adapters"] == 2
assert response.json["environment"] == "YES=1\nNO=0"
assert response.json["console_resolution"] == "1280x1024"
assert response.json["extra_hosts"] == "test:127.0.0.1"
response = await client.post(app.url_path_for("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"
assert response.json()["project_id"] == compute_project.id
assert response.json()["container_id"] == "8bd8153ea8f5"
assert response.json()["image"] == "nginx:latest"
assert response.json()["adapters"] == 2
assert response.json()["environment"] == "YES=1\nNO=0"
assert response.json()["console_resolution"] == "1280x1024"
assert response.json()["extra_hosts"] == "test:127.0.0.1"
@pytest.mark.asyncio
async def test_docker_start(compute_api, vm):
async def test_docker_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.start", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/docker/nodes/{node_id}/start".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("start_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_docker_stop(compute_api, vm):
async def test_docker_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.stop", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/docker/nodes/{node_id}/stop".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("stop_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_docker_reload(compute_api, vm):
async def test_docker_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.restart", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/docker/nodes/{node_id}/reload".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("reload_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_docker_delete(compute_api, vm):
async def test_docker_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.delete", return_value=True) as mock:
response = await compute_api.delete("/projects/{project_id}/docker/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.delete(app.url_path_for("delete_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_docker_pause(compute_api, vm):
async def test_docker_pause(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.pause", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/docker/nodes/{node_id}/pause".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("pause_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_docker_unpause(compute_api, vm):
async def test_docker_unpause(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.unpause", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/docker/nodes/{node_id}/unpause".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("unpause_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_docker_nio_create_udp(compute_api, vm):
async def test_docker_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -144,13 +154,17 @@ async def test_docker_nio_create_udp(compute_api, vm):
"rport": 4343,
"rhost": "127.0.0.1"}
response = await compute_api.post("/projects/{project_id}/docker/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
assert response.json["type"] == "nio_udp"
url = app.url_path_for("create_docker_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
response = await client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
@pytest.mark.asyncio
async def test_docker_update_nio(compute_api, vm):
async def test_docker_update_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -159,23 +173,37 @@ async def test_docker_update_nio(compute_api, vm):
"rhost": "127.0.0.1"
}
response = await compute_api.post("/projects/{project_id}/docker/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
url = app.url_path_for("create_docker_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
response = await client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
url = app.url_path_for("update_docker_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.adapter_update_nio_binding"):
response = await compute_api.put("/projects/{project_id}/docker/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201, response.body.decode()
response = await client.put(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
@pytest.mark.asyncio
async def test_docker_delete_nio(compute_api, vm):
async def test_docker_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("delete_docker_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.adapter_remove_nio_binding"):
response = await compute_api.delete("/projects/{project_id}/docker/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204
response = await client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_docker_update(compute_api, vm, free_console_port):
async def test_docker_update(app: FastAPI, client: AsyncClient, vm: dict, free_console_port: int) -> None:
params = {
"name": "test",
@@ -186,48 +214,62 @@ async def test_docker_update(compute_api, vm, free_console_port):
}
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.update") as mock:
response = await compute_api.put("/projects/{project_id}/docker/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
response = await client.put(app.url_path_for("update_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == 200
assert mock.called
assert response.json["name"] == "test"
assert response.json["console"] == free_console_port
assert response.json["start_command"] == "yes"
assert response.json["environment"] == "GNS3=1\nGNS4=0"
assert response.json["extra_hosts"] == "test:127.0.0.1"
assert response.json()["name"] == "test"
assert response.json()["console"] == free_console_port
assert response.json()["start_command"] == "yes"
assert response.json()["environment"] == "GNS3=1\nGNS4=0"
assert response.json()["extra_hosts"] == "test:127.0.0.1"
@pytest.mark.asyncio
async def test_docker_start_capture(compute_api, vm):
async def test_docker_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("start_docker_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
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 compute_api.post("/projects/{project_id}/docker/nodes/{node_id}/adapters/0/ports/0/capture/start".format(project_id=vm["project_id"], node_id=vm["node_id"]), body=params)
assert response.status_code == 200
response = await 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"]
assert "test.pcap" in response.json()["pcap_file_path"]
@pytest.mark.asyncio
async def test_docker_stop_capture(compute_api, vm):
async def test_docker_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("stop_docker_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
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 compute_api.post("/projects/{project_id}/docker/nodes/{node_id}/adapters/0/ports/0/capture/stop".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204
response = await client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
@pytest.mark.asyncio
async def test_docker_duplicate(compute_api, compute_project, base_params, vm):
async def test_docker_duplicate(app: FastAPI, 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 compute_api.post("/projects/{project_id}/docker/nodes".format(project_id=compute_project.id), base_params)
assert response.status_code == 201
response = await client.post(app.url_path_for("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 compute_api.post("/projects/{project_id}/docker/nodes/{node_id}/duplicate".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
params = {"destination_node_id": response.json()["node_id"]}
response = await client.post(app.url_path_for("duplicate_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_201_CREATED
+19 -21
View File
@@ -19,10 +19,12 @@ import pytest
import sys
import os
import stat
from unittest.mock import patch
from fastapi import FastAPI, status
from httpx import AsyncClient
from tests.utils import asyncio_patch
pytestmark = pytest.mark.asyncio
# @pytest.yield_fixture(scope="module")
# async def vm(compute_api, compute_project, fake_image):
@@ -139,7 +141,7 @@ from tests.utils import asyncio_patch
@pytest.fixture
def fake_image(tmpdir):
def fake_image(tmpdir) -> str:
"""Create a fake Dynamips image on disk"""
path = str(tmpdir / "7200.bin")
@@ -150,7 +152,7 @@ def fake_image(tmpdir):
@pytest.fixture
def fake_file(tmpdir):
def fake_file(tmpdir) -> str:
"""Create a fake file disk"""
path = str(tmpdir / "7200.txt")
@@ -160,24 +162,21 @@ def fake_file(tmpdir):
return path
@pytest.mark.asyncio
async def test_images(compute_api, tmpdir, fake_image, fake_file):
async def test_images(app: FastAPI, client: AsyncClient, tmpdir, fake_image: str, fake_file: str) -> None:
with patch("gns3server.utils.images.default_images_directory", return_value=str(tmpdir)):
response = await compute_api.get("/dynamips/images")
assert response.status_code == 200
assert response.json == [{"filename": "7200.bin",
"path": "7200.bin",
"filesize": 7,
"md5sum": "b0d5aa897d937aced5a6b1046e8f7e2e"
}]
response = await client.get(app.url_path_for("get_dynamips_images"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == [{"filename": "7200.bin",
"path": "7200.bin",
"filesize": 7,
"md5sum": "b0d5aa897d937aced5a6b1046e8f7e2e"}]
@pytest.mark.asyncio
async def test_upload_image(compute_api, images_dir):
async def test_upload_image(app: FastAPI, client: AsyncClient, images_dir: str) -> None:
response = await compute_api.post("/dynamips/images/test2", body=b"TEST", raw=True)
assert response.status_code == 204
response = await client.post(app.url_path_for("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:
assert f.read() == "TEST"
@@ -188,13 +187,12 @@ async def test_upload_image(compute_api, images_dir):
@pytest.mark.skipif(not sys.platform.startswith("win") and os.getuid() == 0, reason="Root can delete any image")
@pytest.mark.asyncio
async def test_upload_image_permission_denied(compute_api, images_dir):
async def test_upload_image_permission_denied(app: FastAPI, 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 compute_api.post("/dynamips/images/test2", body=b"TEST", raw=True)
assert response.status_code == 409
response = await client.post(app.url_path_for("upload_dynamips_image", filename="test2"), content=b"TEST")
assert response.status_code == status.HTTP_409_CONFLICT
+207 -144
View File
@@ -21,14 +21,19 @@ import stat
import sys
import uuid
from fastapi import FastAPI, status
from httpx import AsyncClient
from tests.utils import asyncio_patch
from unittest.mock import patch
pytestmark = pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
from gns3server.compute.project import Project
pytestmark = [pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows"),
pytest.mark.asyncio]
@pytest.fixture
def fake_iou_bin(images_dir):
def fake_iou_bin(images_dir) -> str:
"""Create a fake IOU image on disk"""
path = os.path.join(images_dir, "IOU", "iou.bin")
@@ -39,44 +44,44 @@ def fake_iou_bin(images_dir):
@pytest.fixture
def base_params(tmpdir, fake_iou_bin):
def base_params(tmpdir, fake_iou_bin) -> dict:
"""Return standard parameters"""
return {"application_id": 42, "name": "PC TEST 1", "path": "iou.bin"}
@pytest.fixture
@pytest.mark.asyncio
async def vm(compute_api, compute_project, base_params):
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> dict:
response = await compute_api.post("/projects/{project_id}/iou/nodes".format(project_id=compute_project.id), base_params)
assert response.status_code == 201
return response.json
response = await client.post(app.url_path_for("create_iou_node", project_id=compute_project.id), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
return response.json()
def startup_config_file(compute_project, vm):
def startup_config_file(compute_project: Project, vm: dict) -> str:
directory = os.path.join(compute_project.path, "project-files", "iou", vm["node_id"])
os.makedirs(directory, exist_ok=True)
return os.path.join(directory, "startup-config.cfg")
@pytest.mark.asyncio
async def test_iou_create(compute_api, compute_project, base_params):
async def test_iou_create(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> None:
response = await compute_api.post("/projects/{project_id}/iou/nodes".format(project_id=compute_project.id), base_params)
assert response.status_code == 201
assert response.json["name"] == "PC TEST 1"
assert response.json["project_id"] == compute_project.id
assert response.json["serial_adapters"] == 2
assert response.json["ethernet_adapters"] == 2
assert response.json["ram"] == 256
assert response.json["nvram"] == 128
assert response.json["l1_keepalives"] is False
response = await client.post(app.url_path_for("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
assert response.json()["serial_adapters"] == 2
assert response.json()["ethernet_adapters"] == 2
assert response.json()["ram"] == 256
assert response.json()["nvram"] == 128
assert response.json()["l1_keepalives"] is False
@pytest.mark.asyncio
async def test_iou_create_with_params(compute_api, compute_project, base_params):
async def test_iou_create_with_params(app: FastAPI,
client: AsyncClient,
compute_project: Project,
base_params: dict) -> None:
params = base_params
params["ram"] = 1024
@@ -87,23 +92,26 @@ async def test_iou_create_with_params(compute_api, compute_project, base_params)
params["startup_config_content"] = "hostname test"
params["use_default_iou_values"] = False
response = await compute_api.post("/projects/{project_id}/iou/nodes".format(project_id=compute_project.id), params)
assert response.status_code == 201
assert response.json["name"] == "PC TEST 1"
assert response.json["project_id"] == compute_project.id
assert response.json["serial_adapters"] == 4
assert response.json["ethernet_adapters"] == 0
assert response.json["ram"] == 1024
assert response.json["nvram"] == 512
assert response.json["l1_keepalives"] is True
assert response.json["use_default_iou_values"] is False
response = await client.post(app.url_path_for("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
assert response.json()["serial_adapters"] == 4
assert response.json()["ethernet_adapters"] == 0
assert response.json()["ram"] == 1024
assert response.json()["nvram"] == 512
assert response.json()["l1_keepalives"] is True
assert response.json()["use_default_iou_values"] is False
with open(startup_config_file(compute_project, response.json)) as f:
with open(startup_config_file(compute_project, response.json())) as f:
assert f.read() == "hostname test"
@pytest.mark.asyncio
async def test_iou_create_startup_config_already_exist(compute_api, compute_project, base_params):
async def test_iou_create_startup_config_already_exist(
app: FastAPI,
client: AsyncClient,
compute_project: Project,
base_params: dict) -> None:
"""We don't erase a startup-config if already exist at project creation"""
node_id = str(uuid.uuid4())
@@ -115,78 +123,78 @@ async def test_iou_create_startup_config_already_exist(compute_api, compute_proj
params["node_id"] = node_id
params["startup_config_content"] = "hostname test"
response = await compute_api.post("/projects/{project_id}/iou/nodes".format(project_id=compute_project.id), params)
assert response.status_code == 201
response = await client.post(app.url_path_for("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:
with open(startup_config_file(compute_project, response.json())) as f:
assert f.read() == "echo hello"
@pytest.mark.asyncio
async def test_iou_get(compute_api, compute_project, vm):
async def test_iou_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await compute_api.get("/projects/{project_id}/iou/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 200
assert response.json["name"] == "PC TEST 1"
assert response.json["project_id"] == compute_project.id
assert response.json["serial_adapters"] == 2
assert response.json["ethernet_adapters"] == 2
assert response.json["ram"] == 256
assert response.json["nvram"] == 128
assert response.json["l1_keepalives"] is False
response = await client.get(app.url_path_for("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
assert response.json()["serial_adapters"] == 2
assert response.json()["ethernet_adapters"] == 2
assert response.json()["ram"] == 256
assert response.json()["nvram"] == 128
assert response.json()["l1_keepalives"] is False
@pytest.mark.asyncio
async def test_iou_start(compute_api, vm):
async def test_iou_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.start", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/iou/nodes/{node_id}/start".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("start_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json={})
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_iou_start_with_iourc(compute_api, vm):
async def test_iou_start_with_iourc(app: FastAPI, 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 compute_api.post("/projects/{project_id}/iou/nodes/{node_id}/start".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
response = await client.post(app.url_path_for("start_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert mock.called
assert response.status_code == 204
response = await compute_api.get("/projects/{project_id}/iou/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 200
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_iou_stop(compute_api, vm):
async def test_iou_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.stop", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/iou/nodes/{node_id}/stop".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("stop_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_iou_reload(compute_api, vm):
async def test_iou_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.reload", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/iou/nodes/{node_id}/reload".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("reload_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_iou_delete(compute_api, vm):
async def test_iou_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.iou.IOU.delete_node", return_value=True) as mock:
response = await compute_api.delete("/projects/{project_id}/iou/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.delete(app.url_path_for("delete_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_iou_update(compute_api, vm, free_console_port):
async def test_iou_update(app: FastAPI, client: AsyncClient, vm: dict, free_console_port: int) -> None:
params = {
"name": "test",
@@ -199,91 +207,122 @@ async def test_iou_update(compute_api, vm, free_console_port):
"use_default_iou_values": True,
}
response = await compute_api.put("/projects/{project_id}/iou/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 200
assert response.json["name"] == "test"
assert response.json["console"] == free_console_port
assert response.json["ethernet_adapters"] == 4
assert response.json["serial_adapters"] == 0
assert response.json["ram"] == 512
assert response.json["nvram"] == 2048
assert response.json["l1_keepalives"] is True
assert response.json["use_default_iou_values"] is True
response = await client.put(app.url_path_for("update_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "test"
assert response.json()["console"] == free_console_port
assert response.json()["ethernet_adapters"] == 4
assert response.json()["serial_adapters"] == 0
assert response.json()["ram"] == 512
assert response.json()["nvram"] == 2048
assert response.json()["l1_keepalives"] is True
assert response.json()["use_default_iou_values"] is True
@pytest.mark.asyncio
async def test_iou_nio_create_udp(compute_api, vm):
async def test_iou_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"}
response = await compute_api.post("/projects/{project_id}/iou/nodes/{node_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
assert response.json["type"] == "nio_udp"
url = app.url_path_for("create_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
response = await client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
@pytest.mark.asyncio
async def test_iou_nio_update_udp(compute_api, vm):
async def test_iou_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"}
await compute_api.post("/projects/{project_id}/iou/nodes/{node_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
url = app.url_path_for("create_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
await client.post(url, json=params)
params["filters"] = {}
response = await compute_api.put("/projects/{project_id}/iou/nodes/{node_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201, response.body.decode()
assert response.json["type"] == "nio_udp"
url = app.url_path_for("update_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
response = await client.put(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
@pytest.mark.asyncio
async def test_iou_nio_create_ethernet(compute_api, vm, ethernet_device):
async def test_iou_nio_create_ethernet(app: FastAPI, client: AsyncClient, vm: dict, ethernet_device: str) -> None:
params = {
"type": "nio_ethernet",
"ethernet_device": ethernet_device
}
response = await compute_api.post("/projects/{project_id}/iou/nodes/{node_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
assert response.json["type"] == "nio_ethernet"
assert response.json["ethernet_device"] == ethernet_device
url = app.url_path_for("create_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
response = await 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
@pytest.mark.asyncio
async def test_iou_nio_create_ethernet_different_port(compute_api, vm, ethernet_device):
async def test_iou_nio_create_ethernet_different_port(app: FastAPI,
client: AsyncClient,
vm: dict,
ethernet_device: str) -> None:
params = {
"type": "nio_ethernet",
"ethernet_device": ethernet_device
}
response = await compute_api.post("/projects/{project_id}/iou/nodes/{node_id}/adapters/0/ports/3/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
assert response.json["type"] == "nio_ethernet"
assert response.json["ethernet_device"] == ethernet_device
url = app.url_path_for("create_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="3")
response = await 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
@pytest.mark.asyncio
async def test_iou_nio_create_tap(compute_api, vm, ethernet_device):
async def test_iou_nio_create_tap(app: FastAPI, client: AsyncClient, vm: dict, ethernet_device: str) -> None:
params = {
"type": "nio_tap",
"tap_device": ethernet_device
}
url = app.url_path_for("create_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
with patch("gns3server.compute.base_manager.BaseManager.has_privileged_access", return_value=True):
response = await compute_api.post("/projects/{project_id}/iou/nodes/{node_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
assert response.json["type"] == "nio_tap"
response = await client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_tap"
@pytest.mark.asyncio
async def test_iou_delete_nio(compute_api, vm):
async def test_iou_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -292,33 +331,57 @@ async def test_iou_delete_nio(compute_api, vm):
"rhost": "127.0.0.1"
}
await compute_api.post("/projects/{project_id}/iou/nodes/{node_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
response = await compute_api.delete("/projects/{project_id}/iou/nodes/{node_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204
url = app.url_path_for("create_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
await client.post(url, json=params)
url = app.url_path_for("delete_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
response = await client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_iou_start_capture(compute_api, vm):
async def test_iou_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"capture_file_name": "test.pcap",
"data_link_type": "DLT_EN10MB"
}
url = app.url_path_for("start_iou_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
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 compute_api.post("/projects/{project_id}/iou/nodes/{node_id}/adapters/0/ports/0/capture/start".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 200
response = await 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"]
assert "test.pcap" in response.json()["pcap_file_path"]
@pytest.mark.asyncio
async def test_iou_stop_capture(compute_api, vm):
async def test_iou_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("stop_iou_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
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 compute_api.post("/projects/{project_id}/iou/nodes/{node_id}/adapters/0/ports/0/capture/stop".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204
response = await client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
@@ -327,24 +390,22 @@ async def test_iou_stop_capture(compute_api, vm):
#
# with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.get_nio"):
# with asyncio_patch("gns3server.compute.iou.IOU.stream_pcap_file"):
# response = await compute_api.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 == 200
# 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)
# assert response.status_code == status.HTTP_200_OK
@pytest.mark.asyncio
async def test_images(compute_api, fake_iou_bin):
async def test_images(app: FastAPI, client: AsyncClient, fake_iou_bin: str) -> None:
response = await compute_api.get("/iou/images")
assert response.status_code == 200
assert response.json == [{"filename": "iou.bin", "path": "iou.bin", "filesize": 7, "md5sum": "e573e8f5c93c6c00783f20c7a170aa6c"}]
response = await client.get(app.url_path_for("get_iou_images"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == [{"filename": "iou.bin", "path": "iou.bin", "filesize": 7, "md5sum": "e573e8f5c93c6c00783f20c7a170aa6c"}]
@pytest.mark.asyncio
async def test_image_vm(compute_api, tmpdir):
async def test_image_vm(app: FastAPI, client: AsyncClient, tmpdir) -> None:
with patch("gns3server.compute.IOU.get_images_directory", return_value=str(tmpdir)):
response = await compute_api.post("/iou/images/test2", body="TEST", raw=True)
assert response.status_code == 204
response = await client.post(app.url_path_for("download_iou_image", filename="test2"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
with open(str(tmpdir / "test2")) as f:
assert f.read() == "TEST"
@@ -354,13 +415,15 @@ async def test_image_vm(compute_api, tmpdir):
assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf"
@pytest.mark.asyncio
async def test_iou_duplicate(compute_api, compute_project, vm, base_params):
async def test_iou_duplicate(app: FastAPI, client: AsyncClient, vm: dict, base_params: dict) -> None:
# create destination node first
response = await compute_api.post("/projects/{project_id}/iou/nodes".format(project_id=compute_project.id), base_params)
assert response.status_code == 201
response = await client.post(app.url_path_for("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 compute_api.post("/projects/{project_id}/iou/nodes/{node_id}/duplicate".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
params = {"destination_node_id": response.json()["node_id"]}
response = await client.post(app.url_path_for("duplicate_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_201_CREATED
+97 -54
View File
@@ -17,41 +17,45 @@
import pytest
from fastapi import FastAPI, status
from httpx import AsyncClient
from tests.utils import asyncio_patch
from gns3server.compute.project import Project
pytestmark = pytest.mark.asyncio
@pytest.fixture(scope="function")
@pytest.mark.asyncio
async def vm(compute_api, compute_project, ubridge_path, on_gns3vm):
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, ubridge_path: str, on_gns3vm) -> dict:
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat._start_ubridge"):
response = await compute_api.post("/projects/{project_id}/nat/nodes".format(project_id=compute_project.id), {"name": "Nat 1"})
assert response.status_code == 201
return response.json
response = await client.post(app.url_path_for("create_nat_node", project_id=compute_project.id),
json={"name": "Nat 1"})
assert response.status_code == status.HTTP_201_CREATED
return response.json()
@pytest.mark.asyncio
async def test_nat_create(compute_api, compute_project, on_gns3vm):
async def test_nat_create(app: FastAPI, client: AsyncClient, compute_project: Project, on_gns3vm) -> None:
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat._start_ubridge"):
response = await compute_api.post("/projects/{project_id}/nat/nodes".format(project_id=compute_project.id), {"name": "Nat 1"})
assert response.status_code == 201
assert response.json["name"] == "Nat 1"
assert response.json["project_id"] == compute_project.id
response = await client.post(app.url_path_for("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
@pytest.mark.asyncio
async def test_nat_get(compute_api, compute_project, vm):
async def test_nat_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await compute_api.get("/projects/{project_id}/nat/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 200
assert response.json["name"] == "Nat 1"
assert response.json["project_id"] == compute_project.id
assert response.json["status"] == "started"
response = await client.get(app.url_path_for("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"
@pytest.mark.asyncio
async def test_nat_nio_create_udp(compute_api, vm):
async def test_nat_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -60,14 +64,19 @@ async def test_nat_nio_create_udp(compute_api, vm):
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_nat_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.add_nio"):
response = await compute_api.post("/projects/{project_id}/nat/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
assert response.json["type"] == "nio_udp"
response = await client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
@pytest.mark.asyncio
async def test_nat_nio_update_udp(compute_api, vm):
async def test_nat_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -76,15 +85,26 @@ async def test_nat_nio_update_udp(compute_api, vm):
"rhost": "127.0.0.1"
}
await compute_api.post("/projects/{project_id}/nat/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
url = app.url_path_for("create_nat_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
await client.post(url, json=params)
params["filters"] = {}
response = await compute_api.put("/projects/{project_id}/nat/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201, response.body.decode()
assert response.json["type"] == "nio_udp"
url = app.url_path_for("update_nat_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
response = await client.put(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
@pytest.mark.asyncio
async def test_nat_delete_nio(compute_api, vm):
async def test_nat_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -93,50 +113,73 @@ async def test_nat_delete_nio(compute_api, vm):
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_nat_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.add_nio"):
await compute_api.post("/projects/{project_id}/nat/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
await client.post(url, json=params)
url = app.url_path_for("delete_nat_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.remove_nio") as mock:
response = await compute_api.delete("/projects/{project_id}/nat/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.delete(url)
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_nat_delete(compute_api, vm):
async def test_nat_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None:
response = await compute_api.delete("/projects/{project_id}/nat/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204
response = await client.delete(app.url_path_for("delete_nat_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_nat_update(compute_api, vm):
async def test_nat_update(app: FastAPI, client: AsyncClient, vm: dict) -> None:
response = await compute_api.put("/projects/{project_id}/nat/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]), {"name": "test"})
assert response.status_code == 200
assert response.json["name"] == "test"
response = await client.put(app.url_path_for("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"
@pytest.mark.asyncio
async def test_nat_start_capture(compute_api, vm):
async def test_nat_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"capture_file_name": "test.pcap",
"data_link_type": "DLT_EN10MB"
}
url = app.url_path_for("start_nat_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.start_capture") as mock:
response = await compute_api.post("/projects/{project_id}/nat/nodes/{node_id}/adapters/0/ports/0/capture/start".format(project_id=vm["project_id"], node_id=vm["node_id"]), body=params)
assert response.status_code == 200
response = await 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"]
assert "test.pcap" in response.json()["pcap_file_path"]
@pytest.mark.asyncio
async def test_nat_stop_capture(compute_api, vm):
async def test_nat_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("stop_nat_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.stop_capture") as mock:
response = await compute_api.post("/projects/{project_id}/nat/nodes/{node_id}/adapters/0/ports/0/capture/stop".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204
response = await client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
@@ -145,5 +188,5 @@ async def test_nat_stop_capture(compute_api, vm):
#
# with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.get_nio"):
# with asyncio_patch("gns3server.compute.builtin.Builtin.stream_pcap_file"):
# response = await compute_api.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 == 200
# 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)
# assert response.status_code == status.HTTP_200_OK
@@ -21,11 +21,11 @@ import json
from gns3server.compute.notification_manager import NotificationManager
@pytest.mark.asyncio
async def test_notification_ws(compute_api):
# FIXME: how to test websockets
pass
# @pytest.mark.asyncio
# async def test_notification_ws(compute_api):
#
# # FIXME: how to test websockets
# pass
#with compute_api.ws("/notifications/ws") as ws:
+82 -100
View File
@@ -21,12 +21,17 @@ import os
from unittest.mock import patch
from tests.utils import asyncio_patch
from fastapi import FastAPI, status
from httpx import AsyncClient
from gns3server.compute.project_manager import ProjectManager
from gns3server.compute.project import Project
pytestmark = pytest.mark.asyncio
@pytest.fixture
def base_params(tmpdir):
def base_params(tmpdir) -> dict:
"""Return standard parameters"""
params = {
@@ -37,109 +42,103 @@ def base_params(tmpdir):
return params
@pytest.mark.asyncio
async def test_create_project_with_path(compute_api, base_params):
async def test_create_project_with_path(app: FastAPI, client: AsyncClient, base_params: dict) -> None:
with patch("gns3server.compute.project.Project.is_local", return_value=True):
response = await compute_api.post("/projects", base_params)
assert response.status_code == 201
assert response.json["project_id"] == base_params["project_id"]
response = await client.post(app.url_path_for("create_compute_project"), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["project_id"] == base_params["project_id"]
@pytest.mark.asyncio
async def test_create_project_with_path_and_empty_variables(compute_api, base_params):
async def test_create_project_with_path_and_empty_variables(app: FastAPI,
client: AsyncClient,
base_params: dict) -> None:
base_params["variables"] = None
with patch("gns3server.compute.project.Project.is_local", return_value=True):
response = await compute_api.post("/projects", base_params)
assert response.status_code == 201
assert response.json["project_id"] == base_params["project_id"]
response = await client.post(app.url_path_for("create_compute_project"), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["project_id"] == base_params["project_id"]
@pytest.mark.asyncio
async def test_create_project_without_dir(compute_api, base_params):
async def test_create_project_without_dir(app: FastAPI, client: AsyncClient, base_params: dict) -> None:
del base_params["path"]
response = await compute_api.post("/projects", base_params)
assert response.status_code == 201
assert response.json["project_id"] == base_params["project_id"]
assert response.json["name"] == base_params["name"]
response = await client.post(app.url_path_for("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"]
@pytest.mark.asyncio
async def test_show_project(compute_api, base_params):
async def test_show_project(app: FastAPI, client: AsyncClient, base_params: dict) -> None:
response = await compute_api.post("/projects", base_params)
assert response.status_code == 201
response = await compute_api.get("/projects/{project_id}".format(project_id=base_params["project_id"]))
response = await client.post(app.url_path_for("create_compute_project"), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
response = await client.get(app.url_path_for("get_compute_project", project_id=base_params["project_id"]))
#print(response.json.keys())
#assert len(response.json.keys()) == 3
assert response.json["project_id"] == base_params["project_id"]
assert response.json["name"] == base_params["name"]
assert response.json["variables"] is None
#print(response.json().keys())
#assert len(response.json().keys()) == 3
assert response.json()["project_id"] == base_params["project_id"]
assert response.json()["name"] == base_params["name"]
assert response.json()["variables"] is None
@pytest.mark.asyncio
async def test_show_project_invalid_uuid(compute_api):
async def test_show_project_invalid_uuid(app: FastAPI, client: AsyncClient) -> None:
response = await compute_api.get("/projects/50010203-0405-0607-0809-0a0b0c0d0e42")
assert response.status_code == 404
response = await client.get(app.url_path_for("get_compute_project",
project_id="50010203-0405-0607-0809-0a0b0c0d0e42"))
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.asyncio
async def test_list_projects(compute_api):
async def test_list_projects(app: FastAPI, client: AsyncClient) -> dict:
ProjectManager.instance()._projects = {}
params = {"name": "test", "project_id": "51010203-0405-0607-0809-0a0b0c0d0e0f"}
response = await compute_api.post("/projects", params)
assert response.status_code == 201
response = await client.post(app.url_path_for("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 compute_api.post("/projects", params)
assert response.status_code == 201
response = await client.post(app.url_path_for("create_compute_project"), json=params)
assert response.status_code == status.HTTP_201_CREATED
response = await compute_api.get("/projects")
assert response.status_code == 200
assert len(response.json) == 2
assert "51010203-0405-0607-0809-0a0b0c0d0e0f" in [p["project_id"] for p in response.json]
response = await client.get(app.url_path_for("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()]
@pytest.mark.asyncio
async def test_delete_project(compute_api, compute_project):
async def test_delete_project(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
with asyncio_patch("gns3server.compute.project.Project.delete", return_value=True) as mock:
response = await compute_api.delete("/projects/{project_id}".format(project_id=compute_project.id))
assert response.status_code == 204
response = await client.delete(app.url_path_for("delete_compute_project", project_id=compute_project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
@pytest.mark.asyncio
async def test_update_project(compute_api, base_params):
async def test_update_project(app: FastAPI, client: AsyncClient, base_params: dict) -> None:
response = await compute_api.post("/projects", base_params)
assert response.status_code == 201
response = await client.post(app.url_path_for("create_compute_project"), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
params = {"variables": [{"name": "TEST1", "value": "VAL1"}]}
response = await compute_api.put("/projects/{project_id}".format(project_id=base_params["project_id"]), params)
assert response.status_code == 200
assert response.json["variables"] == [{"name": "TEST1", "value": "VAL1"}]
response = await client.put(app.url_path_for("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"}]
@pytest.mark.asyncio
async def test_delete_project_invalid_uuid(compute_api):
async def test_delete_project_invalid_uuid(app: FastAPI, client: AsyncClient) -> None:
response = await compute_api.delete("/projects/{project_id}".format(project_id=uuid.uuid4()))
assert response.status_code == 404
response = await client.delete(app.url_path_for("delete_compute_project", project_id=str(uuid.uuid4())))
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.asyncio
async def test_close_project(compute_api, compute_project):
async def test_close_project(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
with asyncio_patch("gns3server.compute.project.Project.close", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/close".format(project_id=compute_project.id))
assert response.status_code == 204
response = await client.post(app.url_path_for("close_compute_project", project_id=compute_project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
@@ -148,20 +147,18 @@ async def test_close_project(compute_api, compute_project):
#
# ProjectHandler._notifications_listening = {compute_project.id: 2}
# with asyncio_patch("gns3server.compute.project.Project.close", return_value=True) as mock:
# response = await compute_api.post("/projects/{project_id}/close".format(project_id=compute_project.id))
# assert response.status_code == 204
# response = await 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
@pytest.mark.asyncio
async def test_close_project_invalid_uuid(compute_api):
async def test_close_project_invalid_uuid(app: FastAPI, client: AsyncClient) -> None:
response = await compute_api.post("/projects/{project_id}/close".format(project_id=uuid.uuid4()))
assert response.status_code == 404
response = await client.post(app.url_path_for("close_compute_project", project_id=str(uuid.uuid4())))
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.asyncio
async def test_get_file(compute_api, tmpdir):
async def test_get_file(app: FastAPI, client: AsyncClient, tmpdir) -> None:
with patch("gns3server.config.Config.get_section_config", return_value={"projects_path": str(tmpdir)}):
project = ProjectManager.instance().create_project(project_id="01010203-0405-0607-0809-0a0b0c0d0e0b")
@@ -169,48 +166,33 @@ async def test_get_file(compute_api, tmpdir):
with open(os.path.join(project.path, "hello"), "w+") as f:
f.write("world")
response = await compute_api.get("/projects/{project_id}/files/hello".format(project_id=project.id), raw=True)
assert response.status_code == 200
response = await client.get(app.url_path_for("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 compute_api.get("/projects/{project_id}/files/false".format(project_id=project.id), raw=True)
assert response.status_code == 404
response = await client.get(app.url_path_for("get_compute_project_file", project_id=project.id, file_path="false"))
assert response.status_code == status.HTTP_404_NOT_FOUND
response = await compute_api.get("/projects/{project_id}/files/../hello".format(project_id=project.id), raw=True)
assert response.status_code == 404
response = await client.get(app.url_path_for("get_compute_project_file",
project_id=project.id,
file_path="../hello"))
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.asyncio
async def test_write_file(compute_api, tmpdir):
async def test_write_file(app: FastAPI, client: AsyncClient, tmpdir) -> None:
with patch("gns3server.config.Config.get_section_config", return_value={"projects_path": str(tmpdir)}):
project = ProjectManager.instance().create_project(project_id="01010203-0405-0607-0809-0a0b0c0d0e0b")
response = await compute_api.post("/projects/{project_id}/files/hello".format(project_id=project.id), body="world", raw=True)
assert response.status_code == 204
response = await client.post(app.url_path_for("write_compute_project_file",
project_id=project.id,
file_path="hello"), content=b"world")
assert response.status_code == status.HTTP_204_NO_CONTENT
with open(os.path.join(project.path, "hello")) as f:
assert f.read() == "world"
response = await compute_api.post("/projects/{project_id}/files/../hello".format(project_id=project.id), raw=True)
assert response.status_code == 404
@pytest.mark.asyncio
async def test_stream_file(compute_api, tmpdir):
with patch("gns3server.config.Config.get_section_config", return_value={"projects_path": str(tmpdir)}):
project = ProjectManager.instance().create_project(project_id="01010203-0405-0607-0809-0a0b0c0d0e0b")
with open(os.path.join(project.path, "hello"), "w+") as f:
f.write("world")
response = await compute_api.get("/projects/{project_id}/files/hello".format(project_id=project.id), raw=True)
assert response.status_code == 200
assert response.content == b"world"
response = await compute_api.get("/projects/{project_id}/files/false".format(project_id=project.id), raw=True)
assert response.status_code == 404
response = await compute_api.get("/projects/{project_id}/files/../hello".format(project_id=project.id), raw=True)
assert response.status_code == 404
response = await client.post(app.url_path_for("write_compute_project_file",
project_id=project.id,
file_path="../hello"))
assert response.status_code == status.HTTP_404_NOT_FOUND
+250 -171
View File
@@ -16,16 +16,22 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import pytest
import uuid
import os
import sys
import stat
from fastapi import FastAPI, status
from httpx import AsyncClient
from tests.utils import asyncio_patch
from unittest.mock import patch
from gns3server.compute.project import Project
pytestmark = pytest.mark.asyncio
@pytest.fixture
def fake_qemu_bin(monkeypatch, tmpdir):
def fake_qemu_bin(monkeypatch, tmpdir) -> str:
monkeypatch.setenv("PATH", str(tmpdir))
if sys.platform.startswith("win"):
@@ -40,7 +46,7 @@ def fake_qemu_bin(monkeypatch, tmpdir):
@pytest.fixture
def fake_qemu_vm(images_dir):
def fake_qemu_vm(images_dir) -> str:
img_dir = os.path.join(images_dir, "QEMU")
bin_path = os.path.join(img_dir, "linux载.img")
@@ -51,140 +57,165 @@ def fake_qemu_vm(images_dir):
@pytest.fixture
def base_params(tmpdir, fake_qemu_bin):
def base_params(tmpdir, fake_qemu_bin) -> dict:
"""Return standard parameters"""
return {"name": "PC TEST 1", "qemu_path": fake_qemu_bin}
@pytest.fixture
@pytest.mark.asyncio
async def vm(compute_api, compute_project, base_params):
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> None:
response = await compute_api.post("/projects/{project_id}/qemu/nodes".format(project_id=compute_project.id), base_params)
assert response.status_code == 201
return response.json
response = await client.post(app.url_path_for("create_qemu_node", project_id=compute_project.id), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
return response.json()
@pytest.mark.asyncio
async def test_qemu_create(compute_api, compute_project, base_params, fake_qemu_bin):
async def test_qemu_create(app: FastAPI,
client: AsyncClient,
compute_project: Project,
base_params: dict,
fake_qemu_bin: str) -> None:
response = await compute_api.post("/projects/{project_id}/qemu/nodes".format(project_id=compute_project.id), base_params)
assert response.status_code == 201
assert response.json["name"] == "PC TEST 1"
assert response.json["project_id"] == compute_project.id
assert response.json["qemu_path"] == fake_qemu_bin
assert response.json["platform"] == "x86_64"
response = await client.post(app.url_path_for("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
assert response.json()["qemu_path"] == fake_qemu_bin
assert response.json()["platform"] == "x86_64"
@pytest.mark.asyncio
async def test_qemu_create_platform(compute_api, compute_project, base_params, fake_qemu_bin):
async def test_qemu_create_platform(app: FastAPI,
client: AsyncClient,
compute_project: Project,
base_params: dict,
fake_qemu_bin: str):
base_params["qemu_path"] = None
base_params["platform"] = "x86_64"
response = await compute_api.post("/projects/{project_id}/qemu/nodes".format(project_id=compute_project.id), base_params)
assert response.status_code == 201
assert response.json["name"] == "PC TEST 1"
assert response.json["project_id"] == compute_project.id
assert response.json["qemu_path"] == fake_qemu_bin
assert response.json["platform"] == "x86_64"
response = await client.post(app.url_path_for("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
assert response.json()["qemu_path"] == fake_qemu_bin
assert response.json()["platform"] == "x86_64"
@pytest.mark.asyncio
async def test_qemu_create_with_params(compute_api, compute_project, base_params, fake_qemu_vm):
async def test_qemu_create_with_params(app: FastAPI,
client: AsyncClient,
compute_project: Project,
base_params: dict,
fake_qemu_vm: str):
params = base_params
params["ram"] = 1024
params["hda_disk_image"] = "linux载.img"
response = await compute_api.post("/projects/{project_id}/qemu/nodes".format(project_id=compute_project.id), params)
assert response.status_code == 201
assert response.json["name"] == "PC TEST 1"
assert response.json["project_id"] == compute_project.id
assert response.json["ram"] == 1024
assert response.json["hda_disk_image"] == "linux载.img"
assert response.json["hda_disk_image_md5sum"] == "c4ca4238a0b923820dcc509a6f75849b"
response = await client.post(app.url_path_for("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
assert response.json()["ram"] == 1024
assert response.json()["hda_disk_image"] == "linux载.img"
assert response.json()["hda_disk_image_md5sum"] == "c4ca4238a0b923820dcc509a6f75849b"
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
@pytest.mark.asyncio
async def test_qemu_create_with_project_file(compute_api, compute_project, base_params, fake_qemu_vm):
async def test_qemu_create_with_project_file(app: FastAPI,
client: AsyncClient,
compute_project: Project,
base_params: dict,
fake_qemu_vm: str) -> None:
response = await compute_api.post("/projects/{project_id}/files/hello.img".format(project_id=compute_project.id), body=b"world", raw=True)
assert response.status_code == 204
response = await client.post(app.url_path_for("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 compute_api.post("/projects/{project_id}/qemu/nodes".format(project_id=compute_project.id), params)
assert response.status_code == 201
assert response.json["hda_disk_image"] == "hello.img"
assert response.json["hda_disk_image_md5sum"] == "7d793037a0760186574b0282f2f435e7"
response = await client.post(app.url_path_for("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"
@pytest.mark.asyncio
async def test_qemu_get(compute_api, compute_project, vm):
async def test_qemu_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict):
response = await compute_api.get("/projects/{project_id}/qemu/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 200
assert response.json["name"] == "PC TEST 1"
assert response.json["project_id"] == compute_project.id
assert response.json["node_directory"] == os.path.join(compute_project.path, "project-files", "qemu", vm["node_id"])
response = await client.get(app.url_path_for("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
assert response.json()["node_directory"] == os.path.join(compute_project.path,
"project-files",
"qemu",
vm["node_id"])
@pytest.mark.asyncio
async def test_qemu_start(compute_api, vm):
async def test_qemu_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.start", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/qemu/nodes/{node_id}/start".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("start_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_qemu_stop(compute_api, vm):
async def test_qemu_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.stop", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/qemu/nodes/{node_id}/stop".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("stop_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_qemu_reload(compute_api, vm):
async def test_qemu_reload(app: FastAPI, client: AsyncClient, vm) -> None:
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.reload", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/qemu/nodes/{node_id}/reload".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("reload_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_qemu_suspend(compute_api, vm):
async def test_qemu_suspend(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.suspend", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/qemu/nodes/{node_id}/suspend".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("suspend_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_qemu_resume(compute_api, vm):
async def test_qemu_resume(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.resume", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/qemu/nodes/{node_id}/resume".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("resume_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_qemu_delete(compute_api, vm):
async def test_qemu_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.qemu.Qemu.delete_node", return_value=True) as mock:
response = await compute_api.delete("/projects/{project_id}/qemu/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.delete(app.url_path_for("delete_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_qemu_update(compute_api, vm, free_console_port, fake_qemu_vm):
async def test_qemu_update(app: FastAPI,
client: AsyncClient,
vm: dict,
free_console_port: int,
fake_qemu_vm: str) -> None:
params = {
"name": "test",
@@ -193,16 +224,17 @@ async def test_qemu_update(compute_api, vm, free_console_port, fake_qemu_vm):
"hdb_disk_image": "linux载.img"
}
response = await compute_api.put("/projects/{project_id}/qemu/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 200
assert response.json["name"] == "test"
assert response.json["console"] == free_console_port
assert response.json["hdb_disk_image"] == "linux载.img"
assert response.json["ram"] == 1024
response = await client.put(app.url_path_for("update_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "test"
assert response.json()["console"] == free_console_port
assert response.json()["hdb_disk_image"] == "linux载.img"
assert response.json()["ram"] == 1024
@pytest.mark.asyncio
async def test_qemu_nio_create_udp(compute_api, vm):
async def test_qemu_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -212,14 +244,21 @@ async def test_qemu_nio_create_udp(compute_api, vm):
}
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.add_ubridge_udp_connection"):
await compute_api.put("/projects/{project_id}/qemu/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]), {"adapters": 2})
response = await compute_api.post("/projects/{project_id}/qemu/nodes/{node_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
assert response.json["type"] == "nio_udp"
await client.put(app.url_path_for("update_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json={"adapters": 2})
url = app.url_path_for("create_qemu_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
response = await client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
@pytest.mark.asyncio
async def test_qemu_nio_update_udp(compute_api, vm):
async def test_qemu_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -228,17 +267,31 @@ async def test_qemu_nio_update_udp(compute_api, vm):
"rhost": "127.0.0.1"
}
await compute_api.put("/projects/{project_id}/qemu/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]), {"adapters": 2})
await compute_api.post("/projects/{project_id}/qemu/nodes/{node_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
await client.put(app.url_path_for("update_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json={"adapters": 2})
url = app.url_path_for("create_qemu_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
await client.post(url, json=params)
params["filters"] = {}
response = await compute_api.put("/projects/{project_id}/qemu/nodes/{node_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201, response.body.decode()
assert response.json["type"] == "nio_udp"
url = app.url_path_for("update_qemu_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
response = await client.put(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
@pytest.mark.asyncio
async def test_qemu_delete_nio(compute_api, vm):
async def test_qemu_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -248,57 +301,69 @@ async def test_qemu_delete_nio(compute_api, vm):
}
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM._ubridge_send"):
await compute_api.put("/projects/{project_id}/qemu/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]), {"adapters": 2})
await compute_api.post("/projects/{project_id}/qemu/nodes/{node_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
response = await compute_api.delete("/projects/{project_id}/qemu/nodes/{node_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204
await client.put(app.url_path_for("update_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json={"adapters": 2})
url = app.url_path_for("create_qemu_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
await client.post(url, json=params)
url = app.url_path_for("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)
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_qemu_list_binaries(compute_api, vm):
async def test_qemu_list_binaries(app: FastAPI, 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 compute_api.get("/qemu/binaries".format(project_id=vm["project_id"]))
response = await client.get(app.url_path_for("get_qemu_binaries"))
assert mock.called_with(None)
assert response.status_code == 200
assert response.json == ret
assert response.status_code == status.HTTP_200_OK
assert response.json() == ret
@pytest.mark.asyncio
async def test_qemu_list_binaries_filter(compute_api, vm):
ret = [
{"path": "/tmp/x86_64", "version": "2.2.0"},
{"path": "/tmp/alpha", "version": "2.1.0"},
{"path": "/tmp/i386", "version": "2.1.0"}
]
with asyncio_patch("gns3server.compute.qemu.Qemu.binary_list", return_value=ret) as mock:
response = await compute_api.get("/qemu/binaries".format(project_id=vm["project_id"]), body={"archs": ["i386"]})
assert response.status_code == 200
assert mock.called_with(["i386"])
assert response.json == ret
# async def test_qemu_list_binaries_filter(app: FastAPI, client: AsyncClient, vm: dict) -> None:
#
# ret = [
# {"path": "/tmp/x86_64", "version": "2.2.0"},
# {"path": "/tmp/alpha", "version": "2.1.0"},
# {"path": "/tmp/i386", "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("get_qemu_binaries"),
# json={"archs": ["i386"]})
# assert response.status_code == status.HTTP_200_OK
# assert mock.called_with(["i386"])
# assert response.json() == ret
@pytest.mark.asyncio
async def test_images(compute_api, fake_qemu_vm):
async def test_images(app: FastAPI, client: AsyncClient, fake_qemu_vm) -> None:
response = await compute_api.get("/qemu/images")
assert response.status_code == 200
assert {"filename": "linux载.img", "path": "linux载.img", "md5sum": "c4ca4238a0b923820dcc509a6f75849b", "filesize": 1} in response.json
response = await client.get(app.url_path_for("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()
@pytest.mark.asyncio
async def test_upload_image(compute_api, tmpdir):
async def test_upload_image(app: FastAPI, client: AsyncClient, tmpdir: str) -> None:
with patch("gns3server.compute.Qemu.get_images_directory", return_value=str(tmpdir)):
response = await compute_api.post("/qemu/images/test2使", body="TEST", raw=True)
assert response.status_code == 204
print(os.listdir(tmpdir))
response = await client.post(app.url_path_for("upload_qemu_image",
filename="test2使"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
with open(str(tmpdir / "test2使")) as f:
assert f.read() == "TEST"
@@ -307,12 +372,13 @@ async def test_upload_image(compute_api, tmpdir):
assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf"
@pytest.mark.asyncio
async def test_upload_image_ova(compute_api, tmpdir):
async def test_upload_image_ova(app: FastAPI, client: AsyncClient, tmpdir:str) -> None:
with patch("gns3server.compute.Qemu.get_images_directory", return_value=str(tmpdir)):
response = await compute_api.post("/qemu/images/test2.ova/test2.vmdk", body="TEST", raw=True)
assert response.status_code == 204
response = await client.post(app.url_path_for("upload_qemu_image",
filename="test2.ova/test2.vmdk"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
with open(str(tmpdir / "test2.ova" / "test2.vmdk")) as f:
assert f.read() == "TEST"
@@ -322,28 +388,27 @@ async def test_upload_image_ova(compute_api, tmpdir):
assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf"
@pytest.mark.asyncio
async def test_upload_image_forbiden_location(compute_api, tmpdir):
async def test_upload_image_forbiden_location(app: FastAPI, client: AsyncClient, tmpdir: str) -> None:
with patch("gns3server.compute.Qemu.get_images_directory", return_value=str(tmpdir)):
response = await compute_api.post("/qemu/images/../../test2", body="TEST", raw=True)
assert response.status_code == 404
response = await client.post(app.url_path_for("upload_qemu_image",
filename="/qemu/images/../../test2"), content=b"TEST")
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")
@pytest.mark.asyncio
async def test_upload_image_permission_denied(compute_api, images_dir):
async def test_upload_image_permission_denied(app: FastAPI, 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 compute_api.post("/qemu/images/test2", body="TEST", raw=True)
assert response.status_code == 409
response = await client.post(app.url_path_for("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(compute_api):
async def test_create_img_relative(app: FastAPI, client: AsyncClient):
params = {
"qemu_img": "/tmp/qemu-img",
@@ -356,12 +421,11 @@ async def test_create_img_relative(compute_api):
"size": 100
}
with asyncio_patch("gns3server.compute.Qemu.create_disk"):
response = await compute_api.post("/qemu/img", params)
assert response.status_code == 204
response = await client.post(app.url_path_for("create_qemu_image"), json=params)
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_create_img_absolute_non_local(compute_api, config):
async def test_create_img_absolute_non_local(app: FastAPI, client: AsyncClient, config: dict) -> None:
config.set("Server", "local", "false")
params = {
@@ -375,12 +439,11 @@ async def test_create_img_absolute_non_local(compute_api, config):
"size": 100
}
with asyncio_patch("gns3server.compute.Qemu.create_disk"):
response = await compute_api.post("/qemu/img", params)
response = await client.post(app.url_path_for("create_qemu_image"), json=params)
assert response.status_code == 403
@pytest.mark.asyncio
async def test_create_img_absolute_local(compute_api, config):
async def test_create_img_absolute_local(app: FastAPI, client: AsyncClient, config: dict) -> None:
config.set("Server", "local", "true")
params = {
@@ -394,60 +457,76 @@ async def test_create_img_absolute_local(compute_api, config):
"size": 100
}
with asyncio_patch("gns3server.compute.Qemu.create_disk"):
response = await compute_api.post("/qemu/img", params)
assert response.status_code == 204
response = await client.post(app.url_path_for("create_qemu_image"), json=params)
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_capabilities(compute_api):
async def test_capabilities(app: FastAPI, client: AsyncClient) -> None:
with asyncio_patch("gns3server.compute.Qemu.get_kvm_archs", return_value=["x86_64"]):
response = await compute_api.get("/qemu/capabilities")
assert response.json["kvm"] == ["x86_64"]
response = await client.get(app.url_path_for("get_qemu_capabilities"))
assert response.json()["kvm"] == ["x86_64"]
@pytest.mark.asyncio
async def test_qemu_duplicate(compute_api, compute_project, vm, base_params):
async def test_qemu_duplicate(app: FastAPI,
client: AsyncClient,
compute_project: Project,
vm: dict,
base_params: dict) -> None:
# create destination node first
response = await compute_api.post("/projects/{project_id}/qemu/nodes".format(project_id=compute_project.id), base_params)
assert response.status_code == 201
response = await client.post(app.url_path_for("create_qemu_node",
project_id=vm["project_id"]), json=base_params)
params = {"destination_node_id": response.json["node_id"]}
response = await compute_api.post("/projects/{project_id}/qemu/nodes/{node_id}/duplicate".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
assert response.status_code == status.HTTP_201_CREATED
params = {"destination_node_id": response.json()["node_id"]}
response = await client.post(app.url_path_for("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(compute_api, vm):
async def test_qemu_start_capture(app: FastAPI, client: AsyncClient, vm):
params = {
"capture_file_name": "test.pcap",
"data_link_type": "DLT_EN10MB"
}
url = app.url_path_for("start_qemu_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
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 compute_api.post("/projects/{project_id}/qemu/nodes/{node_id}/adapters/0/ports/0/capture/start".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 200
response = await 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"]
assert "test.pcap" in response.json()["pcap_file_path"]
@pytest.mark.asyncio
async def test_qemu_stop_capture(compute_api, vm):
async def test_qemu_stop_capture(app: FastAPI, client: AsyncClient, vm):
url = app.url_path_for("stop_qemu_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
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 compute_api.post("/projects/{project_id}/qemu/nodes/{node_id}/adapters/0/ports/0/capture/stop".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204
response = await client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
# @pytest.mark.asyncio
# async def test_qemu_pcap(compute_api, vm, compute_project):
# async def test_qemu_pcap(app: FastAPI, 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 compute_api.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 == 200
# 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)
# assert response.status_code == status.HTTP_200_OK
+104 -66
View File
@@ -16,13 +16,19 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import pytest
from fastapi import FastAPI, status
from httpx import AsyncClient
from tests.utils import asyncio_patch
from unittest.mock import patch
from gns3server.compute.project import Project
pytestmark = pytest.mark.asyncio
@pytest.fixture(scope="function")
@pytest.mark.asyncio
async def vm(compute_api, compute_project):
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
vboxmanage_path = "/fake/VboxManage"
params = {
@@ -32,16 +38,16 @@ async def vm(compute_api, compute_project):
}
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.create", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/virtualbox/nodes".format(project_id=compute_project.id), params)
response = await client.post(app.url_path_for("create_virtualbox_node", project_id=compute_project.id),
json=params)
assert mock.called
assert response.status_code == 201
assert response.status_code == status.HTTP_201_CREATED
with patch("gns3server.compute.virtualbox.VirtualBox.find_vboxmanage", return_value=vboxmanage_path):
return response.json
return response.json()
@pytest.mark.asyncio
async def test_vbox_create(compute_api, compute_project):
async def test_vbox_create(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
params = {
"name": "VM1",
@@ -50,68 +56,76 @@ async def test_vbox_create(compute_api, compute_project):
}
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.create", return_value=True):
response = await compute_api.post("/projects/{project_id}/virtualbox/nodes".format(project_id=compute_project.id), params)
assert response.status_code == 201
assert response.json["name"] == "VM1"
assert response.json["project_id"] == compute_project.id
response = await client.post(app.url_path_for("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
@pytest.mark.asyncio
async def test_vbox_get(compute_api, compute_project, vm):
async def test_vbox_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await compute_api.get("/projects/{project_id}/virtualbox/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 200
assert response.json["name"] == "VMTEST"
assert response.json["project_id"] == compute_project.id
response = await client.get(app.url_path_for("get_virtualbox_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
@pytest.mark.asyncio
async def test_vbox_start(compute_api, vm):
async def test_vbox_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.check_hw_virtualization", return_value=True):
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.start", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/virtualbox/nodes/{node_id}/start".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("start_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vbox_stop(compute_api, vm):
async def test_vbox_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.stop", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/virtualbox/nodes/{node_id}/stop".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("stop_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vbox_suspend(compute_api, vm):
async def test_vbox_suspend(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.suspend", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/virtualbox/nodes/{node_id}/suspend".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("suspend_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vbox_resume(compute_api, vm):
async def test_vbox_resume(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.resume", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/virtualbox/nodes/{node_id}/resume".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("resume_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vbox_reload(compute_api, vm):
async def test_vbox_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.reload", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/virtualbox/nodes/{node_id}/reload".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("reload_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vbox_nio_create_udp(compute_api, vm):
async def test_vbox_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -120,18 +134,24 @@ async def test_vbox_nio_create_udp(compute_api, vm):
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_virtualbox_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch('gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.adapter_add_nio_binding') as mock:
response = await compute_api.post("/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 client.post(url, json=params)
assert mock.called
args, kwgars = mock.call_args
assert args[0] == 0
assert response.status_code == 201
assert response.json["type"] == "nio_udp"
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
# @pytest.mark.asyncio
# async def test_vbox_nio_update_udp(compute_api, vm):
# async def test_vbox_nio_update_udp(app: FastAPI, client: AsyncClient, vm):
#
# params = {
# "type": "nio_udp",
@@ -143,68 +163,86 @@ async def test_vbox_nio_create_udp(compute_api, vm):
#
# 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 compute_api.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 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 == 201
# assert response.json["type"] == "nio_udp"
# assert response.status_code == status.HTTP_201_CREATED
# assert response.json()["type"] == "nio_udp"
@pytest.mark.asyncio
async def test_vbox_delete_nio(compute_api, vm):
async def test_vbox_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("delete_virtualbox_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch('gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.adapter_remove_nio_binding') as mock:
response = await compute_api.delete("/projects/{project_id}/virtualbox/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.delete(url)
assert mock.called
args, kwgars = mock.call_args
assert args[0] == 0
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vbox_update(compute_api, vm, free_console_port):
async def test_vbox_update(app: FastAPI, client: AsyncClient, vm, free_console_port):
params = {
"name": "test",
"console": free_console_port
}
response = await compute_api.put("/projects/{project_id}/virtualbox/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 200
assert response.json["name"] == "test"
assert response.json["console"] == free_console_port
response = await client.put(app.url_path_for("update_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "test"
assert response.json()["console"] == free_console_port
@pytest.mark.asyncio
async def test_virtualbox_start_capture(compute_api, vm):
async def test_virtualbox_start_capture(app: FastAPI, client: AsyncClient, vm):
params = {
"capture_file_name": "test.pcap",
"data_link_type": "DLT_EN10MB"
}
url = app.url_path_for("start_virtualbox_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
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 compute_api.post("/projects/{project_id}/virtualbox/nodes/{node_id}/adapters/0/ports/0/capture/start".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 200
response = await 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"]
assert "test.pcap" in response.json()["pcap_file_path"]
@pytest.mark.asyncio
async def test_virtualbox_stop_capture(compute_api, vm):
async def test_virtualbox_stop_capture(app: FastAPI, client: AsyncClient, vm):
url = app.url_path_for("stop_virtualbox_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
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 compute_api.post("/projects/{project_id}/virtualbox/nodes/{node_id}/adapters/0/ports/0/capture/stop".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204
response = await client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
# @pytest.mark.asyncio
# async def test_virtualbox_pcap(compute_api, vm, compute_project):
# async def test_virtualbox_pcap(app: FastAPI, 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 compute_api.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 == 200
# 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)
# assert response.status_code == status.HTTP_200_OK
+102 -71
View File
@@ -16,13 +16,19 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import pytest
from fastapi import FastAPI, status
from httpx import AsyncClient
from tests.utils import asyncio_patch
from unittest.mock import patch
from gns3server.compute.project import Project
pytestmark = pytest.mark.asyncio
@pytest.fixture(scope="function")
@pytest.mark.asyncio
async def vm(compute_api, compute_project, vmx_path):
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, vmx_path: str) -> dict:
params = {
"name": "VMTEST",
@@ -31,15 +37,15 @@ async def vm(compute_api, compute_project, vmx_path):
}
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.create", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/vmware/nodes".format(project_id=compute_project.id), params)
assert mock.called
assert response.status_code == 201, response.body.decode()
return response.json
response = await client.post(app.url_path_for("create_vmware_node", project_id=compute_project.id),
json=params)
assert mock.called
assert response.status_code == status.HTTP_201_CREATED
return response.json()
@pytest.fixture
@pytest.mark.asyncio
def vmx_path(tmpdir):
def vmx_path(tmpdir: str) -> str:
"""
Return a fake VMX file
"""
@@ -50,8 +56,7 @@ def vmx_path(tmpdir):
return path
@pytest.mark.asyncio
async def test_vmware_create(compute_api, compute_project, vmx_path):
async def test_vmware_create(app: FastAPI, client: AsyncClient, compute_project: Project, vmx_path: str) -> None:
params = {
"name": "VM1",
@@ -60,70 +65,74 @@ async def test_vmware_create(compute_api, compute_project, vmx_path):
}
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.create", return_value=True):
response = await compute_api.post("/projects/{project_id}/vmware/nodes".format(project_id=compute_project.id), params)
assert response.status_code == 201, response.body.decode()
assert response.json["name"] == "VM1"
assert response.json["project_id"] == compute_project.id
response = await client.post(app.url_path_for("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
@pytest.mark.asyncio
async def test_vmware_get(compute_api, compute_project, vm):
async def test_vmware_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await compute_api.get("/projects/{project_id}/vmware/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 200
assert response.json["name"] == "VMTEST"
assert response.json["project_id"] == compute_project.id
response = await client.get(app.url_path_for("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
@pytest.mark.asyncio
async def test_vmware_start(compute_api, vm):
async def test_vmware_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.check_hw_virtualization", return_value=True) as mock1:
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.start", return_value=True) as mock2:
response = await compute_api.post("/projects/{project_id}/vmware/nodes/{node_id}/start".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("start_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock1.called
assert mock2.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vmware_stop(compute_api, vm):
async def test_vmware_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.stop", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/vmware/nodes/{node_id}/stop".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("stop_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vmware_suspend(compute_api, vm):
async def test_vmware_suspend(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.suspend", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/vmware/nodes/{node_id}/suspend".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("suspend_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vmware_resume(compute_api, vm):
async def test_vmware_resume(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.resume", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/vmware/nodes/{node_id}/resume".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("resume_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vmware_reload(compute_api, vm):
async def test_vmware_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.reload", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/vmware/nodes/{node_id}/reload".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("reload_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vmware_nio_create_udp(compute_api, vm):
async def test_vmware_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -132,18 +141,24 @@ async def test_vmware_nio_create_udp(compute_api, vm):
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_vmware_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch('gns3server.compute.vmware.vmware_vm.VMwareVM.adapter_add_nio_binding') as mock:
response = await compute_api.post("/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 client.post(url, json=params)
assert mock.called
args, kwgars = mock.call_args
assert args[0] == 0
assert response.status_code == 201
assert response.json["type"] == "nio_udp"
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
# @pytest.mark.asyncio
# async def test_vmware_nio_update_udp(compute_api, vm):
# async def test_vmware_nio_update_udp(app: FastAPI, client: AsyncClient, vm):
#
# params = {
# "type": "nio_udp",
@@ -156,68 +171,84 @@ async def test_vmware_nio_create_udp(compute_api, vm):
# 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 compute_api.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 == 201
# assert response.json["type"] == "nio_udp"
# 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)
# assert response.status_code == status.HTTP_201_CREATED
# assert response.json()["type"] == "nio_udp"
@pytest.mark.asyncio
async def test_vmware_delete_nio(compute_api, vm):
async def test_vmware_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("delete_vmware_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch('gns3server.compute.vmware.vmware_vm.VMwareVM.adapter_remove_nio_binding') as mock:
response = await compute_api.delete("/projects/{project_id}/vmware/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.delete(url)
assert mock.called
args, kwgars = mock.call_args
assert args[0] == 0
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vmware_update(compute_api, vm, free_console_port):
async def test_vmware_update(app: FastAPI, client: AsyncClient, vm: dict, free_console_port: int) -> None:
params = {
"name": "test",
"console": free_console_port
}
response = await compute_api.put("/projects/{project_id}/vmware/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 200
assert response.json["name"] == "test"
assert response.json["console"] == free_console_port
response = await client.put(app.url_path_for("update_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "test"
assert response.json()["console"] == free_console_port
@pytest.mark.asyncio
async def test_vmware_start_capture(compute_api, vm):
async def test_vmware_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"capture_file_name": "test.pcap",
"data_link_type": "DLT_EN10MB"
}
url = app.url_path_for("start_vmware_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
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 compute_api.post("/projects/{project_id}/vmware/nodes/{node_id}/adapters/0/ports/0/capture/start".format(project_id=vm["project_id"], node_id=vm["node_id"]), body=params)
assert response.status_code == 200
response = await 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"]
assert "test.pcap" in response.json()["pcap_file_path"]
@pytest.mark.asyncio
async def test_vmware_stop_capture(compute_api, vm):
async def test_vmware_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("stop_vmware_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
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 compute_api.post("/projects/{project_id}/vmware/nodes/{node_id}/adapters/0/ports/0/capture/stop".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204
response = await client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
# @pytest.mark.asyncio
# async def test_vmware_pcap(compute_api, vm, compute_project):
# async def test_vmware_pcap(app: FastAPI, 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 compute_api.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 == 200
# 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)
# assert response.status_code == status.HTTP_200_OK
+135 -90
View File
@@ -16,72 +16,75 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import pytest
import uuid
from fastapi import FastAPI, status
from httpx import AsyncClient
from tests.utils import asyncio_patch
from unittest.mock import patch
from gns3server.compute.project import Project
pytestmark = pytest.mark.asyncio
@pytest.fixture
@pytest.mark.asyncio
async def vm(compute_api, compute_project):
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
params = {"name": "PC TEST 1"}
response = await compute_api.post("/projects/{project_id}/vpcs/nodes".format(project_id=compute_project.id), params)
assert response.status_code == 201
return response.json
response = await client.post(app.url_path_for("create_vpcs_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
return response.json()
@pytest.mark.asyncio
async def test_vpcs_create(compute_api, compute_project):
async def test_vpcs_create(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
params = {"name": "PC TEST 1"}
response = await compute_api.post("/projects/{project_id}/vpcs/nodes".format(project_id=compute_project.id), params)
assert response.status_code == 201
assert response.json["name"] == "PC TEST 1"
assert response.json["project_id"] == compute_project.id
response = await client.post(app.url_path_for("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
@pytest.mark.asyncio
async def test_vpcs_get(compute_api, compute_project, vm):
async def test_vpcs_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await compute_api.get("/projects/{project_id}/vpcs/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 200
assert response.json["name"] == "PC TEST 1"
assert response.json["project_id"] == compute_project.id
assert response.json["status"] == "stopped"
response = await client.get(app.url_path_for("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"
@pytest.mark.asyncio
async def test_vpcs_create_startup_script(compute_api, compute_project):
async def test_vpcs_create_startup_script(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
params = {
"name": "PC TEST 1",
"startup_script": "ip 192.168.1.2\necho TEST"
}
response = await compute_api.post("/projects/{project_id}/vpcs/nodes".format(project_id=compute_project.id), params)
assert response.status_code == 201
assert response.json["name"] == "PC TEST 1"
assert response.json["project_id"] == compute_project.id
response = await client.post(app.url_path_for("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
@pytest.mark.asyncio
async def test_vpcs_create_port(compute_api, compute_project, free_console_port):
async def test_vpcs_create_port(app: FastAPI,
client: AsyncClient,
compute_project: Project,
free_console_port: int) -> None:
params = {
"name": "PC TEST 1",
"console": free_console_port
}
response = await compute_api.post("/projects/{project_id}/vpcs/nodes".format(project_id=compute_project.id), params)
assert response.status_code == 201
assert response.json["name"] == "PC TEST 1"
assert response.json["project_id"] == compute_project.id
assert response.json["console"] == free_console_port
response = await client.post(app.url_path_for("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
@pytest.mark.asyncio
async def test_vpcs_nio_create_udp(compute_api, vm):
async def test_vpcs_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -90,14 +93,19 @@ async def test_vpcs_nio_create_udp(compute_api, vm):
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_vpcs_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.add_ubridge_udp_connection"):
response = await compute_api.post("/projects/{project_id}/vpcs/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
assert response.json["type"] == "nio_udp"
response = await client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
@pytest.mark.asyncio
async def test_vpcs_nio_update_udp(compute_api, vm):
async def test_vpcs_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -106,18 +114,28 @@ async def test_vpcs_nio_update_udp(compute_api, vm):
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_vpcs_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.add_ubridge_udp_connection"):
response = await compute_api.post("/projects/{project_id}/vpcs/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
response = await client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
params["filters"] = {}
response = await compute_api.put("/projects/{project_id}/vpcs/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201, response.body.decode("utf-8")
assert response.json["type"] == "nio_udp"
url = app.url_path_for("update_vpcs_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
response = await client.put(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
@pytest.mark.asyncio
async def test_vpcs_delete_nio(compute_api, vm):
async def test_vpcs_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -127,63 +145,78 @@ async def test_vpcs_delete_nio(compute_api, vm):
}
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM._ubridge_send"):
await compute_api.post("/projects/{project_id}/vpcs/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
response = await compute_api.delete("/projects/{project_id}/vpcs/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204, response.body.decode()
url = app.url_path_for("create_vpcs_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
await client.post(url, json=params)
url = app.url_path_for("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)
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vpcs_start(compute_api, vm):
async def test_vpcs_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.start", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/vpcs/nodes/{node_id}/start".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("start_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vpcs_stop(compute_api, vm):
async def test_vpcs_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.stop", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/vpcs/nodes/{node_id}/stop".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("stop_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vpcs_reload(compute_api, vm):
async def test_vpcs_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.reload", return_value=True) as mock:
response = await compute_api.post("/projects/{project_id}/vpcs/nodes/{node_id}/reload".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.post(app.url_path_for("reload_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vpcs_delete(compute_api, vm):
async def test_vpcs_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vpcs.VPCS.delete_node", return_value=True) as mock:
response = await compute_api.delete("/projects/{project_id}/vpcs/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.delete(app.url_path_for("delete_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == 204
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.asyncio
async def test_vpcs_duplicate(compute_api, compute_project, vm):
async def test_vpcs_duplicate(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
# create destination node first
params = {"name": "PC TEST 1"}
response = await compute_api.post("/projects/{project_id}/vpcs/nodes".format(project_id=compute_project.id), params)
assert response.status_code == 201
response = await client.post(app.url_path_for("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 compute_api.post("/projects/{project_id}/vpcs/nodes/{node_id}/duplicate".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 201
params = {"destination_node_id": response.json()["node_id"]}
response = await client.post(app.url_path_for("duplicate_vpcs_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_vpcs_update(compute_api, vm, free_console_port):
async def test_vpcs_update(app: FastAPI, client: AsyncClient, vm: dict, free_console_port: int) -> None:
console_port = free_console_port
params = {
@@ -191,42 +224,54 @@ async def test_vpcs_update(compute_api, vm, free_console_port):
"console": console_port
}
response = await compute_api.put("/projects/{project_id}/vpcs/nodes/{node_id}".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
assert response.status_code == 200
assert response.json["name"] == "test"
assert response.json["console"] == console_port
response = await client.put(app.url_path_for("update_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "test"
assert response.json()["console"] == console_port
@pytest.mark.asyncio
async def test_vpcs_start_capture(compute_api, vm):
async def test_vpcs_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
params = {
"capture_file_name": "test.pcap",
"data_link_type": "DLT_EN10MB"
}
url = app.url_path_for("start_vpcs_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
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 compute_api.post("/projects/{project_id}/vpcs/nodes/{node_id}/adapters/0/ports/0/capture/start".format(project_id=vm["project_id"], node_id=vm["node_id"]), body=params)
assert response.status_code == 200
response = await 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"]
assert "test.pcap" in response.json()["pcap_file_path"]
@pytest.mark.asyncio
async def test_vpcs_stop_capture(compute_api, vm):
async def test_vpcs_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("stop_vpcs_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
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 compute_api.post("/projects/{project_id}/vpcs/nodes/{node_id}/adapters/0/ports/0/capture/stop".format(project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == 204
response = await client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
# @pytest.mark.asyncio
# async def test_vpcs_pcap(compute_api, vm, compute_project):
# async def test_vpcs_pcap(app: FastAPI, 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 compute_api.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 == 200
# 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)
# assert response.status_code == status.HTTP_200_OK