ExternalVendorCode/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2024-12-20 13:23:10 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Return the current controller hostname/IP from any compute

Browse Source
This commit is contained in:
grossmj 2021-10-21 21:38:36 +10:30
parent dddfe7c531
commit bb15b13a9d
18 changed files with 215 additions and 198 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -56,6 +56,8 @@ compute_api = FastAPI(
version="v3",
)
compute_api.state.controller_host = None
@compute_api.exception_handler(ComputeError)
async def controller_error_handler(request: Request, exc: ComputeError):

7
gns3server/api/routes/compute/capabilities.py
View File

@ -21,7 +21,7 @@ API routes for capabilities
import sys
import psutil
from fastapi import APIRouter
from fastapi import APIRouter, Request
from gns3server.version import __version__
from gns3server.compute import MODULES
@ -32,12 +32,15 @@ router = APIRouter()
@router.get("/capabilities", response_model=schemas.Capabilities)
def get_capabilities() -> dict:
def get_capabilities(request: Request) -> dict:
node_types = []
for module in MODULES:
node_types.extend(module.node_types())
# record the controller hostname or IP address
request.app.state.controller_host = request.client.host
return {
"version": __version__,
"platform": sys.platform,

19
gns3server/api/routes/controller/controller.py
View File

@ -18,8 +18,9 @@ import asyncio
import signal
import os
from fastapi import APIRouter, Depends, Response, status
from fastapi import APIRouter, Depends, Request, Response, status
from fastapi.encoders import jsonable_encoder
from fastapi.routing import Mount
from typing import List
from gns3server.config import Config
@ -41,13 +42,24 @@ router = APIRouter()
"/version",
response_model=schemas.Version,
)
def get_version() -> dict:
def get_version(request: Request) -> dict:
"""
Return the server version number.
"""
# retrieve the controller host information from the mounted
# compute subapp
controller_host = None
for route in request.app.routes:
if isinstance(route, Mount) and route.name == "compute":
controller_host = route.app.state.controller_host
local_server = Config.instance().settings.Server.local
return {"version": __version__, "local": local_server}
return {
"controller_host": controller_host,
"version": __version__,
"local": local_server
}
@router.post(
@ -61,7 +73,6 @@ def check_version(version: schemas.Version) -> dict:
Check if version is the same as the server.
"""
print(version.version)
if version.version != __version__:
raise ControllerError(f"Client version {version.version} is not the same as server version {__version__}")
return {"version": __version__}

2
gns3server/api/server.py
View File

@ -78,7 +78,7 @@ def get_application() -> FastAPI:
application.add_event_handler("shutdown", tasks.create_shutdown_handler(application))
application.include_router(index.router, tags=["Index"])
application.include_router(controller.router, prefix="/v3")
application.mount("/v3/compute", compute_api)
application.mount("/v3/compute", compute_api, name="compute")
return application

1
gns3server/schemas/version.py
View File

@ -20,5 +20,6 @@ from typing import Optional
class Version(BaseModel):
controller_host: Optional[str] = Field(None, description="Controller hostname or IP address")
version: str = Field(..., description="Version number")
local: Optional[bool] = Field(None, description="Whether this is a local server or not")

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

@ -32,7 +32,7 @@ pytestmark = pytest.mark.asyncio
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
async def test_get(app: FastAPI, client: AsyncClient, windows_platform) -> None:
response = await client.get(app.url_path_for("get_capabilities"))
response = await client.get(app.url_path_for("compute:get_capabilities"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == {'node_types': ['cloud', 'ethernet_hub', 'ethernet_switch', 'nat', 'vpcs', 'virtualbox', 'dynamips', 'frame_relay_switch', 'atm_switch', 'qemu', 'vmware', 'docker', 'iou'],
'version': __version__,
@ -46,7 +46,7 @@ async def test_get(app: FastAPI, client: AsyncClient, windows_platform) -> None:
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
async def test_get_on_gns3vm(app: FastAPI, client: AsyncClient, on_gns3vm) -> None:
response = await client.get(app.url_path_for("get_capabilities"))
response = await client.get(app.url_path_for("compute:get_capabilities"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == {'node_types': ['cloud', 'ethernet_hub', 'ethernet_switch', 'nat', 'vpcs', 'virtualbox', 'dynamips', 'frame_relay_switch', 'atm_switch', 'qemu', 'vmware', 'docker', 'iou'],
'version': __version__,

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

@ -31,7 +31,7 @@ pytestmark = pytest.mark.asyncio
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 client.post(app.url_path_for("create_cloud", project_id=compute_project.id),
response = await client.post(app.url_path_for("compute:create_cloud", project_id=compute_project.id),
json={"name": "Cloud 1"})
assert response.status_code == status.HTTP_201_CREATED
return response.json()
@ -40,7 +40,7 @@ async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, on_gns
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 client.post(app.url_path_for("create_cloud", project_id=compute_project.id),
response = await client.post(app.url_path_for("compute:create_cloud", project_id=compute_project.id),
json={"name": "Cloud 1"})
assert response.status_code == 201
assert response.json()["name"] == "Cloud 1"
@ -49,7 +49,7 @@ async def test_cloud_create(app: FastAPI, client: AsyncClient, compute_project:
async def test_get_cloud(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await client.get(app.url_path_for("get_cloud", project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.get(app.url_path_for("compute:get_cloud", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "Cloud 1"
assert response.json()["project_id"] == compute_project.id
@ -63,7 +63,7 @@ async def test_cloud_nio_create_udp(app: FastAPI, client: AsyncClient, compute_p
"rport": 4343,
"rhost": "127.0.0.1"}
url = app.url_path_for("create_cloud_nio",
url = app.url_path_for("compute:create_cloud_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -80,7 +80,7 @@ async def test_cloud_nio_update_udp(app: FastAPI, client: AsyncClient, compute_p
"rport": 4343,
"rhost": "127.0.0.1"}
url = app.url_path_for("create_cloud_nio",
url = app.url_path_for("compute:create_cloud_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -88,7 +88,7 @@ async def test_cloud_nio_update_udp(app: FastAPI, client: AsyncClient, compute_p
await client.post(url, json=params)
params["filters"] = {}
url = app.url_path_for("create_cloud_nio",
url = app.url_path_for("compute:create_cloud_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -105,14 +105,14 @@ async def test_cloud_delete_nio(app: FastAPI, client: AsyncClient, compute_proje
"rport": 4343,
"rhost": "127.0.0.1"}
url = app.url_path_for("create_cloud_nio",
url = app.url_path_for("compute: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",
url = app.url_path_for("compute:delete_cloud_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -124,13 +124,13 @@ async def test_cloud_delete_nio(app: FastAPI, client: AsyncClient, compute_proje
async def test_cloud_delete(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await client.delete(app.url_path_for("delete_cloud", project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.delete(app.url_path_for("compute:delete_cloud", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_cloud_update(app: FastAPI, client: AsyncClient, vm: dict) -> None:
response = await client.put(app.url_path_for("update_cloud", project_id=vm["project_id"], node_id=vm["node_id"]),
response = await client.put(app.url_path_for("compute:update_cloud", project_id=vm["project_id"], node_id=vm["node_id"]),
json={"name": "test"})
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "test"
@ -144,7 +144,7 @@ async def test_cloud_start_capture(app: FastAPI, client: AsyncClient, vm: dict)
}
with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud.start_capture") as mock:
response = await client.post(app.url_path_for("start_cloud_capture",
response = await client.post(app.url_path_for("compute:start_cloud_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -158,7 +158,7 @@ async def test_cloud_start_capture(app: FastAPI, client: AsyncClient, vm: dict)
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 client.post(app.url_path_for("stop_cloud_capture",
response = await client.post(app.url_path_for("compute:stop_cloud_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",

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

@ -29,21 +29,21 @@ pytestmark = pytest.mark.asyncio
async def test_udp_allocation(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
response = await client.post(app.url_path_for("allocate_udp_port", project_id=compute_project.id), json={})
response = await client.post(app.url_path_for("compute:allocate_udp_port", project_id=compute_project.id), json={})
assert response.status_code == status.HTTP_201_CREATED
assert response.json()['udp_port'] is not None
async def test_interfaces(app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("network_interfaces"))
response = await client.get(app.url_path_for("compute:network_interfaces"))
assert response.status_code == status.HTTP_200_OK
assert isinstance(response.json(), list)
async def test_version_output(app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("compute_version"))
response = await client.get(app.url_path_for("compute:compute_version"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == {'local': True, 'version': __version__}
@ -57,5 +57,5 @@ async def test_version_output(app: FastAPI, client: AsyncClient) -> None:
async def test_statistics_output(app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("compute_statistics"))
response = await client.get(app.url_path_for("compute:compute_statistics"))
assert response.status_code == status.HTTP_200_OK

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

@ -62,7 +62,7 @@ async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, base_p
with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "nginx"}]):
with asyncio_patch("gns3server.compute.docker.Docker.query", return_value={"Id": "8bd8153ea8f5"}):
with asyncio_patch("gns3server.compute.docker.DockerVM._get_container_state", return_value="exited"):
response = await client.post(app.url_path_for("create_docker_node", project_id=compute_project.id),
response = await client.post(app.url_path_for("compute:create_docker_node", project_id=compute_project.id),
json=base_params)
assert response.status_code == status.HTTP_201_CREATED
return response.json()
@ -72,7 +72,7 @@ async def test_docker_create(app: FastAPI, client: AsyncClient, compute_project:
with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "nginx"}]):
with asyncio_patch("gns3server.compute.docker.Docker.query", return_value={"Id": "8bd8153ea8f5"}):
response = await client.post(app.url_path_for("create_docker_node", project_id=compute_project.id),
response = await client.post(app.url_path_for("compute:create_docker_node", project_id=compute_project.id),
json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
@ -89,7 +89,7 @@ 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 client.post(app.url_path_for("start_docker_node",
response = await client.post(app.url_path_for("compute:start_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -99,7 +99,7 @@ async def test_docker_start(app: FastAPI, client: AsyncClient, vm: dict) -> None
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 client.post(app.url_path_for("stop_docker_node",
response = await client.post(app.url_path_for("compute:stop_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -109,7 +109,7 @@ async def test_docker_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
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 client.post(app.url_path_for("reload_docker_node",
response = await client.post(app.url_path_for("compute:reload_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -119,7 +119,7 @@ async def test_docker_reload(app: FastAPI, client: AsyncClient, vm: dict) -> Non
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 client.delete(app.url_path_for("delete_docker_node",
response = await client.delete(app.url_path_for("compute:delete_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -129,7 +129,7 @@ async def test_docker_delete(app: FastAPI, client: AsyncClient, vm: dict) -> Non
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 client.post(app.url_path_for("pause_docker_node",
response = await client.post(app.url_path_for("compute:pause_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -139,7 +139,7 @@ async def test_docker_pause(app: FastAPI, client: AsyncClient, vm: dict) -> None
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 client.post(app.url_path_for("unpause_docker_node",
response = await client.post(app.url_path_for("compute:unpause_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -154,7 +154,7 @@ async def test_docker_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict
"rport": 4343,
"rhost": "127.0.0.1"}
url = app.url_path_for("create_docker_node_nio",
url = app.url_path_for("compute:create_docker_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -173,7 +173,7 @@ async def test_docker_update_nio(app: FastAPI, client: AsyncClient, vm: dict) ->
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_docker_node_nio",
url = app.url_path_for("compute:create_docker_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -181,7 +181,7 @@ async def test_docker_update_nio(app: FastAPI, client: AsyncClient, vm: dict) ->
response = await client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
url = app.url_path_for("update_docker_node_nio",
url = app.url_path_for("compute:update_docker_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -193,7 +193,7 @@ async def test_docker_update_nio(app: FastAPI, client: AsyncClient, vm: dict) ->
async def test_docker_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("delete_docker_node_nio",
url = app.url_path_for("compute:delete_docker_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -214,7 +214,7 @@ async def test_docker_update(app: FastAPI, client: AsyncClient, vm: dict, free_c
}
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.update") as mock:
response = await client.put(app.url_path_for("update_docker_node",
response = await client.put(app.url_path_for("compute:update_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
@ -229,7 +229,7 @@ async def test_docker_update(app: FastAPI, client: AsyncClient, vm: dict, free_c
async def test_docker_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("start_docker_node_capture",
url = app.url_path_for("compute:start_docker_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -246,7 +246,7 @@ async def test_docker_start_capture(app: FastAPI, client: AsyncClient, vm: dict)
async def test_docker_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("stop_docker_node_capture",
url = app.url_path_for("compute:stop_docker_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -264,12 +264,12 @@ async def test_docker_duplicate(app: FastAPI, client: AsyncClient, vm: dict, bas
# create destination node first
with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "nginx"}]):
with asyncio_patch("gns3server.compute.docker.Docker.query", return_value={"Id": "8bd8153ea8f5"}):
response = await client.post(app.url_path_for("create_docker_node",
response = await client.post(app.url_path_for("compute:create_docker_node",
project_id=vm["project_id"]), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
params = {"destination_node_id": response.json()["node_id"]}
response = await client.post(app.url_path_for("duplicate_docker_node",
response = await client.post(app.url_path_for("compute:duplicate_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_201_CREATED

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

@ -165,7 +165,7 @@ def fake_file(tmpdir) -> str:
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 client.get(app.url_path_for("get_dynamips_images"))
response = await client.get(app.url_path_for("compute:get_dynamips_images"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == [{"filename": "7200.bin",
"path": "7200.bin",
@ -175,7 +175,7 @@ async def test_images(app: FastAPI, client: AsyncClient, tmpdir, fake_image: str
async def test_upload_image(app: FastAPI, client: AsyncClient, images_dir: str) -> None:
response = await client.post(app.url_path_for("upload_dynamips_image", filename="test2"), content=b"TEST")
response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename="test2"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
with open(os.path.join(images_dir, "IOS", "test2")) as f:
@ -189,23 +189,23 @@ async def test_upload_image(app: FastAPI, client: AsyncClient, images_dir: str)
async def test_upload_image_forbidden_location(app: FastAPI, client: AsyncClient) -> None:
file_path = "%2e%2e/hello"
response = await client.post(app.url_path_for("upload_dynamips_image", filename=file_path), content=b"TEST")
response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename=file_path), content=b"TEST")
assert response.status_code == status.HTTP_403_FORBIDDEN
async def test_download_image(app: FastAPI, client: AsyncClient, images_dir: str) -> None:
response = await client.post(app.url_path_for("upload_dynamips_image", filename="test3"), content=b"TEST")
response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename="test3"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
response = await client.get(app.url_path_for("download_dynamips_image", filename="test3"))
response = await client.get(app.url_path_for("compute:download_dynamips_image", filename="test3"))
assert response.status_code == status.HTTP_200_OK
async def test_download_image_forbidden(app: FastAPI, client: AsyncClient, tmpdir) -> None:
file_path = "foo/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/etc/passwd"
response = await client.get(app.url_path_for("download_dynamips_image", filename=file_path))
response = await client.get(app.url_path_for("compute:download_dynamips_image", filename=file_path))
assert response.status_code == status.HTTP_403_FORBIDDEN
@ -217,5 +217,5 @@ async def test_upload_image_permission_denied(app: FastAPI, client: AsyncClient,
f.write("")
os.chmod(os.path.join(images_dir, "IOS", "test2.tmp"), 0)
response = await client.post(app.url_path_for("upload_dynamips_image", filename="test2"), content=b"TEST")
response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename="test2"), content=b"TEST")
assert response.status_code == status.HTTP_409_CONFLICT

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

@ -53,7 +53,7 @@ def base_params(tmpdir, fake_iou_bin) -> dict:
@pytest.fixture
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> dict:
response = await client.post(app.url_path_for("create_iou_node", project_id=compute_project.id), json=base_params)
response = await client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
return response.json()
@ -67,7 +67,7 @@ def startup_config_file(compute_project: Project, vm: dict) -> str:
async def test_iou_create(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> None:
response = await client.post(app.url_path_for("create_iou_node", project_id=compute_project.id), json=base_params)
response = await client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@ -92,7 +92,7 @@ async def test_iou_create_with_params(app: FastAPI,
params["startup_config_content"] = "hostname test"
params["use_default_iou_values"] = False
response = await client.post(app.url_path_for("create_iou_node", project_id=compute_project.id), json=params)
response = await client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@ -123,7 +123,7 @@ async def test_iou_create_startup_config_already_exist(
params["node_id"] = node_id
params["startup_config_content"] = "hostname test"
response = await client.post(app.url_path_for("create_iou_node", project_id=compute_project.id), json=params)
response = await client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
with open(startup_config_file(compute_project, response.json())) as f:
@ -132,7 +132,7 @@ async def test_iou_create_startup_config_already_exist(
async def test_iou_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await client.get(app.url_path_for("get_iou_node", project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.get(app.url_path_for("compute:get_iou_node", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@ -146,7 +146,7 @@ async def test_iou_get(app: FastAPI, client: AsyncClient, compute_project: Proje
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 client.post(app.url_path_for("start_iou_node",
response = await client.post(app.url_path_for("compute:start_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json={})
assert mock.called
@ -157,7 +157,7 @@ async def test_iou_start_with_iourc(app: FastAPI, client: AsyncClient, vm: dict)
params = {"iourc_content": "test"}
with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.start", return_value=True) as mock:
response = await client.post(app.url_path_for("start_iou_node",
response = await client.post(app.url_path_for("compute:start_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert mock.called
@ -167,7 +167,7 @@ async def test_iou_start_with_iourc(app: FastAPI, client: AsyncClient, vm: dict)
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 client.post(app.url_path_for("stop_iou_node",
response = await client.post(app.url_path_for("compute:stop_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -177,7 +177,7 @@ async def test_iou_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
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 client.post(app.url_path_for("reload_iou_node",
response = await client.post(app.url_path_for("compute:reload_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -187,7 +187,7 @@ async def test_iou_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None:
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 client.delete(app.url_path_for("delete_iou_node",
response = await client.delete(app.url_path_for("compute:delete_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -207,7 +207,7 @@ async def test_iou_update(app: FastAPI, client: AsyncClient, vm: dict, free_cons
"use_default_iou_values": True,
}
response = await client.put(app.url_path_for("update_iou_node",
response = await client.put(app.url_path_for("compute:update_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
@ -228,7 +228,7 @@ async def test_iou_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -
"rport": 4343,
"rhost": "127.0.0.1"}
url = app.url_path_for("create_iou_node_nio",
url = app.url_path_for("compute:create_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
@ -245,7 +245,7 @@ async def test_iou_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -
"rport": 4343,
"rhost": "127.0.0.1"}
url = app.url_path_for("create_iou_node_nio",
url = app.url_path_for("compute:create_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
@ -254,7 +254,7 @@ async def test_iou_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -
await client.post(url, json=params)
params["filters"] = {}
url = app.url_path_for("update_iou_node_nio",
url = app.url_path_for("compute:update_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
@ -271,7 +271,7 @@ async def test_iou_nio_create_ethernet(app: FastAPI, client: AsyncClient, vm: di
"ethernet_device": ethernet_device
}
url = app.url_path_for("create_iou_node_nio",
url = app.url_path_for("compute:create_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
@ -293,7 +293,7 @@ async def test_iou_nio_create_ethernet_different_port(app: FastAPI,
"ethernet_device": ethernet_device
}
url = app.url_path_for("create_iou_node_nio",
url = app.url_path_for("compute:create_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -311,7 +311,7 @@ async def test_iou_nio_create_tap(app: FastAPI, client: AsyncClient, vm: dict, e
"tap_device": ethernet_device
}
url = app.url_path_for("create_iou_node_nio",
url = app.url_path_for("compute:create_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
@ -331,7 +331,7 @@ async def test_iou_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> No
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_iou_node_nio",
url = app.url_path_for("compute:create_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
@ -339,7 +339,7 @@ async def test_iou_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> No
await client.post(url, json=params)
url = app.url_path_for("delete_iou_node_nio",
url = app.url_path_for("compute:delete_iou_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
@ -356,7 +356,7 @@ async def test_iou_start_capture(app: FastAPI, client: AsyncClient, vm: dict) ->
"data_link_type": "DLT_EN10MB"
}
url = app.url_path_for("start_iou_node_capture",
url = app.url_path_for("compute:start_iou_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -372,7 +372,7 @@ async def test_iou_start_capture(app: FastAPI, client: AsyncClient, vm: dict) ->
async def test_iou_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("stop_iou_node_capture",
url = app.url_path_for("compute:stop_iou_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -396,7 +396,7 @@ async def test_iou_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) ->
async def test_images(app: FastAPI, client: AsyncClient, fake_iou_bin: str) -> None:
response = await client.get(app.url_path_for("get_iou_images"))
response = await client.get(app.url_path_for("compute:get_iou_images"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == [{"filename": "iou.bin", "path": "iou.bin", "filesize": 7, "md5sum": "e573e8f5c93c6c00783f20c7a170aa6c"}]
@ -404,7 +404,7 @@ async def test_images(app: FastAPI, client: AsyncClient, fake_iou_bin: str) -> N
async def test_upload_image(app: FastAPI, client: AsyncClient, tmpdir) -> None:
with patch("gns3server.compute.IOU.get_images_directory", return_value=str(tmpdir)):
response = await client.post(app.url_path_for("upload_iou_image", filename="test2"), content=b"TEST")
response = await client.post(app.url_path_for("compute:upload_iou_image", filename="test2"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
with open(str(tmpdir / "test2")) as f:
@ -418,35 +418,35 @@ async def test_upload_image(app: FastAPI, client: AsyncClient, tmpdir) -> None:
async def test_upload_image_forbidden_location(app: FastAPI, client: AsyncClient) -> None:
file_path = "%2e%2e/hello"
response = await client.post(app.url_path_for("upload_dynamips_image", filename=file_path), content=b"TEST")
response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename=file_path), content=b"TEST")
assert response.status_code == status.HTTP_403_FORBIDDEN
async def test_download_image(app: FastAPI, client: AsyncClient, images_dir: str) -> None:
response = await client.post(app.url_path_for("upload_dynamips_image", filename="test3"), content=b"TEST")
response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename="test3"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
response = await client.get(app.url_path_for("download_dynamips_image", filename="test3"))
response = await client.get(app.url_path_for("compute:download_dynamips_image", filename="test3"))
assert response.status_code == status.HTTP_200_OK
async def test_download_image_forbidden(app: FastAPI, client: AsyncClient, tmpdir) -> None:
file_path = "foo/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/etc/passwd"
response = await client.get(app.url_path_for("download_iou_image", filename=file_path))
response = await client.get(app.url_path_for("compute:download_iou_image", filename=file_path))
assert response.status_code == status.HTTP_403_FORBIDDEN
async def test_iou_duplicate(app: FastAPI, client: AsyncClient, vm: dict, base_params: dict) -> None:
# create destination node first
response = await client.post(app.url_path_for("create_iou_node", project_id=vm["project_id"]), json=base_params)
response = await client.post(app.url_path_for("compute:create_iou_node", project_id=vm["project_id"]), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
params = {"destination_node_id": response.json()["node_id"]}
response = await client.post(app.url_path_for("duplicate_iou_node",
response = await client.post(app.url_path_for("compute:duplicate_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_201_CREATED

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

@ -30,7 +30,7 @@ pytestmark = pytest.mark.asyncio
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 client.post(app.url_path_for("create_nat_node", project_id=compute_project.id),
response = await client.post(app.url_path_for("compute:create_nat_node", project_id=compute_project.id),
json={"name": "Nat 1"})
assert response.status_code == status.HTTP_201_CREATED
return response.json()
@ -39,7 +39,7 @@ async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, ubridg
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 client.post(app.url_path_for("create_nat_node", project_id=compute_project.id),
response = await client.post(app.url_path_for("compute:create_nat_node", project_id=compute_project.id),
json={"name": "Nat 1"})
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "Nat 1"
@ -48,7 +48,7 @@ async def test_nat_create(app: FastAPI, client: AsyncClient, compute_project: Pr
async def test_nat_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await client.get(app.url_path_for("get_nat_node", project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.get(app.url_path_for("compute:get_nat_node", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "Nat 1"
assert response.json()["project_id"] == compute_project.id
@ -64,7 +64,7 @@ async def test_nat_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_nat_node_nio",
url = app.url_path_for("compute:create_nat_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -85,7 +85,7 @@ async def test_nat_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_nat_node_nio",
url = app.url_path_for("compute:create_nat_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -94,7 +94,7 @@ async def test_nat_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -
await client.post(url, json=params)
params["filters"] = {}
url = app.url_path_for("update_nat_node_nio",
url = app.url_path_for("compute:update_nat_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -113,7 +113,7 @@ async def test_nat_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> No
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_nat_node_nio",
url = app.url_path_for("compute:create_nat_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -121,7 +121,7 @@ async def test_nat_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> No
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.add_nio"):
await client.post(url, json=params)
url = app.url_path_for("delete_nat_node_nio",
url = app.url_path_for("compute:delete_nat_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -134,7 +134,7 @@ async def test_nat_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> No
async def test_nat_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None:
response = await client.delete(app.url_path_for("delete_nat_node",
response = await client.delete(app.url_path_for("compute:delete_nat_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
@ -143,7 +143,7 @@ async def test_nat_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_nat_update(app: FastAPI, client: AsyncClient, vm: dict) -> None:
response = await client.put(app.url_path_for("update_nat_node",
response = await client.put(app.url_path_for("compute:update_nat_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json={"name": "test"})
assert response.status_code == status.HTTP_200_OK
@ -157,7 +157,7 @@ async def test_nat_start_capture(app: FastAPI, client: AsyncClient, vm: dict) ->
"data_link_type": "DLT_EN10MB"
}
url = app.url_path_for("start_nat_node_capture",
url = app.url_path_for("compute:start_nat_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -171,7 +171,7 @@ async def test_nat_start_capture(app: FastAPI, client: AsyncClient, vm: dict) ->
async def test_nat_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("stop_nat_node_capture",
url = app.url_path_for("compute:stop_nat_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",

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

@ -45,7 +45,7 @@ def base_params(tmpdir) -> dict:
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 client.post(app.url_path_for("create_compute_project"), json=base_params)
response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["project_id"] == base_params["project_id"]
@ -57,7 +57,7 @@ async def test_create_project_with_path_and_empty_variables(app: FastAPI,
base_params["variables"] = None
with patch("gns3server.compute.project.Project.is_local", return_value=True):
response = await client.post(app.url_path_for("create_compute_project"), json=base_params)
response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["project_id"] == base_params["project_id"]
@ -65,7 +65,7 @@ async def test_create_project_with_path_and_empty_variables(app: FastAPI,
async def test_create_project_without_dir(app: FastAPI, client: AsyncClient, base_params: dict) -> None:
del base_params["path"]
response = await client.post(app.url_path_for("create_compute_project"), json=base_params)
response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["project_id"] == base_params["project_id"]
assert response.json()["name"] == base_params["name"]
@ -73,9 +73,9 @@ async def test_create_project_without_dir(app: FastAPI, client: AsyncClient, bas
async def test_show_project(app: FastAPI, client: AsyncClient, base_params: dict) -> None:
response = await client.post(app.url_path_for("create_compute_project"), json=base_params)
response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
response = await client.get(app.url_path_for("get_compute_project", project_id=base_params["project_id"]))
response = await client.get(app.url_path_for("compute:get_compute_project", project_id=base_params["project_id"]))
#print(response.json().keys())
#assert len(response.json().keys()) == 3
@ -86,7 +86,7 @@ async def test_show_project(app: FastAPI, client: AsyncClient, base_params: dict
async def test_show_project_invalid_uuid(app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("get_compute_project",
response = await client.get(app.url_path_for("compute:get_compute_project",
project_id="50010203-0405-0607-0809-0a0b0c0d0e42"))
assert response.status_code == status.HTTP_404_NOT_FOUND
@ -96,13 +96,13 @@ 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 client.post(app.url_path_for("create_compute_project"), json=params)
response = await client.post(app.url_path_for("compute:create_compute_project"), json=params)
assert response.status_code == status.HTTP_201_CREATED
params = {"name": "test", "project_id": "52010203-0405-0607-0809-0a0b0c0d0e0b"}
response = await client.post(app.url_path_for("create_compute_project"), json=params)
response = await client.post(app.url_path_for("compute:create_compute_project"), json=params)
assert response.status_code == status.HTTP_201_CREATED
response = await client.get(app.url_path_for("get_compute_projects"))
response = await client.get(app.url_path_for("compute:get_compute_projects"))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 2
assert "51010203-0405-0607-0809-0a0b0c0d0e0f" in [p["project_id"] for p in response.json()]
@ -111,18 +111,18 @@ async def test_list_projects(app: FastAPI, client: AsyncClient) -> dict:
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 client.delete(app.url_path_for("delete_compute_project", project_id=compute_project.id))
response = await client.delete(app.url_path_for("compute:delete_compute_project", project_id=compute_project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
async def test_update_project(app: FastAPI, client: AsyncClient, base_params: dict) -> None:
response = await client.post(app.url_path_for("create_compute_project"), json=base_params)
response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
params = {"variables": [{"name": "TEST1", "value": "VAL1"}]}
response = await client.put(app.url_path_for("update_compute_project", project_id=base_params["project_id"]),
response = await client.put(app.url_path_for("compute:update_compute_project", project_id=base_params["project_id"]),
json=params)
assert response.status_code == status.HTTP_200_OK
assert response.json()["variables"] == [{"name": "TEST1", "value": "VAL1"}]
@ -130,14 +130,14 @@ async def test_update_project(app: FastAPI, client: AsyncClient, base_params: di
async def test_delete_project_invalid_uuid(app: FastAPI, client: AsyncClient) -> None:
response = await client.delete(app.url_path_for("delete_compute_project", project_id=str(uuid.uuid4())))
response = await client.delete(app.url_path_for("compute:delete_compute_project", project_id=str(uuid.uuid4())))
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_close_project(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
with asyncio_patch("gns3server.compute.project.Project.close", return_value=True) as mock:
response = await client.post(app.url_path_for("close_compute_project", project_id=compute_project.id))
response = await client.post(app.url_path_for("compute:close_compute_project", project_id=compute_project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
@ -154,7 +154,7 @@ async def test_close_project(app: FastAPI, client: AsyncClient, compute_project:
async def test_close_project_invalid_uuid(app: FastAPI, client: AsyncClient) -> None:
response = await client.post(app.url_path_for("close_compute_project", project_id=str(uuid.uuid4())))
response = await client.post(app.url_path_for("compute:close_compute_project", project_id=str(uuid.uuid4())))
assert response.status_code == status.HTTP_404_NOT_FOUND
@ -166,14 +166,14 @@ async def test_get_file(app: FastAPI, client: AsyncClient, config, tmpdir) -> No
with open(os.path.join(project.path, "hello"), "w+") as f:
f.write("world")
response = await client.get(app.url_path_for("get_compute_project_file", project_id=project.id, file_path="hello"))
response = await client.get(app.url_path_for("compute:get_compute_project_file", project_id=project.id, file_path="hello"))
assert response.status_code == status.HTTP_200_OK
assert response.content == b"world"
response = await client.get(app.url_path_for("get_compute_project_file", project_id=project.id, file_path="false"))
response = await client.get(app.url_path_for("compute:get_compute_project_file", project_id=project.id, file_path="false"))
assert response.status_code == status.HTTP_404_NOT_FOUND
response = await client.get(app.url_path_for("get_compute_project_file",
response = await client.get(app.url_path_for("compute:get_compute_project_file",
project_id=project.id,
file_path="../hello"))
assert response.status_code == status.HTTP_404_NOT_FOUND
@ -186,7 +186,7 @@ async def test_get_file_forbidden_location(app: FastAPI, client: AsyncClient, co
file_path = "foo/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/etc/passwd"
response = await client.get(
app.url_path_for(
"get_compute_project_file",
"compute:get_compute_project_file",
project_id=project.id,
file_path=file_path
)
@ -199,7 +199,7 @@ async def test_write_file(app: FastAPI, client: AsyncClient, config, tmpdir) ->
config.settings.Server.projects_path = str(tmpdir)
project = ProjectManager.instance().create_project(project_id="01010203-0405-0607-0809-0a0b0c0d0e0b")
response = await client.post(app.url_path_for("write_compute_project_file",
response = await client.post(app.url_path_for("compute:write_compute_project_file",
project_id=project.id,
file_path="hello"), content=b"world")
assert response.status_code == status.HTTP_204_NO_CONTENT
@ -207,7 +207,7 @@ async def test_write_file(app: FastAPI, client: AsyncClient, config, tmpdir) ->
with open(os.path.join(project.path, "hello")) as f:
assert f.read() == "world"
response = await client.post(app.url_path_for("write_compute_project_file",
response = await client.post(app.url_path_for("compute:write_compute_project_file",
project_id=project.id,
file_path="../hello"))
assert response.status_code == status.HTTP_404_NOT_FOUND
@ -219,7 +219,7 @@ async def test_write_file_forbidden_location(app: FastAPI, client: AsyncClient,
project = ProjectManager.instance().create_project(project_id="01010203-0405-0607-0809-0a0b0c0d0e0b")
file_path = "%2e%2e/hello"
response = await client.post(app.url_path_for("write_compute_project_file",
response = await client.post(app.url_path_for("compute:write_compute_project_file",
project_id=project.id,
file_path=file_path), content=b"world")
assert response.status_code == status.HTTP_403_FORBIDDEN

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

@ -66,7 +66,7 @@ def base_params(tmpdir, fake_qemu_bin) -> dict:
@pytest.fixture
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> None:
response = await client.post(app.url_path_for("create_qemu_node", project_id=compute_project.id), json=base_params)
response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
return response.json()
@ -77,7 +77,7 @@ async def test_qemu_create(app: FastAPI,
base_params: dict,
fake_qemu_bin: str) -> None:
response = await client.post(app.url_path_for("create_qemu_node", project_id=compute_project.id), json=base_params)
response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@ -93,7 +93,7 @@ async def test_qemu_create_platform(app: FastAPI,
base_params["qemu_path"] = None
base_params["platform"] = "x86_64"
response = await client.post(app.url_path_for("create_qemu_node", project_id=compute_project.id), json=base_params)
response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@ -111,7 +111,7 @@ async def test_qemu_create_with_params(app: FastAPI,
params = base_params
params["ram"] = 1024
params["hda_disk_image"] = "linux载.img"
response = await client.post(app.url_path_for("create_qemu_node", project_id=compute_project.id), json=params)
response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@ -127,13 +127,13 @@ async def test_qemu_create_with_project_file(app: FastAPI,
base_params: dict,
fake_qemu_vm: str) -> None:
response = await client.post(app.url_path_for("write_compute_project_file",
response = await client.post(app.url_path_for("compute:write_compute_project_file",
project_id=compute_project.id,
file_path="hello.img"), content=b"world")
assert response.status_code == status.HTTP_204_NO_CONTENT
params = base_params
params["hda_disk_image"] = "hello.img"
response = await client.post(app.url_path_for("create_qemu_node", project_id=compute_project.id), json=params)
response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["hda_disk_image"] == "hello.img"
assert response.json()["hda_disk_image_md5sum"] == "7d793037a0760186574b0282f2f435e7"
@ -141,7 +141,7 @@ async def test_qemu_create_with_project_file(app: FastAPI,
async def test_qemu_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict):
response = await client.get(app.url_path_for("get_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.get(app.url_path_for("compute:get_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@ -154,7 +154,7 @@ async def test_qemu_get(app: FastAPI, client: AsyncClient, compute_project: Proj
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 client.post(app.url_path_for("start_qemu_node",
response = await client.post(app.url_path_for("compute:start_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -164,7 +164,7 @@ async def test_qemu_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
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 client.post(app.url_path_for("stop_qemu_node",
response = await client.post(app.url_path_for("compute:stop_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -174,7 +174,7 @@ async def test_qemu_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
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 client.post(app.url_path_for("reload_qemu_node",
response = await client.post(app.url_path_for("compute:reload_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -184,7 +184,7 @@ async def test_qemu_reload(app: FastAPI, client: AsyncClient, vm) -> None:
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 client.post(app.url_path_for("suspend_qemu_node",
response = await client.post(app.url_path_for("compute:suspend_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -194,7 +194,7 @@ async def test_qemu_suspend(app: FastAPI, client: AsyncClient, vm: dict) -> None
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 client.post(app.url_path_for("resume_qemu_node",
response = await client.post(app.url_path_for("compute:resume_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -204,7 +204,7 @@ async def test_qemu_resume(app: FastAPI, client: AsyncClient, vm: dict) -> None:
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 client.delete(app.url_path_for("delete_qemu_node",
response = await client.delete(app.url_path_for("compute:delete_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -224,7 +224,7 @@ async def test_qemu_update(app: FastAPI,
"hdb_disk_image": "linux载.img"
}
response = await client.put(app.url_path_for("update_qemu_node",
response = await client.put(app.url_path_for("compute:update_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
@ -244,11 +244,11 @@ async def test_qemu_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict)
}
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.add_ubridge_udp_connection"):
await client.put(app.url_path_for("update_qemu_node",
await client.put(app.url_path_for("compute: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",
url = app.url_path_for("compute:create_qemu_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
@ -267,11 +267,11 @@ async def test_qemu_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict)
"rhost": "127.0.0.1"
}
await client.put(app.url_path_for("update_qemu_node",
await client.put(app.url_path_for("compute: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",
url = app.url_path_for("compute:create_qemu_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
@ -281,7 +281,7 @@ async def test_qemu_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict)
params["filters"] = {}
url = app.url_path_for("update_qemu_node_nio",
url = app.url_path_for("compute:update_qemu_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
@ -301,18 +301,18 @@ async def test_qemu_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> N
}
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM._ubridge_send"):
await client.put(app.url_path_for("update_qemu_node",
await client.put(app.url_path_for("compute: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",
url = app.url_path_for("compute: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",
url = app.url_path_for("compute:delete_qemu_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
@ -327,7 +327,7 @@ async def test_qemu_list_binaries(app: FastAPI, client: AsyncClient, vm: dict) -
{"path": "/tmp/2", "version": "2.1.0"}]
with asyncio_patch("gns3server.compute.qemu.Qemu.binary_list", return_value=ret) as mock:
response = await client.get(app.url_path_for("get_qemu_binaries"))
response = await client.get(app.url_path_for("compute:get_qemu_binaries"))
assert mock.called_with(None)
assert response.status_code == status.HTTP_200_OK
assert response.json() == ret
@ -342,7 +342,7 @@ async def test_qemu_list_binaries(app: FastAPI, client: AsyncClient, vm: dict) -
# ]
#
# with asyncio_patch("gns3server.compute.qemu.Qemu.binary_list", return_value=ret) as mock:
# response = await client.get(app.url_path_for("get_qemu_binaries"),
# response = await client.get(app.url_path_for("compute:get_qemu_binaries"),
# json={"archs": ["i386"]})
# assert response.status_code == status.HTTP_200_OK
# assert mock.called_with(["i386"])
@ -351,7 +351,7 @@ async def test_qemu_list_binaries(app: FastAPI, client: AsyncClient, vm: dict) -
async def test_images(app: FastAPI, client: AsyncClient, fake_qemu_vm) -> None:
response = await client.get(app.url_path_for("get_qemu_images"))
response = await client.get(app.url_path_for("compute:get_qemu_images"))
assert response.status_code == status.HTTP_200_OK
assert {"filename": "linux载.img", "path": "linux载.img", "md5sum": "c4ca4238a0b923820dcc509a6f75849b", "filesize": 1} in response.json()
@ -360,7 +360,7 @@ async def test_upload_image(app: FastAPI, client: AsyncClient, tmpdir: str) -> N
with patch("gns3server.compute.Qemu.get_images_directory", return_value=str(tmpdir)):
response = await client.post(app.url_path_for("upload_qemu_image",
response = await client.post(app.url_path_for("compute:upload_qemu_image",
filename="test2使"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
@ -376,7 +376,7 @@ async def test_upload_image_ova(app: FastAPI, client: AsyncClient, tmpdir:str) -
with patch("gns3server.compute.Qemu.get_images_directory", return_value=str(tmpdir)):
response = await client.post(app.url_path_for("upload_qemu_image",
response = await client.post(app.url_path_for("compute:upload_qemu_image",
filename="test2.ova/test2.vmdk"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
@ -390,24 +390,24 @@ async def test_upload_image_ova(app: FastAPI, client: AsyncClient, tmpdir:str) -
async def test_upload_image_forbidden_location(app: FastAPI, client: AsyncClient, tmpdir: str) -> None:
response = await client.post(app.url_path_for("upload_qemu_image",
response = await client.post(app.url_path_for("compute:upload_qemu_image",
filename="/qemu/images/../../test2"), content=b"TEST")
assert response.status_code == status.HTTP_403_FORBIDDEN
async def test_download_image(app: FastAPI, client: AsyncClient, images_dir: str) -> None:
response = await client.post(app.url_path_for("upload_qemu_image", filename="test3"), content=b"TEST")
response = await client.post(app.url_path_for("compute:upload_qemu_image", filename="test3"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
response = await client.get(app.url_path_for("download_qemu_image", filename="test3"))
response = await client.get(app.url_path_for("compute:download_qemu_image", filename="test3"))
assert response.status_code == status.HTTP_200_OK
async def test_download_image_forbidden_location(app: FastAPI, client: AsyncClient, tmpdir) -> None:
file_path = "foo/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/etc/passwd"
response = await client.get(app.url_path_for("download_qemu_image", filename=file_path))
response = await client.get(app.url_path_for("compute:download_qemu_image", filename=file_path))
assert response.status_code == status.HTTP_403_FORBIDDEN
@ -418,7 +418,7 @@ async def test_upload_image_permission_denied(app: FastAPI, client: AsyncClient,
f.write("")
os.chmod(os.path.join(images_dir, "QEMU", "test2.tmp"), 0)
response = await client.post(app.url_path_for("upload_qemu_image", filename="test2"), content=b"TEST")
response = await client.post(app.url_path_for("compute:upload_qemu_image", filename="test2"), content=b"TEST")
assert response.status_code == status.HTTP_409_CONFLICT
@ -436,7 +436,7 @@ async def test_create_img_relative(app: FastAPI, client: AsyncClient):
"size": 100
}
with asyncio_patch("gns3server.compute.Qemu.create_disk"):
response = await client.post(app.url_path_for("create_qemu_image"), json=params)
response = await client.post(app.url_path_for("compute:create_qemu_image"), json=params)
assert response.status_code == status.HTTP_204_NO_CONTENT
@ -454,7 +454,7 @@ async def test_create_img_absolute_non_local(app: FastAPI, client: AsyncClient,
"size": 100
}
with asyncio_patch("gns3server.compute.Qemu.create_disk"):
response = await client.post(app.url_path_for("create_qemu_image"), json=params)
response = await client.post(app.url_path_for("compute:create_qemu_image"), json=params)
assert response.status_code == 403
@ -471,14 +471,14 @@ async def test_create_img_absolute_local(app: FastAPI, client: AsyncClient, conf
"size": 100
}
with asyncio_patch("gns3server.compute.Qemu.create_disk"):
response = await client.post(app.url_path_for("create_qemu_image"), json=params)
response = await client.post(app.url_path_for("compute:create_qemu_image"), json=params)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_capabilities(app: FastAPI, client: AsyncClient) -> None:
with asyncio_patch("gns3server.compute.Qemu.get_kvm_archs", return_value=["x86_64"]):
response = await client.get(app.url_path_for("get_qemu_capabilities"))
response = await client.get(app.url_path_for("compute:get_qemu_capabilities"))
assert response.json()["kvm"] == ["x86_64"]
@ -489,12 +489,12 @@ async def test_qemu_duplicate(app: FastAPI,
base_params: dict) -> None:
# create destination node first
response = await client.post(app.url_path_for("create_qemu_node",
response = await client.post(app.url_path_for("compute:create_qemu_node",
project_id=vm["project_id"]), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
params = {"destination_node_id": response.json()["node_id"]}
response = await client.post(app.url_path_for("duplicate_qemu_node",
response = await client.post(app.url_path_for("compute:duplicate_qemu_node",
project_id=vm["project_id"], node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_201_CREATED
@ -507,7 +507,7 @@ async def test_qemu_start_capture(app: FastAPI, client: AsyncClient, vm):
"data_link_type": "DLT_EN10MB"
}
url = app.url_path_for("start_qemu_node_capture",
url = app.url_path_for("compute:start_qemu_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -524,7 +524,7 @@ async def test_qemu_start_capture(app: FastAPI, client: AsyncClient, vm):
@pytest.mark.asyncio
async def test_qemu_stop_capture(app: FastAPI, client: AsyncClient, vm):
url = app.url_path_for("stop_qemu_node_capture",
url = app.url_path_for("compute:stop_qemu_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",

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

@ -38,7 +38,7 @@ async def vm(app: FastAPI, client: AsyncClient, compute_project: Project) -> Non
}
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.create", return_value=True) as mock:
response = await client.post(app.url_path_for("create_virtualbox_node", project_id=compute_project.id),
response = await client.post(app.url_path_for("compute:create_virtualbox_node", project_id=compute_project.id),
json=params)
assert mock.called
assert response.status_code == status.HTTP_201_CREATED
@ -56,7 +56,7 @@ async def test_vbox_create(app: FastAPI, client: AsyncClient, compute_project: P
}
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.create", return_value=True):
response = await client.post(app.url_path_for("create_virtualbox_node", project_id=compute_project.id),
response = await client.post(app.url_path_for("compute:create_virtualbox_node", project_id=compute_project.id),
json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "VM1"
@ -65,7 +65,7 @@ async def test_vbox_create(app: FastAPI, client: AsyncClient, compute_project: P
async def test_vbox_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await client.get(app.url_path_for("get_virtualbox_node",
response = await client.get(app.url_path_for("compute:get_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert response.status_code == status.HTTP_200_OK
@ -77,7 +77,7 @@ async def test_vbox_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.start", return_value=True) as mock:
response = await client.post(app.url_path_for("start_virtualbox_node",
response = await client.post(app.url_path_for("compute:start_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -87,7 +87,7 @@ async def test_vbox_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
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 client.post(app.url_path_for("stop_virtualbox_node",
response = await client.post(app.url_path_for("compute:stop_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -97,7 +97,7 @@ async def test_vbox_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
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 client.post(app.url_path_for("suspend_virtualbox_node",
response = await client.post(app.url_path_for("compute:suspend_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -107,7 +107,7 @@ async def test_vbox_suspend(app: FastAPI, client: AsyncClient, vm: dict) -> None
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 client.post(app.url_path_for("resume_virtualbox_node",
response = await client.post(app.url_path_for("compute:resume_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -117,7 +117,7 @@ async def test_vbox_resume(app: FastAPI, client: AsyncClient, vm: dict) -> None:
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 client.post(app.url_path_for("reload_virtualbox_node",
response = await client.post(app.url_path_for("compute:reload_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -133,7 +133,7 @@ async def test_vbox_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict)
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_virtualbox_node_nio",
url = app.url_path_for("compute:create_virtualbox_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -170,7 +170,7 @@ async def test_vbox_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict)
async def test_vbox_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("delete_virtualbox_node_nio",
url = app.url_path_for("compute:delete_virtualbox_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -192,7 +192,7 @@ async def test_vbox_update(app: FastAPI, client: AsyncClient, vm, free_console_p
"console": free_console_port
}
response = await client.put(app.url_path_for("update_virtualbox_node",
response = await client.put(app.url_path_for("compute:update_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
@ -208,7 +208,7 @@ async def test_virtualbox_start_capture(app: FastAPI, client: AsyncClient, vm):
"data_link_type": "DLT_EN10MB"
}
url = app.url_path_for("start_virtualbox_node_capture",
url = app.url_path_for("compute:start_virtualbox_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -225,7 +225,7 @@ async def test_virtualbox_start_capture(app: FastAPI, client: AsyncClient, vm):
@pytest.mark.asyncio
async def test_virtualbox_stop_capture(app: FastAPI, client: AsyncClient, vm):
url = app.url_path_for("stop_virtualbox_node_capture",
url = app.url_path_for("compute:stop_virtualbox_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",

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

@ -37,7 +37,7 @@ async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, vmx_pa
}
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.create", return_value=True) as mock:
response = await client.post(app.url_path_for("create_vmware_node", project_id=compute_project.id),
response = await client.post(app.url_path_for("compute:create_vmware_node", project_id=compute_project.id),
json=params)
assert mock.called
assert response.status_code == status.HTTP_201_CREATED
@ -65,7 +65,7 @@ async def test_vmware_create(app: FastAPI, client: AsyncClient, compute_project:
}
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.create", return_value=True):
response = await client.post(app.url_path_for("create_vmware_node", project_id=compute_project.id),
response = await client.post(app.url_path_for("compute:create_vmware_node", project_id=compute_project.id),
json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "VM1"
@ -74,7 +74,7 @@ async def test_vmware_create(app: FastAPI, client: AsyncClient, compute_project:
async def test_vmware_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await client.get(app.url_path_for("get_vmware_node", project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.get(app.url_path_for("compute:get_vmware_node", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "VMTEST"
assert response.json()["project_id"] == compute_project.id
@ -83,7 +83,7 @@ async def test_vmware_get(app: FastAPI, client: AsyncClient, compute_project: Pr
async def test_vmware_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.start", return_value=True) as mock:
response = await client.post(app.url_path_for("start_vmware_node",
response = await client.post(app.url_path_for("compute:start_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -93,7 +93,7 @@ async def test_vmware_start(app: FastAPI, client: AsyncClient, vm: dict) -> None
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 client.post(app.url_path_for("stop_vmware_node",
response = await client.post(app.url_path_for("compute:stop_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -103,7 +103,7 @@ async def test_vmware_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
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 client.post(app.url_path_for("suspend_vmware_node",
response = await client.post(app.url_path_for("compute:suspend_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -113,7 +113,7 @@ async def test_vmware_suspend(app: FastAPI, client: AsyncClient, vm: dict) -> No
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 client.post(app.url_path_for("resume_vmware_node",
response = await client.post(app.url_path_for("compute:resume_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -123,7 +123,7 @@ async def test_vmware_resume(app: FastAPI, client: AsyncClient, vm: dict) -> Non
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 client.post(app.url_path_for("reload_vmware_node",
response = await client.post(app.url_path_for("compute:reload_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -139,7 +139,7 @@ async def test_vmware_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_vmware_node_nio",
url = app.url_path_for("compute:create_vmware_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -176,7 +176,7 @@ async def test_vmware_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict
async def test_vmware_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("delete_vmware_node_nio",
url = app.url_path_for("compute:delete_vmware_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -198,7 +198,7 @@ async def test_vmware_update(app: FastAPI, client: AsyncClient, vm: dict, free_c
"console": free_console_port
}
response = await client.put(app.url_path_for("update_vmware_node",
response = await client.put(app.url_path_for("compute:update_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
@ -213,7 +213,7 @@ async def test_vmware_start_capture(app: FastAPI, client: AsyncClient, vm: dict)
"data_link_type": "DLT_EN10MB"
}
url = app.url_path_for("start_vmware_node_capture",
url = app.url_path_for("compute:start_vmware_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -230,7 +230,7 @@ async def test_vmware_start_capture(app: FastAPI, client: AsyncClient, vm: dict)
async def test_vmware_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("stop_vmware_node_capture",
url = app.url_path_for("compute:stop_vmware_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",

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

@ -31,7 +31,7 @@ pytestmark = pytest.mark.asyncio
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
params = {"name": "PC TEST 1"}
response = await client.post(app.url_path_for("create_vpcs_node", project_id=compute_project.id), json=params)
response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
return response.json()
@ -39,7 +39,7 @@ async def vm(app: FastAPI, client: AsyncClient, compute_project: Project) -> Non
async def test_vpcs_create(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
params = {"name": "PC TEST 1"}
response = await client.post(app.url_path_for("create_vpcs_node", project_id=compute_project.id), json=params)
response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@ -47,7 +47,7 @@ async def test_vpcs_create(app: FastAPI, client: AsyncClient, compute_project: P
async def test_vpcs_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await client.get(app.url_path_for("get_vpcs_node", project_id=vm["project_id"], node_id=vm["node_id"]))
response = await client.get(app.url_path_for("compute:get_vpcs_node", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@ -61,7 +61,7 @@ async def test_vpcs_create_startup_script(app: FastAPI, client: AsyncClient, com
"startup_script": "ip 192.168.1.2\necho TEST"
}
response = await client.post(app.url_path_for("create_vpcs_node", project_id=compute_project.id), json=params)
response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@ -77,7 +77,7 @@ async def test_vpcs_create_port(app: FastAPI,
"console": free_console_port
}
response = await client.post(app.url_path_for("create_vpcs_node", project_id=compute_project.id), json=params)
response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@ -93,7 +93,7 @@ async def test_vpcs_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict)
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_vpcs_node_nio",
url = app.url_path_for("compute:create_vpcs_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -114,7 +114,7 @@ async def test_vpcs_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict)
"rhost": "127.0.0.1"
}
url = app.url_path_for("create_vpcs_node_nio",
url = app.url_path_for("compute:create_vpcs_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -125,7 +125,7 @@ async def test_vpcs_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict)
assert response.status_code == status.HTTP_201_CREATED
params["filters"] = {}
url = app.url_path_for("update_vpcs_node_nio",
url = app.url_path_for("compute:update_vpcs_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -145,14 +145,14 @@ async def test_vpcs_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> N
}
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM._ubridge_send"):
url = app.url_path_for("create_vpcs_node_nio",
url = app.url_path_for("compute: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",
url = app.url_path_for("compute:delete_vpcs_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -164,7 +164,7 @@ async def test_vpcs_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> N
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 client.post(app.url_path_for("start_vpcs_node",
response = await client.post(app.url_path_for("compute:start_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -175,7 +175,7 @@ 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 client.post(app.url_path_for("stop_vpcs_node",
response = await client.post(app.url_path_for("compute:stop_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -185,7 +185,7 @@ async def test_vpcs_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
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 client.post(app.url_path_for("reload_vpcs_node",
response = await client.post(app.url_path_for("compute:reload_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -195,7 +195,7 @@ async def test_vpcs_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None:
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 client.delete(app.url_path_for("delete_vpcs_node",
response = await client.delete(app.url_path_for("compute:delete_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@ -206,11 +206,11 @@ async def test_vpcs_duplicate(app: FastAPI, client: AsyncClient, compute_project
# create destination node first
params = {"name": "PC TEST 1"}
response = await client.post(app.url_path_for("create_vpcs_node", project_id=compute_project.id), json=params)
response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
params = {"destination_node_id": response.json()["node_id"]}
response = await client.post(app.url_path_for("duplicate_vpcs_node",
response = await client.post(app.url_path_for("compute:duplicate_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_201_CREATED
@ -224,7 +224,7 @@ async def test_vpcs_update(app: FastAPI, client: AsyncClient, vm: dict, free_con
"console": console_port
}
response = await client.put(app.url_path_for("update_vpcs_node",
response = await client.put(app.url_path_for("compute:update_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
@ -239,7 +239,7 @@ async def test_vpcs_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -
"data_link_type": "DLT_EN10MB"
}
url = app.url_path_for("start_vpcs_node_capture",
url = app.url_path_for("compute:start_vpcs_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@ -255,7 +255,7 @@ async def test_vpcs_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -
async def test_vpcs_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("stop_vpcs_node_capture",
url = app.url_path_for("compute:stop_vpcs_node_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",

2
tests/api/routes/controller/test_version.py
View File

@ -29,7 +29,7 @@ async def test_version_output(app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("get_version"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == {'local': True, 'version': __version__}
assert response.json() == {'controller_host': None, 'local': True, 'version': __version__}
async def test_version_input(app: FastAPI, client: AsyncClient) -> None: