NgTech-LTD/gns3-server
2
0 Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-11-03 05:51:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

Secure controller to compute communication using HTTP basic authentication

Browse Source
This commit is contained in:
grossmj committed 2021-11-18 18:37:10 +10:30
1 parent 55e50dae4b
commit 38388f7ae5
24 files changed
+483 -416
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/api/routes/compute/test_capabilities.py
+4 -4
View File
@@ -30,9 +30,9 @@ pytestmark = pytest.mark.asyncio
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
async def test_get(app: FastAPI, client: AsyncClient, windows_platform) -> None:
async def test_get(app: FastAPI, compute_client: AsyncClient, windows_platform) -> None:
response = await client.get(app.url_path_for("compute:get_capabilities"))
response = await compute_client.get(app.url_path_for("compute:get_capabilities"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == {'node_types': ['cloud', 'ethernet_hub', 'ethernet_switch', 'nat', 'vpcs', 'virtualbox', 'dynamips', 'frame_relay_switch', 'atm_switch', 'qemu', 'vmware', 'docker', 'iou'],
'version': __version__,
@@ -44,9 +44,9 @@ async def test_get(app: FastAPI, client: AsyncClient, windows_platform) -> None:
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
async def test_get_on_gns3vm(app: FastAPI, client: AsyncClient, on_gns3vm) -> None:
async def test_get_on_gns3vm(app: FastAPI, compute_client: AsyncClient, on_gns3vm) -> None:
response = await client.get(app.url_path_for("compute:get_capabilities"))
response = await compute_client.get(app.url_path_for("compute:get_capabilities"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == {'node_types': ['cloud', 'ethernet_hub', 'ethernet_switch', 'nat', 'vpcs', 'virtualbox', 'dynamips', 'frame_relay_switch', 'atm_switch', 'qemu', 'vmware', 'docker', 'iou'],
'version': __version__,
tests/api/routes/compute/test_cloud_nodes.py
+22 -22
View File
@@ -28,35 +28,35 @@ pytestmark = pytest.mark.asyncio
@pytest.fixture(scope="function")
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, on_gns3vm) -> dict:
async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project, on_gns3vm) -> dict:
with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._start_ubridge"):
response = await client.post(app.url_path_for("compute:create_cloud", project_id=compute_project.id),
response = await compute_client.post(app.url_path_for("compute:create_cloud", project_id=compute_project.id),
json={"name": "Cloud 1"})
assert response.status_code == status.HTTP_201_CREATED
return response.json()
async def test_cloud_create(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
async def test_cloud_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None:
with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._start_ubridge"):
response = await client.post(app.url_path_for("compute:create_cloud", project_id=compute_project.id),
response = await compute_client.post(app.url_path_for("compute:create_cloud", project_id=compute_project.id),
json={"name": "Cloud 1"})
assert response.status_code == 201
assert response.json()["name"] == "Cloud 1"
assert response.json()["project_id"] == compute_project.id
async def test_get_cloud(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
async def test_get_cloud(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await client.get(app.url_path_for("compute:get_cloud", project_id=vm["project_id"], node_id=vm["node_id"]))
response = await compute_client.get(app.url_path_for("compute:get_cloud", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "Cloud 1"
assert response.json()["project_id"] == compute_project.id
assert response.json()["status"] == "started"
async def test_cloud_nio_create_udp(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
async def test_cloud_nio_create_udp(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None:
params = {"type": "nio_udp",
"lport": 4242,
@@ -68,12 +68,12 @@ async def test_cloud_nio_create_udp(app: FastAPI, client: AsyncClient, compute_p
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
async def test_cloud_nio_update_udp(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
async def test_cloud_nio_update_udp(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None:
params = {"type": "nio_udp",
"lport": 4242,
@@ -85,7 +85,7 @@ async def test_cloud_nio_update_udp(app: FastAPI, client: AsyncClient, compute_p
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
await client.post(url, json=params)
await compute_client.post(url, json=params)
params["filters"] = {}
url = app.url_path_for("compute:create_cloud_nio",
@@ -93,12 +93,12 @@ async def test_cloud_nio_update_udp(app: FastAPI, client: AsyncClient, compute_p
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
response = await client.put(url, json=params)
response = await compute_client.put(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
async def test_cloud_delete_nio(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
async def test_cloud_delete_nio(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None:
params = {"type": "nio_udp",
"lport": 4242,
@@ -110,7 +110,7 @@ async def test_cloud_delete_nio(app: FastAPI, client: AsyncClient, compute_proje
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
await client.post(url, json=params)
await compute_client.post(url, json=params)
url = app.url_path_for("compute:delete_cloud_nio",
project_id=vm["project_id"],
@@ -118,25 +118,25 @@ async def test_cloud_delete_nio(app: FastAPI, client: AsyncClient, compute_proje
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._start_ubridge"):
response = await client.delete(url)
response = await compute_client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_cloud_delete(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
async def test_cloud_delete(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await client.delete(app.url_path_for("compute:delete_cloud", project_id=vm["project_id"], node_id=vm["node_id"]))
response = await compute_client.delete(app.url_path_for("compute:delete_cloud", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_cloud_update(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_cloud_update(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
response = await client.put(app.url_path_for("compute:update_cloud", project_id=vm["project_id"], node_id=vm["node_id"]),
response = await compute_client.put(app.url_path_for("compute:update_cloud", project_id=vm["project_id"], node_id=vm["node_id"]),
json={"name": "test"})
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "test"
async def test_cloud_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_cloud_start_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"capture_file_name": "test.pcap",
@@ -144,7 +144,7 @@ async def test_cloud_start_capture(app: FastAPI, client: AsyncClient, vm: dict)
}
with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud.start_capture") as mock:
response = await client.post(app.url_path_for("compute:start_cloud_capture",
response = await compute_client.post(app.url_path_for("compute:start_cloud_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
@@ -155,10 +155,10 @@ async def test_cloud_start_capture(app: FastAPI, client: AsyncClient, vm: dict)
assert "test.pcap" in response.json()["pcap_file_path"]
async def test_cloud_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_cloud_stop_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud.stop_capture") as mock:
response = await client.post(app.url_path_for("compute:stop_cloud_capture",
response = await compute_client.post(app.url_path_for("compute:stop_cloud_capture",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
tests/api/routes/compute/test_compute.py
+14 -8
View File
@@ -27,27 +27,33 @@ from gns3server.compute.project import Project
pytestmark = pytest.mark.asyncio
async def test_udp_allocation(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
async def test_udp_allocation(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None:
response = await client.post(app.url_path_for("compute:allocate_udp_port", project_id=compute_project.id), json={})
response = await compute_client.post(app.url_path_for("compute:allocate_udp_port", project_id=compute_project.id), json={})
assert response.status_code == status.HTTP_201_CREATED
assert response.json()['udp_port'] is not None
async def test_interfaces(app: FastAPI, client: AsyncClient) -> None:
async def test_interfaces(app: FastAPI, compute_client: AsyncClient) -> None:
response = await client.get(app.url_path_for("compute:network_interfaces"))
response = await compute_client.get(app.url_path_for("compute:network_interfaces"))
assert response.status_code == status.HTTP_200_OK
assert isinstance(response.json(), list)
async def test_version_output(app: FastAPI, client: AsyncClient) -> None:
async def test_version_output(app: FastAPI, compute_client: AsyncClient) -> None:
response = await client.get(app.url_path_for("compute:compute_version"))
response = await compute_client.get(app.url_path_for("compute:compute_version"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == {'local': True, 'version': __version__}
async def test_compute_authentication(app: FastAPI, compute_client: AsyncClient) -> None:
response = await compute_client.get(app.url_path_for("compute:compute_version"), auth=("admin", "invalid_password"))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
# @pytest.mark.asyncio
# async def test_debug_output(compute_api):
#
@@ -55,7 +61,7 @@ async def test_version_output(app: FastAPI, client: AsyncClient) -> None:
# assert response.status_code == 200
async def test_statistics_output(app: FastAPI, client: AsyncClient) -> None:
async def test_statistics_output(app: FastAPI, compute_client: AsyncClient) -> None:
response = await client.get(app.url_path_for("compute:compute_statistics"))
response = await compute_client.get(app.url_path_for("compute:compute_statistics"))
assert response.status_code == status.HTTP_200_OK
tests/api/routes/compute/test_docker_nodes.py
+32 -32
View File
@@ -57,22 +57,22 @@ def base_params() -> dict:
@pytest.fixture
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> dict:
async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project, base_params: dict) -> dict:
with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "nginx"}]):
with asyncio_patch("gns3server.compute.docker.Docker.query", return_value={"Id": "8bd8153ea8f5"}):
with asyncio_patch("gns3server.compute.docker.DockerVM._get_container_state", return_value="exited"):
response = await client.post(app.url_path_for("compute:create_docker_node", project_id=compute_project.id),
response = await compute_client.post(app.url_path_for("compute:create_docker_node", project_id=compute_project.id),
json=base_params)
assert response.status_code == status.HTTP_201_CREATED
return response.json()
async def test_docker_create(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> None:
async def test_docker_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project, base_params: dict) -> None:
with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "nginx"}]):
with asyncio_patch("gns3server.compute.docker.Docker.query", return_value={"Id": "8bd8153ea8f5"}):
response = await client.post(app.url_path_for("compute:create_docker_node", project_id=compute_project.id),
response = await compute_client.post(app.url_path_for("compute:create_docker_node", project_id=compute_project.id),
json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
@@ -85,68 +85,68 @@ async def test_docker_create(app: FastAPI, client: AsyncClient, compute_project:
assert response.json()["extra_hosts"] == "test:127.0.0.1"
async def test_docker_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_docker_start(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.start", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:start_docker_node",
response = await compute_client.post(app.url_path_for("compute:start_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_docker_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_docker_stop(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.stop", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:stop_docker_node",
response = await compute_client.post(app.url_path_for("compute:stop_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_docker_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_docker_reload(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.restart", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:reload_docker_node",
response = await compute_client.post(app.url_path_for("compute:reload_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_docker_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_docker_delete(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.delete", return_value=True) as mock:
response = await client.delete(app.url_path_for("compute:delete_docker_node",
response = await compute_client.delete(app.url_path_for("compute:delete_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_docker_pause(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_docker_pause(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.pause", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:pause_docker_node",
response = await compute_client.post(app.url_path_for("compute:pause_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_docker_unpause(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_docker_unpause(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.unpause", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:unpause_docker_node",
response = await compute_client.post(app.url_path_for("compute:unpause_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_docker_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_docker_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -159,12 +159,12 @@ async def test_docker_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
async def test_docker_update_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_docker_update_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -178,7 +178,7 @@ async def test_docker_update_nio(app: FastAPI, client: AsyncClient, vm: dict) ->
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
url = app.url_path_for("compute:update_docker_node_nio",
@@ -187,11 +187,11 @@ async def test_docker_update_nio(app: FastAPI, client: AsyncClient, vm: dict) ->
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.adapter_update_nio_binding"):
response = await client.put(url, json=params)
response = await compute_client.put(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
async def test_docker_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_docker_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("compute:delete_docker_node_nio",
project_id=vm["project_id"],
@@ -199,11 +199,11 @@ async def test_docker_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) ->
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.adapter_remove_nio_binding"):
response = await client.delete(url)
response = await compute_client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_docker_update(app: FastAPI, client: AsyncClient, vm: dict, free_console_port: int) -> None:
async def test_docker_update(app: FastAPI, compute_client: AsyncClient, vm: dict, free_console_port: int) -> None:
params = {
"name": "test",
@@ -214,7 +214,7 @@ async def test_docker_update(app: FastAPI, client: AsyncClient, vm: dict, free_c
}
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.update") as mock:
response = await client.put(app.url_path_for("compute:update_docker_node",
response = await compute_client.put(app.url_path_for("compute:update_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
@@ -227,7 +227,7 @@ async def test_docker_update(app: FastAPI, client: AsyncClient, vm: dict, free_c
assert response.json()["extra_hosts"] == "test:127.0.0.1"
async def test_docker_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_docker_start_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("compute:start_docker_node_capture",
project_id=vm["project_id"],
@@ -238,13 +238,13 @@ async def test_docker_start_capture(app: FastAPI, client: AsyncClient, vm: dict)
with patch("gns3server.compute.docker.docker_vm.DockerVM.is_running", return_value=True):
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.start_capture") as mock:
params = {"capture_file_name": "test.pcap", "data_link_type": "DLT_EN10MB"}
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_200_OK
assert mock.called
assert "test.pcap" in response.json()["pcap_file_path"]
async def test_docker_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_docker_stop_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("compute:stop_docker_node_capture",
project_id=vm["project_id"],
@@ -254,22 +254,22 @@ async def test_docker_stop_capture(app: FastAPI, client: AsyncClient, vm: dict)
with patch("gns3server.compute.docker.docker_vm.DockerVM.is_running", return_value=True):
with asyncio_patch("gns3server.compute.docker.docker_vm.DockerVM.stop_capture") as mock:
response = await client.post(url)
response = await compute_client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
async def test_docker_duplicate(app: FastAPI, client: AsyncClient, vm: dict, base_params: dict) -> None:
async def test_docker_duplicate(app: FastAPI, compute_client: AsyncClient, vm: dict, base_params: dict) -> None:
# create destination node first
with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "nginx"}]):
with asyncio_patch("gns3server.compute.docker.Docker.query", return_value={"Id": "8bd8153ea8f5"}):
response = await client.post(app.url_path_for("compute:create_docker_node",
response = await compute_client.post(app.url_path_for("compute:create_docker_node",
project_id=vm["project_id"]), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
params = {"destination_node_id": response.json()["node_id"]}
response = await client.post(app.url_path_for("compute:duplicate_docker_node",
response = await compute_client.post(app.url_path_for("compute:duplicate_docker_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_201_CREATED
tests/api/routes/compute/test_dynamips_nodes.py
+13 -13
View File
@@ -162,10 +162,10 @@ def fake_file(tmpdir) -> str:
return path
async def test_images(app: FastAPI, client: AsyncClient, tmpdir, fake_image: str, fake_file: str) -> None:
async def test_images(app: FastAPI, compute_client: AsyncClient, tmpdir, fake_image: str, fake_file: str) -> None:
with patch("gns3server.utils.images.default_images_directory", return_value=str(tmpdir)):
response = await client.get(app.url_path_for("compute:get_dynamips_images"))
response = await compute_client.get(app.url_path_for("compute:get_dynamips_images"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == [{"filename": "7200.bin",
"path": "7200.bin",
@@ -173,9 +173,9 @@ async def test_images(app: FastAPI, client: AsyncClient, tmpdir, fake_image: str
"md5sum": "b0d5aa897d937aced5a6b1046e8f7e2e"}]
async def test_upload_image(app: FastAPI, client: AsyncClient, images_dir: str) -> None:
async def test_upload_image(app: FastAPI, compute_client: AsyncClient, images_dir: str) -> None:
response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename="test2"), content=b"TEST")
response = await compute_client.post(app.url_path_for("compute:upload_dynamips_image", filename="test2"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
with open(os.path.join(images_dir, "IOS", "test2")) as f:
@@ -186,36 +186,36 @@ async def test_upload_image(app: FastAPI, client: AsyncClient, images_dir: str)
assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf"
async def test_upload_image_forbidden_location(app: FastAPI, client: AsyncClient) -> None:
async def test_upload_image_forbidden_location(app: FastAPI, compute_client: AsyncClient) -> None:
file_path = "%2e%2e/hello"
response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename=file_path), content=b"TEST")
response = await compute_client.post(app.url_path_for("compute:upload_dynamips_image", filename=file_path), content=b"TEST")
assert response.status_code == status.HTTP_403_FORBIDDEN
async def test_download_image(app: FastAPI, client: AsyncClient, images_dir: str) -> None:
async def test_download_image(app: FastAPI, compute_client: AsyncClient, images_dir: str) -> None:
response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename="test3"), content=b"TEST")
response = await compute_client.post(app.url_path_for("compute:upload_dynamips_image", filename="test3"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
response = await client.get(app.url_path_for("compute:download_dynamips_image", filename="test3"))
response = await compute_client.get(app.url_path_for("compute:download_dynamips_image", filename="test3"))
assert response.status_code == status.HTTP_200_OK
async def test_download_image_forbidden(app: FastAPI, client: AsyncClient, tmpdir) -> None:
async def test_download_image_forbidden(app: FastAPI, compute_client: AsyncClient, tmpdir) -> None:
file_path = "foo/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/etc/passwd"
response = await client.get(app.url_path_for("compute:download_dynamips_image", filename=file_path))
response = await compute_client.get(app.url_path_for("compute:download_dynamips_image", filename=file_path))
assert response.status_code == status.HTTP_403_FORBIDDEN
@pytest.mark.skipif(not sys.platform.startswith("win") and os.getuid() == 0, reason="Root can delete any image")
async def test_upload_image_permission_denied(app: FastAPI, client: AsyncClient, images_dir: str) -> None:
async def test_upload_image_permission_denied(app: FastAPI, compute_client: AsyncClient, images_dir: str) -> None:
os.makedirs(os.path.join(images_dir, "IOS"), exist_ok=True)
with open(os.path.join(images_dir, "IOS", "test2.tmp"), "w+") as f:
f.write("")
os.chmod(os.path.join(images_dir, "IOS", "test2.tmp"), 0)
response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename="test2"), content=b"TEST")
response = await compute_client.post(app.url_path_for("compute:upload_dynamips_image", filename="test2"), content=b"TEST")
assert response.status_code == status.HTTP_409_CONFLICT
tests/api/routes/compute/test_iou_nodes.py
+55 -55
View File
@@ -51,9 +51,9 @@ def base_params(tmpdir, fake_iou_bin) -> dict:
@pytest.fixture
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> dict:
async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project, base_params: dict) -> dict:
response = await client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=base_params)
response = await compute_client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
return response.json()
@@ -65,9 +65,9 @@ def startup_config_file(compute_project: Project, vm: dict) -> str:
return os.path.join(directory, "startup-config.cfg")
async def test_iou_create(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> None:
async def test_iou_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project, base_params: dict) -> None:
response = await client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=base_params)
response = await compute_client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@@ -79,7 +79,7 @@ async def test_iou_create(app: FastAPI, client: AsyncClient, compute_project: Pr
async def test_iou_create_with_params(app: FastAPI,
client: AsyncClient,
compute_client: AsyncClient,
compute_project: Project,
base_params: dict) -> None:
@@ -92,7 +92,7 @@ async def test_iou_create_with_params(app: FastAPI,
params["startup_config_content"] = "hostname test"
params["use_default_iou_values"] = False
response = await client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=params)
response = await compute_client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@@ -109,7 +109,7 @@ async def test_iou_create_with_params(app: FastAPI,
async def test_iou_create_startup_config_already_exist(
app: FastAPI,
client: AsyncClient,
compute_client: AsyncClient,
compute_project: Project,
base_params: dict) -> None:
"""We don't erase a startup-config if already exist at project creation"""
@@ -123,16 +123,16 @@ async def test_iou_create_startup_config_already_exist(
params["node_id"] = node_id
params["startup_config_content"] = "hostname test"
response = await client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=params)
response = await compute_client.post(app.url_path_for("compute:create_iou_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
with open(startup_config_file(compute_project, response.json())) as f:
assert f.read() == "echo hello"
async def test_iou_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
async def test_iou_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await client.get(app.url_path_for("compute:get_iou_node", project_id=vm["project_id"], node_id=vm["node_id"]))
response = await compute_client.get(app.url_path_for("compute:get_iou_node", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@@ -143,58 +143,58 @@ async def test_iou_get(app: FastAPI, client: AsyncClient, compute_project: Proje
assert response.json()["l1_keepalives"] is False
async def test_iou_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_iou_start(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.start", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:start_iou_node",
response = await compute_client.post(app.url_path_for("compute:start_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json={})
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_iou_start_with_iourc(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_iou_start_with_iourc(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {"iourc_content": "test"}
with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.start", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:start_iou_node",
response = await compute_client.post(app.url_path_for("compute:start_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_iou_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_iou_stop(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.stop", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:stop_iou_node",
response = await compute_client.post(app.url_path_for("compute:stop_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_iou_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_iou_reload(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.reload", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:reload_iou_node",
response = await compute_client.post(app.url_path_for("compute:reload_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_iou_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_iou_delete(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.iou.IOU.delete_node", return_value=True) as mock:
response = await client.delete(app.url_path_for("compute:delete_iou_node",
response = await compute_client.delete(app.url_path_for("compute:delete_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_iou_update(app: FastAPI, client: AsyncClient, vm: dict, free_console_port: int) -> None:
async def test_iou_update(app: FastAPI, compute_client: AsyncClient, vm: dict, free_console_port: int) -> None:
params = {
"name": "test",
@@ -207,7 +207,7 @@ async def test_iou_update(app: FastAPI, client: AsyncClient, vm: dict, free_cons
"use_default_iou_values": True,
}
response = await client.put(app.url_path_for("compute:update_iou_node",
response = await compute_client.put(app.url_path_for("compute:update_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
@@ -221,7 +221,7 @@ async def test_iou_update(app: FastAPI, client: AsyncClient, vm: dict, free_cons
assert response.json()["use_default_iou_values"] is True
async def test_iou_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_iou_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {"type": "nio_udp",
"lport": 4242,
@@ -233,12 +233,12 @@ async def test_iou_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
async def test_iou_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_iou_nio_update_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {"type": "nio_udp",
"lport": 4242,
@@ -251,7 +251,7 @@ async def test_iou_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -
adapter_number="1",
port_number="0")
await client.post(url, json=params)
await compute_client.post(url, json=params)
params["filters"] = {}
url = app.url_path_for("compute:update_iou_node_nio",
@@ -259,12 +259,12 @@ async def test_iou_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
response = await client.put(url, json=params)
response = await compute_client.put(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
async def test_iou_nio_create_ethernet(app: FastAPI, client: AsyncClient, vm: dict, ethernet_device: str) -> None:
async def test_iou_nio_create_ethernet(app: FastAPI, compute_client: AsyncClient, vm: dict, ethernet_device: str) -> None:
params = {
"type": "nio_ethernet",
@@ -277,14 +277,14 @@ async def test_iou_nio_create_ethernet(app: FastAPI, client: AsyncClient, vm: di
adapter_number="1",
port_number="0")
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_ethernet"
assert response.json()["ethernet_device"] == ethernet_device
async def test_iou_nio_create_ethernet_different_port(app: FastAPI,
client: AsyncClient,
compute_client: AsyncClient,
vm: dict,
ethernet_device: str) -> None:
@@ -298,13 +298,13 @@ async def test_iou_nio_create_ethernet_different_port(app: FastAPI,
node_id=vm["node_id"],
adapter_number="0",
port_number="3")
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_ethernet"
assert response.json()["ethernet_device"] == ethernet_device
async def test_iou_nio_create_tap(app: FastAPI, client: AsyncClient, vm: dict, ethernet_device: str) -> None:
async def test_iou_nio_create_tap(app: FastAPI, compute_client: AsyncClient, vm: dict, ethernet_device: str) -> None:
params = {
"type": "nio_tap",
@@ -317,12 +317,12 @@ async def test_iou_nio_create_tap(app: FastAPI, client: AsyncClient, vm: dict, e
adapter_number="1",
port_number="0")
with patch("gns3server.compute.base_manager.BaseManager.has_privileged_access", return_value=True):
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_tap"
async def test_iou_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_iou_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -337,7 +337,7 @@ async def test_iou_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> No
adapter_number="1",
port_number="0")
await client.post(url, json=params)
await compute_client.post(url, json=params)
url = app.url_path_for("compute:delete_iou_node_nio",
project_id=vm["project_id"],
@@ -345,11 +345,11 @@ async def test_iou_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> No
adapter_number="1",
port_number="0")
response = await client.delete(url)
response = await compute_client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_iou_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_iou_start_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"capture_file_name": "test.pcap",
@@ -364,13 +364,13 @@ async def test_iou_start_capture(app: FastAPI, client: AsyncClient, vm: dict) ->
with patch("gns3server.compute.iou.iou_vm.IOUVM.is_running", return_value=True):
with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.start_capture") as mock:
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_200_OK
assert mock.called
assert "test.pcap" in response.json()["pcap_file_path"]
async def test_iou_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_iou_stop_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("compute:stop_iou_node_capture",
project_id=vm["project_id"],
@@ -380,7 +380,7 @@ async def test_iou_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) ->
with patch("gns3server.compute.iou.iou_vm.IOUVM.is_running", return_value=True):
with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.stop_capture") as mock:
response = await client.post(url)
response = await compute_client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
@@ -390,21 +390,21 @@ async def test_iou_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) ->
#
# with asyncio_patch("gns3server.compute.iou.iou_vm.IOUVM.get_nio"):
# with asyncio_patch("gns3server.compute.iou.IOU.stream_pcap_file"):
# response = await client.get("/projects/{project_id}/iou/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True)
# response = await compute_client.get("/projects/{project_id}/iou/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True)
# assert response.status_code == status.HTTP_200_OK
async def test_images(app: FastAPI, client: AsyncClient, fake_iou_bin: str) -> None:
async def test_images(app: FastAPI, compute_client: AsyncClient, fake_iou_bin: str) -> None:
response = await client.get(app.url_path_for("compute:get_iou_images"))
response = await compute_client.get(app.url_path_for("compute:get_iou_images"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == [{"filename": "iou.bin", "path": "iou.bin", "filesize": 7, "md5sum": "e573e8f5c93c6c00783f20c7a170aa6c"}]
async def test_upload_image(app: FastAPI, client: AsyncClient, tmpdir) -> None:
async def test_upload_image(app: FastAPI, compute_client: AsyncClient, tmpdir) -> None:
with patch("gns3server.compute.IOU.get_images_directory", return_value=str(tmpdir)):
response = await client.post(app.url_path_for("compute:upload_iou_image", filename="test2"), content=b"TEST")
response = await compute_client.post(app.url_path_for("compute:upload_iou_image", filename="test2"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
with open(str(tmpdir / "test2")) as f:
@@ -415,38 +415,38 @@ async def test_upload_image(app: FastAPI, client: AsyncClient, tmpdir) -> None:
assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf"
async def test_upload_image_forbidden_location(app: FastAPI, client: AsyncClient) -> None:
async def test_upload_image_forbidden_location(app: FastAPI, compute_client: AsyncClient) -> None:
file_path = "%2e%2e/hello"
response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename=file_path), content=b"TEST")
response = await compute_client.post(app.url_path_for("compute:upload_dynamips_image", filename=file_path), content=b"TEST")
assert response.status_code == status.HTTP_403_FORBIDDEN
async def test_download_image(app: FastAPI, client: AsyncClient, images_dir: str) -> None:
async def test_download_image(app: FastAPI, compute_client: AsyncClient, images_dir: str) -> None:
response = await client.post(app.url_path_for("compute:upload_dynamips_image", filename="test3"), content=b"TEST")
response = await compute_client.post(app.url_path_for("compute:upload_dynamips_image", filename="test3"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
response = await client.get(app.url_path_for("compute:download_dynamips_image", filename="test3"))
response = await compute_client.get(app.url_path_for("compute:download_dynamips_image", filename="test3"))
assert response.status_code == status.HTTP_200_OK
async def test_download_image_forbidden(app: FastAPI, client: AsyncClient, tmpdir) -> None:
async def test_download_image_forbidden(app: FastAPI, compute_client: AsyncClient, tmpdir) -> None:
file_path = "foo/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/etc/passwd"
response = await client.get(app.url_path_for("compute:download_iou_image", filename=file_path))
response = await compute_client.get(app.url_path_for("compute:download_iou_image", filename=file_path))
assert response.status_code == status.HTTP_403_FORBIDDEN
async def test_iou_duplicate(app: FastAPI, client: AsyncClient, vm: dict, base_params: dict) -> None:
async def test_iou_duplicate(app: FastAPI, compute_client: AsyncClient, vm: dict, base_params: dict) -> None:
# create destination node first
response = await client.post(app.url_path_for("compute:create_iou_node", project_id=vm["project_id"]), json=base_params)
response = await compute_client.post(app.url_path_for("compute:create_iou_node", project_id=vm["project_id"]), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
params = {"destination_node_id": response.json()["node_id"]}
response = await client.post(app.url_path_for("compute:duplicate_iou_node",
response = await compute_client.post(app.url_path_for("compute:duplicate_iou_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_201_CREATED
tests/api/routes/compute/test_nat_nodes.py
+23 -23
View File
@@ -27,35 +27,35 @@ pytestmark = pytest.mark.asyncio
@pytest.fixture(scope="function")
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, ubridge_path: str, on_gns3vm) -> dict:
async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project, ubridge_path: str, on_gns3vm) -> dict:
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat._start_ubridge"):
response = await client.post(app.url_path_for("compute:create_nat_node", project_id=compute_project.id),
response = await compute_client.post(app.url_path_for("compute:create_nat_node", project_id=compute_project.id),
json={"name": "Nat 1"})
assert response.status_code == status.HTTP_201_CREATED
return response.json()
async def test_nat_create(app: FastAPI, client: AsyncClient, compute_project: Project, on_gns3vm) -> None:
async def test_nat_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project, on_gns3vm) -> None:
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat._start_ubridge"):
response = await client.post(app.url_path_for("compute:create_nat_node", project_id=compute_project.id),
response = await compute_client.post(app.url_path_for("compute:create_nat_node", project_id=compute_project.id),
json={"name": "Nat 1"})
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "Nat 1"
assert response.json()["project_id"] == compute_project.id
async def test_nat_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
async def test_nat_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await client.get(app.url_path_for("compute:get_nat_node", project_id=vm["project_id"], node_id=vm["node_id"]))
response = await compute_client.get(app.url_path_for("compute:get_nat_node", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "Nat 1"
assert response.json()["project_id"] == compute_project.id
assert response.json()["status"] == "started"
async def test_nat_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_nat_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -71,12 +71,12 @@ async def test_nat_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -
port_number="0")
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.add_nio"):
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
async def test_nat_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_nat_nio_update_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -91,7 +91,7 @@ async def test_nat_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -
adapter_number="0",
port_number="0")
await client.post(url, json=params)
await compute_client.post(url, json=params)
params["filters"] = {}
url = app.url_path_for("compute:update_nat_node_nio",
@@ -99,12 +99,12 @@ async def test_nat_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
response = await client.put(url, json=params)
response = await compute_client.put(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
async def test_nat_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_nat_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -119,7 +119,7 @@ async def test_nat_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> No
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.add_nio"):
await client.post(url, json=params)
await compute_client.post(url, json=params)
url = app.url_path_for("compute:delete_nat_node_nio",
project_id=vm["project_id"],
@@ -127,30 +127,30 @@ async def test_nat_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> No
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.remove_nio") as mock:
response = await client.delete(url)
response = await compute_client.delete(url)
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_nat_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_nat_delete(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
response = await client.delete(app.url_path_for("compute:delete_nat_node",
response = await compute_client.delete(app.url_path_for("compute:delete_nat_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_nat_update(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_nat_update(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
response = await client.put(app.url_path_for("compute:update_nat_node",
response = await compute_client.put(app.url_path_for("compute:update_nat_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json={"name": "test"})
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "test"
async def test_nat_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_nat_start_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"capture_file_name": "test.pcap",
@@ -163,13 +163,13 @@ async def test_nat_start_capture(app: FastAPI, client: AsyncClient, vm: dict) ->
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.start_capture") as mock:
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_200_OK
assert mock.called
assert "test.pcap" in response.json()["pcap_file_path"]
async def test_nat_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_nat_stop_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("compute:stop_nat_node_capture",
project_id=vm["project_id"],
@@ -178,7 +178,7 @@ async def test_nat_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) ->
port_number="0")
with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.stop_capture") as mock:
response = await client.post(url)
response = await compute_client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
@@ -188,5 +188,5 @@ async def test_nat_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) ->
#
# with asyncio_patch("gns3server.compute.builtin.nodes.nat.Nat.get_nio"):
# with asyncio_patch("gns3server.compute.builtin.Builtin.stream_pcap_file"):
# response = await client.get("/projects/{project_id}/nat/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True)
# response = await compute_client.get("/projects/{project_id}/nat/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True)
# assert response.status_code == status.HTTP_200_OK
tests/api/routes/compute/test_projects.py
+38 -38
View File
@@ -42,40 +42,40 @@ def base_params(tmpdir) -> dict:
return params
async def test_create_project_with_path(app: FastAPI, client: AsyncClient, base_params: dict) -> None:
async def test_create_project_with_path(app: FastAPI, compute_client: AsyncClient, base_params: dict) -> None:
with patch("gns3server.compute.project.Project.is_local", return_value=True):
response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
response = await compute_client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["project_id"] == base_params["project_id"]
async def test_create_project_with_path_and_empty_variables(app: FastAPI,
client: AsyncClient,
compute_client: AsyncClient,
base_params: dict) -> None:
base_params["variables"] = None
with patch("gns3server.compute.project.Project.is_local", return_value=True):
response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
response = await compute_client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["project_id"] == base_params["project_id"]
async def test_create_project_without_dir(app: FastAPI, client: AsyncClient, base_params: dict) -> None:
async def test_create_project_without_dir(app: FastAPI, compute_client: AsyncClient, base_params: dict) -> None:
del base_params["path"]
response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
response = await compute_client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["project_id"] == base_params["project_id"]
assert response.json()["name"] == base_params["name"]
async def test_show_project(app: FastAPI, client: AsyncClient, base_params: dict) -> None:
async def test_show_project(app: FastAPI, compute_client: AsyncClient, base_params: dict) -> None:
response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
response = await compute_client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
response = await client.get(app.url_path_for("compute:get_compute_project", project_id=base_params["project_id"]))
response = await compute_client.get(app.url_path_for("compute:get_compute_project", project_id=base_params["project_id"]))
#print(response.json().keys())
#assert len(response.json().keys()) == 3
@@ -84,60 +84,60 @@ async def test_show_project(app: FastAPI, client: AsyncClient, base_params: dict
assert response.json()["variables"] is None
async def test_show_project_invalid_uuid(app: FastAPI, client: AsyncClient) -> None:
async def test_show_project_invalid_uuid(app: FastAPI, compute_client: AsyncClient) -> None:
response = await client.get(app.url_path_for("compute:get_compute_project",
response = await compute_client.get(app.url_path_for("compute:get_compute_project",
project_id="50010203-0405-0607-0809-0a0b0c0d0e42"))
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_list_projects(app: FastAPI, client: AsyncClient) -> dict:
async def test_list_projects(app: FastAPI, compute_client: AsyncClient) -> dict:
ProjectManager.instance()._projects = {}
params = {"name": "test", "project_id": "51010203-0405-0607-0809-0a0b0c0d0e0f"}
response = await client.post(app.url_path_for("compute:create_compute_project"), json=params)
response = await compute_client.post(app.url_path_for("compute:create_compute_project"), json=params)
assert response.status_code == status.HTTP_201_CREATED
params = {"name": "test", "project_id": "52010203-0405-0607-0809-0a0b0c0d0e0b"}
response = await client.post(app.url_path_for("compute:create_compute_project"), json=params)
response = await compute_client.post(app.url_path_for("compute:create_compute_project"), json=params)
assert response.status_code == status.HTTP_201_CREATED
response = await client.get(app.url_path_for("compute:get_compute_projects"))
response = await compute_client.get(app.url_path_for("compute:get_compute_projects"))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 2
assert "51010203-0405-0607-0809-0a0b0c0d0e0f" in [p["project_id"] for p in response.json()]
async def test_delete_project(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
async def test_delete_project(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None:
with asyncio_patch("gns3server.compute.project.Project.delete", return_value=True) as mock:
response = await client.delete(app.url_path_for("compute:delete_compute_project", project_id=compute_project.id))
response = await compute_client.delete(app.url_path_for("compute:delete_compute_project", project_id=compute_project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
async def test_update_project(app: FastAPI, client: AsyncClient, base_params: dict) -> None:
async def test_update_project(app: FastAPI, compute_client: AsyncClient, base_params: dict) -> None:
response = await client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
response = await compute_client.post(app.url_path_for("compute:create_compute_project"), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
params = {"variables": [{"name": "TEST1", "value": "VAL1"}]}
response = await client.put(app.url_path_for("compute:update_compute_project", project_id=base_params["project_id"]),
response = await compute_client.put(app.url_path_for("compute:update_compute_project", project_id=base_params["project_id"]),
json=params)
assert response.status_code == status.HTTP_200_OK
assert response.json()["variables"] == [{"name": "TEST1", "value": "VAL1"}]
async def test_delete_project_invalid_uuid(app: FastAPI, client: AsyncClient) -> None:
async def test_delete_project_invalid_uuid(app: FastAPI, compute_client: AsyncClient) -> None:
response = await client.delete(app.url_path_for("compute:delete_compute_project", project_id=str(uuid.uuid4())))
response = await compute_client.delete(app.url_path_for("compute:delete_compute_project", project_id=str(uuid.uuid4())))
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_close_project(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
async def test_close_project(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None:
with asyncio_patch("gns3server.compute.project.Project.close", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:close_compute_project", project_id=compute_project.id))
response = await compute_client.post(app.url_path_for("compute:close_compute_project", project_id=compute_project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
@@ -147,18 +147,18 @@ async def test_close_project(app: FastAPI, client: AsyncClient, compute_project:
#
# ProjectHandler._notifications_listening = {compute_project.id: 2}
# with asyncio_patch("gns3server.compute.project.Project.close", return_value=True) as mock:
# response = await client.post("/projects/{project_id}/close".format(project_id=compute_project.id))
# response = await compute_client.post("/projects/{project_id}/close".format(project_id=compute_project.id))
# assert response.status_code == status.HTTP_204_NO_CONTENT
# assert not mock.called
async def test_close_project_invalid_uuid(app: FastAPI, client: AsyncClient) -> None:
async def test_close_project_invalid_uuid(app: FastAPI, compute_client: AsyncClient) -> None:
response = await client.post(app.url_path_for("compute:close_compute_project", project_id=str(uuid.uuid4())))
response = await compute_client.post(app.url_path_for("compute:close_compute_project", project_id=str(uuid.uuid4())))
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_get_file(app: FastAPI, client: AsyncClient, config, tmpdir) -> None:
async def test_get_file(app: FastAPI, compute_client: AsyncClient, config, tmpdir) -> None:
config.settings.Server.projects_path = str(tmpdir)
project = ProjectManager.instance().create_project(project_id="01010203-0405-0607-0809-0a0b0c0d0e0b")
@@ -166,25 +166,25 @@ async def test_get_file(app: FastAPI, client: AsyncClient, config, tmpdir) -> No
with open(os.path.join(project.path, "hello"), "w+") as f:
f.write("world")
response = await client.get(app.url_path_for("compute:get_compute_project_file", project_id=project.id, file_path="hello"))
response = await compute_client.get(app.url_path_for("compute:get_compute_project_file", project_id=project.id, file_path="hello"))
assert response.status_code == status.HTTP_200_OK
assert response.content == b"world"
response = await client.get(app.url_path_for("compute:get_compute_project_file", project_id=project.id, file_path="false"))
response = await compute_client.get(app.url_path_for("compute:get_compute_project_file", project_id=project.id, file_path="false"))
assert response.status_code == status.HTTP_404_NOT_FOUND
response = await client.get(app.url_path_for("compute:get_compute_project_file",
response = await compute_client.get(app.url_path_for("compute:get_compute_project_file",
project_id=project.id,
file_path="../hello"))
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_get_file_forbidden_location(app: FastAPI, client: AsyncClient, config, tmpdir) -> None:
async def test_get_file_forbidden_location(app: FastAPI, compute_client: AsyncClient, config, tmpdir) -> None:
config.settings.Server.projects_path = str(tmpdir)
project = ProjectManager.instance().create_project(project_id="01010203-0405-0607-0809-0a0b0c0d0e0b")
file_path = "foo/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/etc/passwd"
response = await client.get(
response = await compute_client.get(
app.url_path_for(
"compute:get_compute_project_file",
project_id=project.id,
@@ -194,12 +194,12 @@ async def test_get_file_forbidden_location(app: FastAPI, client: AsyncClient, co
assert response.status_code == status.HTTP_403_FORBIDDEN
async def test_write_file(app: FastAPI, client: AsyncClient, config, tmpdir) -> None:
async def test_write_file(app: FastAPI, compute_client: AsyncClient, config, tmpdir) -> None:
config.settings.Server.projects_path = str(tmpdir)
project = ProjectManager.instance().create_project(project_id="01010203-0405-0607-0809-0a0b0c0d0e0b")
response = await client.post(app.url_path_for("compute:write_compute_project_file",
response = await compute_client.post(app.url_path_for("compute:write_compute_project_file",
project_id=project.id,
file_path="hello"), content=b"world")
assert response.status_code == status.HTTP_204_NO_CONTENT
@@ -207,19 +207,19 @@ async def test_write_file(app: FastAPI, client: AsyncClient, config, tmpdir) ->
with open(os.path.join(project.path, "hello")) as f:
assert f.read() == "world"
response = await client.post(app.url_path_for("compute:write_compute_project_file",
response = await compute_client.post(app.url_path_for("compute:write_compute_project_file",
project_id=project.id,
file_path="../hello"))
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_write_file_forbidden_location(app: FastAPI, client: AsyncClient, config, tmpdir) -> None:
async def test_write_file_forbidden_location(app: FastAPI, compute_client: AsyncClient, config, tmpdir) -> None:
config.settings.Server.projects_path = str(tmpdir)
project = ProjectManager.instance().create_project(project_id="01010203-0405-0607-0809-0a0b0c0d0e0b")
file_path = "%2e%2e/hello"
response = await client.post(app.url_path_for("compute:write_compute_project_file",
response = await compute_client.post(app.url_path_for("compute:write_compute_project_file",
project_id=project.id,
file_path=file_path), content=b"world")
assert response.status_code == status.HTTP_403_FORBIDDEN
tests/api/routes/compute/test_qemu_nodes.py
+74 -74
View File
@@ -64,20 +64,20 @@ def base_params(tmpdir, fake_qemu_bin) -> dict:
@pytest.fixture
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, base_params: dict) -> None:
async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project, base_params: dict) -> None:
response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params)
response = await compute_client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
return response.json()
async def test_qemu_create(app: FastAPI,
client: AsyncClient,
compute_client: AsyncClient,
compute_project: Project,
base_params: dict,
fake_qemu_bin: str) -> None:
response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params)
response = await compute_client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@@ -86,14 +86,14 @@ async def test_qemu_create(app: FastAPI,
async def test_qemu_create_platform(app: FastAPI,
client: AsyncClient,
compute_client: AsyncClient,
compute_project: Project,
base_params: dict,
fake_qemu_bin: str):
base_params["qemu_path"] = None
base_params["platform"] = "x86_64"
response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params)
response = await compute_client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@@ -103,7 +103,7 @@ async def test_qemu_create_platform(app: FastAPI,
@pytest.mark.asyncio
async def test_qemu_create_with_params(app: FastAPI,
client: AsyncClient,
compute_client: AsyncClient,
compute_project: Project,
base_params: dict,
fake_qemu_vm: str):
@@ -111,7 +111,7 @@ async def test_qemu_create_with_params(app: FastAPI,
params = base_params
params["ram"] = 1024
params["hda_disk_image"] = "linux载.img"
response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params)
response = await compute_client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@@ -122,26 +122,26 @@ async def test_qemu_create_with_params(app: FastAPI,
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
async def test_qemu_create_with_project_file(app: FastAPI,
client: AsyncClient,
compute_client: AsyncClient,
compute_project: Project,
base_params: dict,
fake_qemu_vm: str) -> None:
response = await client.post(app.url_path_for("compute:write_compute_project_file",
response = await compute_client.post(app.url_path_for("compute:write_compute_project_file",
project_id=compute_project.id,
file_path="hello.img"), content=b"world")
assert response.status_code == status.HTTP_204_NO_CONTENT
params = base_params
params["hda_disk_image"] = "hello.img"
response = await client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params)
response = await compute_client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["hda_disk_image"] == "hello.img"
assert response.json()["hda_disk_image_md5sum"] == "7d793037a0760186574b0282f2f435e7"
async def test_qemu_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict):
async def test_qemu_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict):
response = await client.get(app.url_path_for("compute:get_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"]))
response = await compute_client.get(app.url_path_for("compute:get_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
@@ -151,60 +151,60 @@ async def test_qemu_get(app: FastAPI, client: AsyncClient, compute_project: Proj
vm["node_id"])
async def test_qemu_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_qemu_start(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.start", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:start_qemu_node",
response = await compute_client.post(app.url_path_for("compute:start_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_qemu_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_qemu_stop(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.stop", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:stop_qemu_node",
response = await compute_client.post(app.url_path_for("compute:stop_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_qemu_reload(app: FastAPI, client: AsyncClient, vm) -> None:
async def test_qemu_reload(app: FastAPI, compute_client: AsyncClient, vm) -> None:
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.reload", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:reload_qemu_node",
response = await compute_client.post(app.url_path_for("compute:reload_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_qemu_suspend(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_qemu_suspend(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.suspend", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:suspend_qemu_node",
response = await compute_client.post(app.url_path_for("compute:suspend_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_qemu_resume(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_qemu_resume(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.resume", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:resume_qemu_node",
response = await compute_client.post(app.url_path_for("compute:resume_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_qemu_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_qemu_delete(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.qemu.Qemu.delete_node", return_value=True) as mock:
response = await client.delete(app.url_path_for("compute:delete_qemu_node",
response = await compute_client.delete(app.url_path_for("compute:delete_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
@@ -212,7 +212,7 @@ async def test_qemu_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_qemu_update(app: FastAPI,
client: AsyncClient,
compute_client: AsyncClient,
vm: dict,
free_console_port: int,
fake_qemu_vm: str) -> None:
@@ -224,7 +224,7 @@ async def test_qemu_update(app: FastAPI,
"hdb_disk_image": "linux载.img"
}
response = await client.put(app.url_path_for("compute:update_qemu_node",
response = await compute_client.put(app.url_path_for("compute:update_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
@@ -234,7 +234,7 @@ async def test_qemu_update(app: FastAPI,
assert response.json()["ram"] == 1024
async def test_qemu_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_qemu_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -244,7 +244,7 @@ async def test_qemu_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict)
}
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.add_ubridge_udp_connection"):
await client.put(app.url_path_for("compute:update_qemu_node",
await compute_client.put(app.url_path_for("compute:update_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json={"adapters": 2})
@@ -253,12 +253,12 @@ async def test_qemu_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict)
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
async def test_qemu_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_qemu_nio_update_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -267,7 +267,7 @@ async def test_qemu_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict)
"rhost": "127.0.0.1"
}
await client.put(app.url_path_for("compute:update_qemu_node",
await compute_client.put(app.url_path_for("compute:update_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json={"adapters": 2})
@@ -277,7 +277,7 @@ async def test_qemu_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict)
adapter_number="1",
port_number="0")
await client.post(url, json=params)
await compute_client.post(url, json=params)
params["filters"] = {}
@@ -286,12 +286,12 @@ async def test_qemu_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict)
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
response = await client.put(url, json=params)
response = await compute_client.put(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
async def test_qemu_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_qemu_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -301,7 +301,7 @@ async def test_qemu_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> N
}
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM._ubridge_send"):
await client.put(app.url_path_for("compute:update_qemu_node",
await compute_client.put(app.url_path_for("compute:update_qemu_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json={"adapters": 2})
@@ -310,30 +310,30 @@ async def test_qemu_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> N
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
await client.post(url, json=params)
await compute_client.post(url, json=params)
url = app.url_path_for("compute:delete_qemu_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="1",
port_number="0")
response = await client.delete(url)
response = await compute_client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_qemu_list_binaries(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_qemu_list_binaries(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
ret = [{"path": "/tmp/1", "version": "2.2.0"},
{"path": "/tmp/2", "version": "2.1.0"}]
with asyncio_patch("gns3server.compute.qemu.Qemu.binary_list", return_value=ret) as mock:
response = await client.get(app.url_path_for("compute:get_qemu_binaries"))
response = await compute_client.get(app.url_path_for("compute:get_qemu_binaries"))
assert mock.called_with(None)
assert response.status_code == status.HTTP_200_OK
assert response.json() == ret
# async def test_qemu_list_binaries_filter(app: FastAPI, client: AsyncClient, vm: dict) -> None:
# async def test_qemu_list_binaries_filter(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
#
# ret = [
# {"path": "/tmp/x86_64", "version": "2.2.0"},
@@ -342,25 +342,25 @@ async def test_qemu_list_binaries(app: FastAPI, client: AsyncClient, vm: dict) -
# ]
#
# with asyncio_patch("gns3server.compute.qemu.Qemu.binary_list", return_value=ret) as mock:
# response = await client.get(app.url_path_for("compute:get_qemu_binaries"),
# response = await compute_client.get(app.url_path_for("compute:get_qemu_binaries"),
# json={"archs": ["i386"]})
# assert response.status_code == status.HTTP_200_OK
# assert mock.called_with(["i386"])
# assert response.json() == ret
async def test_images(app: FastAPI, client: AsyncClient, fake_qemu_vm) -> None:
async def test_images(app: FastAPI, compute_client: AsyncClient, fake_qemu_vm) -> None:
response = await client.get(app.url_path_for("compute:get_qemu_images"))
response = await compute_client.get(app.url_path_for("compute:get_qemu_images"))
assert response.status_code == status.HTTP_200_OK
assert {"filename": "linux载.img", "path": "linux载.img", "md5sum": "c4ca4238a0b923820dcc509a6f75849b", "filesize": 1} in response.json()
async def test_upload_image(app: FastAPI, client: AsyncClient, tmpdir: str) -> None:
async def test_upload_image(app: FastAPI, compute_client: AsyncClient, tmpdir: str) -> None:
with patch("gns3server.compute.Qemu.get_images_directory", return_value=str(tmpdir)):
response = await client.post(app.url_path_for("compute:upload_qemu_image",
response = await compute_client.post(app.url_path_for("compute:upload_qemu_image",
filename="test2使"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
@@ -372,11 +372,11 @@ async def test_upload_image(app: FastAPI, client: AsyncClient, tmpdir: str) -> N
assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf"
async def test_upload_image_ova(app: FastAPI, client: AsyncClient, tmpdir:str) -> None:
async def test_upload_image_ova(app: FastAPI, compute_client: AsyncClient, tmpdir:str) -> None:
with patch("gns3server.compute.Qemu.get_images_directory", return_value=str(tmpdir)):
response = await client.post(app.url_path_for("compute:upload_qemu_image",
response = await compute_client.post(app.url_path_for("compute:upload_qemu_image",
filename="test2.ova/test2.vmdk"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
@@ -388,42 +388,42 @@ async def test_upload_image_ova(app: FastAPI, client: AsyncClient, tmpdir:str) -
assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf"
async def test_upload_image_forbidden_location(app: FastAPI, client: AsyncClient, tmpdir: str) -> None:
async def test_upload_image_forbidden_location(app: FastAPI, compute_client: AsyncClient, tmpdir: str) -> None:
response = await client.post(app.url_path_for("compute:upload_qemu_image",
response = await compute_client.post(app.url_path_for("compute:upload_qemu_image",
filename="/qemu/images/../../test2"), content=b"TEST")
assert response.status_code == status.HTTP_403_FORBIDDEN
async def test_download_image(app: FastAPI, client: AsyncClient, images_dir: str) -> None:
async def test_download_image(app: FastAPI, compute_client: AsyncClient, images_dir: str) -> None:
response = await client.post(app.url_path_for("compute:upload_qemu_image", filename="test3"), content=b"TEST")
response = await compute_client.post(app.url_path_for("compute:upload_qemu_image", filename="test3"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
response = await client.get(app.url_path_for("compute:download_qemu_image", filename="test3"))
response = await compute_client.get(app.url_path_for("compute:download_qemu_image", filename="test3"))
assert response.status_code == status.HTTP_200_OK
async def test_download_image_forbidden_location(app: FastAPI, client: AsyncClient, tmpdir) -> None:
async def test_download_image_forbidden_location(app: FastAPI, compute_client: AsyncClient, tmpdir) -> None:
file_path = "foo/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/%2e%2e/etc/passwd"
response = await client.get(app.url_path_for("compute:download_qemu_image", filename=file_path))
response = await compute_client.get(app.url_path_for("compute:download_qemu_image", filename=file_path))
assert response.status_code == status.HTTP_403_FORBIDDEN
@pytest.mark.skipif(not sys.platform.startswith("win") and os.getuid() == 0, reason="Root can delete any image")
async def test_upload_image_permission_denied(app: FastAPI, client: AsyncClient, images_dir: str) -> None:
async def test_upload_image_permission_denied(app: FastAPI, compute_client: AsyncClient, images_dir: str) -> None:
with open(os.path.join(images_dir, "QEMU", "test2.tmp"), "w+") as f:
f.write("")
os.chmod(os.path.join(images_dir, "QEMU", "test2.tmp"), 0)
response = await client.post(app.url_path_for("compute:upload_qemu_image", filename="test2"), content=b"TEST")
response = await compute_client.post(app.url_path_for("compute:upload_qemu_image", filename="test2"), content=b"TEST")
assert response.status_code == status.HTTP_409_CONFLICT
@pytest.mark.asyncio
async def test_create_img_relative(app: FastAPI, client: AsyncClient):
async def test_create_img_relative(app: FastAPI, compute_client: AsyncClient):
params = {
"qemu_img": "/tmp/qemu-img",
@@ -436,11 +436,11 @@ async def test_create_img_relative(app: FastAPI, client: AsyncClient):
"size": 100
}
with asyncio_patch("gns3server.compute.Qemu.create_disk"):
response = await client.post(app.url_path_for("compute:create_qemu_image"), json=params)
response = await compute_client.post(app.url_path_for("compute:create_qemu_image"), json=params)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_create_img_absolute_non_local(app: FastAPI, client: AsyncClient, config) -> None:
async def test_create_img_absolute_non_local(app: FastAPI, compute_client: AsyncClient, config) -> None:
config.settings.Server.local = False
params = {
@@ -454,11 +454,11 @@ async def test_create_img_absolute_non_local(app: FastAPI, client: AsyncClient,
"size": 100
}
with asyncio_patch("gns3server.compute.Qemu.create_disk"):
response = await client.post(app.url_path_for("compute:create_qemu_image"), json=params)
response = await compute_client.post(app.url_path_for("compute:create_qemu_image"), json=params)
assert response.status_code == 403
async def test_create_img_absolute_local(app: FastAPI, client: AsyncClient, config) -> None:
async def test_create_img_absolute_local(app: FastAPI, compute_client: AsyncClient, config) -> None:
params = {
"qemu_img": "/tmp/qemu-img",
@@ -471,36 +471,36 @@ async def test_create_img_absolute_local(app: FastAPI, client: AsyncClient, conf
"size": 100
}
with asyncio_patch("gns3server.compute.Qemu.create_disk"):
response = await client.post(app.url_path_for("compute:create_qemu_image"), json=params)
response = await compute_client.post(app.url_path_for("compute:create_qemu_image"), json=params)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_capabilities(app: FastAPI, client: AsyncClient) -> None:
async def test_capabilities(app: FastAPI, compute_client: AsyncClient) -> None:
with asyncio_patch("gns3server.compute.Qemu.get_kvm_archs", return_value=["x86_64"]):
response = await client.get(app.url_path_for("compute:get_qemu_capabilities"))
response = await compute_client.get(app.url_path_for("compute:get_qemu_capabilities"))
assert response.json()["kvm"] == ["x86_64"]
async def test_qemu_duplicate(app: FastAPI,
client: AsyncClient,
compute_client: AsyncClient,
compute_project: Project,
vm: dict,
base_params: dict) -> None:
# create destination node first
response = await client.post(app.url_path_for("compute:create_qemu_node",
response = await compute_client.post(app.url_path_for("compute:create_qemu_node",
project_id=vm["project_id"]), json=base_params)
assert response.status_code == status.HTTP_201_CREATED
params = {"destination_node_id": response.json()["node_id"]}
response = await client.post(app.url_path_for("compute:duplicate_qemu_node",
response = await compute_client.post(app.url_path_for("compute:duplicate_qemu_node",
project_id=vm["project_id"], node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_201_CREATED
@pytest.mark.asyncio
async def test_qemu_start_capture(app: FastAPI, client: AsyncClient, vm):
async def test_qemu_start_capture(app: FastAPI, compute_client: AsyncClient, vm):
params = {
"capture_file_name": "test.pcap",
@@ -515,14 +515,14 @@ async def test_qemu_start_capture(app: FastAPI, client: AsyncClient, vm):
with patch("gns3server.compute.qemu.qemu_vm.QemuVM.is_running", return_value=True):
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.start_capture") as mock:
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_200_OK
assert mock.called
assert "test.pcap" in response.json()["pcap_file_path"]
@pytest.mark.asyncio
async def test_qemu_stop_capture(app: FastAPI, client: AsyncClient, vm):
async def test_qemu_stop_capture(app: FastAPI, compute_client: AsyncClient, vm):
url = app.url_path_for("compute:stop_qemu_node_capture",
project_id=vm["project_id"],
@@ -532,15 +532,15 @@ async def test_qemu_stop_capture(app: FastAPI, client: AsyncClient, vm):
with patch("gns3server.compute.qemu.qemu_vm.QemuVM.is_running", return_value=True):
with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.stop_capture") as mock:
response = await client.post(url)
response = await compute_client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
# @pytest.mark.asyncio
# async def test_qemu_pcap(app: FastAPI, client: AsyncClient, vm, compute_project):
# async def test_qemu_pcap(app: FastAPI, compute_client: AsyncClient, vm, compute_project):
#
# with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.get_nio"):
# with asyncio_patch("gns3server.compute.qemu.Qemu.stream_pcap_file"):
# response = await client.get("/projects/{project_id}/qemu/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True)
# response = await compute_client.get("/projects/{project_id}/qemu/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True)
# assert response.status_code == status.HTTP_200_OK
tests/api/routes/compute/test_virtualbox_nodes.py
+30 -30
View File
@@ -28,7 +28,7 @@ pytestmark = pytest.mark.asyncio
@pytest.fixture(scope="function")
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None:
vboxmanage_path = "/fake/VboxManage"
params = {
@@ -38,7 +38,7 @@ async def vm(app: FastAPI, client: AsyncClient, compute_project: Project) -> Non
}
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.create", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:create_virtualbox_node", project_id=compute_project.id),
response = await compute_client.post(app.url_path_for("compute:create_virtualbox_node", project_id=compute_project.id),
json=params)
assert mock.called
assert response.status_code == status.HTTP_201_CREATED
@@ -47,7 +47,7 @@ async def vm(app: FastAPI, client: AsyncClient, compute_project: Project) -> Non
return response.json()
async def test_vbox_create(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
async def test_vbox_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None:
params = {
"name": "VM1",
@@ -56,16 +56,16 @@ async def test_vbox_create(app: FastAPI, client: AsyncClient, compute_project: P
}
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.create", return_value=True):
response = await client.post(app.url_path_for("compute:create_virtualbox_node", project_id=compute_project.id),
response = await compute_client.post(app.url_path_for("compute:create_virtualbox_node", project_id=compute_project.id),
json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "VM1"
assert response.json()["project_id"] == compute_project.id
async def test_vbox_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
async def test_vbox_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await client.get(app.url_path_for("compute:get_virtualbox_node",
response = await compute_client.get(app.url_path_for("compute:get_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert response.status_code == status.HTTP_200_OK
@@ -73,58 +73,58 @@ async def test_vbox_get(app: FastAPI, client: AsyncClient, compute_project: Proj
assert response.json()["project_id"] == compute_project.id
async def test_vbox_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vbox_start(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.start", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:start_virtualbox_node",
response = await compute_client.post(app.url_path_for("compute:start_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vbox_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vbox_stop(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.stop", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:stop_virtualbox_node",
response = await compute_client.post(app.url_path_for("compute:stop_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vbox_suspend(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vbox_suspend(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.suspend", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:suspend_virtualbox_node",
response = await compute_client.post(app.url_path_for("compute:suspend_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vbox_resume(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vbox_resume(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.resume", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:resume_virtualbox_node",
response = await compute_client.post(app.url_path_for("compute:resume_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vbox_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vbox_reload(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.reload", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:reload_virtualbox_node",
response = await compute_client.post(app.url_path_for("compute:reload_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vbox_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vbox_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -140,7 +140,7 @@ async def test_vbox_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict)
port_number="0")
with asyncio_patch('gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.adapter_add_nio_binding') as mock:
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert mock.called
args, kwgars = mock.call_args
assert args[0] == 0
@@ -150,7 +150,7 @@ async def test_vbox_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict)
# @pytest.mark.asyncio
# async def test_vbox_nio_update_udp(app: FastAPI, client: AsyncClient, vm):
# async def test_vbox_nio_update_udp(app: FastAPI, compute_client: AsyncClient, vm):
#
# params = {
# "type": "nio_udp",
@@ -162,13 +162,13 @@ async def test_vbox_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict)
#
# with asyncio_patch('gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.ethernet_adapters'):
# with asyncio_patch('gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.adapter_remove_nio_binding'):
# response = await client.put("/projects/{project_id}/virtualbox/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
# response = await compute_client.put("/projects/{project_id}/virtualbox/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
#
# assert response.status_code == status.HTTP_201_CREATED
# assert response.json()["type"] == "nio_udp"
async def test_vbox_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vbox_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("compute:delete_virtualbox_node_nio",
project_id=vm["project_id"],
@@ -177,7 +177,7 @@ async def test_vbox_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> N
port_number="0")
with asyncio_patch('gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.adapter_remove_nio_binding') as mock:
response = await client.delete(url)
response = await compute_client.delete(url)
assert mock.called
args, kwgars = mock.call_args
assert args[0] == 0
@@ -185,14 +185,14 @@ async def test_vbox_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> N
@pytest.mark.asyncio
async def test_vbox_update(app: FastAPI, client: AsyncClient, vm, free_console_port):
async def test_vbox_update(app: FastAPI, compute_client: AsyncClient, vm, free_console_port):
params = {
"name": "test",
"console": free_console_port
}
response = await client.put(app.url_path_for("compute:update_virtualbox_node",
response = await compute_client.put(app.url_path_for("compute:update_virtualbox_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
@@ -201,7 +201,7 @@ async def test_vbox_update(app: FastAPI, client: AsyncClient, vm, free_console_p
@pytest.mark.asyncio
async def test_virtualbox_start_capture(app: FastAPI, client: AsyncClient, vm):
async def test_virtualbox_start_capture(app: FastAPI, compute_client: AsyncClient, vm):
params = {
"capture_file_name": "test.pcap",
@@ -216,14 +216,14 @@ async def test_virtualbox_start_capture(app: FastAPI, client: AsyncClient, vm):
with patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.is_running", return_value=True):
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.start_capture") as mock:
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_200_OK
assert mock.called
assert "test.pcap" in response.json()["pcap_file_path"]
@pytest.mark.asyncio
async def test_virtualbox_stop_capture(app: FastAPI, client: AsyncClient, vm):
async def test_virtualbox_stop_capture(app: FastAPI, compute_client: AsyncClient, vm):
url = app.url_path_for("compute:stop_virtualbox_node_capture",
project_id=vm["project_id"],
@@ -233,15 +233,15 @@ async def test_virtualbox_stop_capture(app: FastAPI, client: AsyncClient, vm):
with patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.is_running", return_value=True):
with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.stop_capture") as mock:
response = await client.post(url)
response = await compute_client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
# @pytest.mark.asyncio
# async def test_virtualbox_pcap(app: FastAPI, client: AsyncClient, vm, compute_project):
# async def test_virtualbox_pcap(app: FastAPI, compute_client: AsyncClient, vm, compute_project):
#
# with asyncio_patch("gns3server.compute.virtualbox.virtualbox_vm.VirtualBoxVM.get_nio"):
# with asyncio_patch("gns3server.compute.virtualbox.VirtualBox.stream_pcap_file"):
# response = await client.get("/projects/{project_id}/virtualbox/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True)
# response = await compute_client.get("/projects/{project_id}/virtualbox/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True)
# assert response.status_code == status.HTTP_200_OK
tests/api/routes/compute/test_vmware_nodes.py
+30 -30
View File
@@ -28,7 +28,7 @@ pytestmark = pytest.mark.asyncio
@pytest.fixture(scope="function")
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, vmx_path: str) -> dict:
async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vmx_path: str) -> dict:
params = {
"name": "VMTEST",
@@ -37,7 +37,7 @@ async def vm(app: FastAPI, client: AsyncClient, compute_project: Project, vmx_pa
}
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.create", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:create_vmware_node", project_id=compute_project.id),
response = await compute_client.post(app.url_path_for("compute:create_vmware_node", project_id=compute_project.id),
json=params)
assert mock.called
assert response.status_code == status.HTTP_201_CREATED
@@ -56,7 +56,7 @@ def vmx_path(tmpdir: str) -> str:
return path
async def test_vmware_create(app: FastAPI, client: AsyncClient, compute_project: Project, vmx_path: str) -> None:
async def test_vmware_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vmx_path: str) -> None:
params = {
"name": "VM1",
@@ -65,72 +65,72 @@ async def test_vmware_create(app: FastAPI, client: AsyncClient, compute_project:
}
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.create", return_value=True):
response = await client.post(app.url_path_for("compute:create_vmware_node", project_id=compute_project.id),
response = await compute_client.post(app.url_path_for("compute:create_vmware_node", project_id=compute_project.id),
json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "VM1"
assert response.json()["project_id"] == compute_project.id
async def test_vmware_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
async def test_vmware_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await client.get(app.url_path_for("compute:get_vmware_node", project_id=vm["project_id"], node_id=vm["node_id"]))
response = await compute_client.get(app.url_path_for("compute:get_vmware_node", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "VMTEST"
assert response.json()["project_id"] == compute_project.id
async def test_vmware_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vmware_start(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.start", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:start_vmware_node",
response = await compute_client.post(app.url_path_for("compute:start_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vmware_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vmware_stop(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.stop", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:stop_vmware_node",
response = await compute_client.post(app.url_path_for("compute:stop_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vmware_suspend(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vmware_suspend(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.suspend", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:suspend_vmware_node",
response = await compute_client.post(app.url_path_for("compute:suspend_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vmware_resume(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vmware_resume(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.resume", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:resume_vmware_node",
response = await compute_client.post(app.url_path_for("compute:resume_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vmware_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vmware_reload(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.reload", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:reload_vmware_node",
response = await compute_client.post(app.url_path_for("compute:reload_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vmware_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vmware_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -146,7 +146,7 @@ async def test_vmware_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict
port_number="0")
with asyncio_patch('gns3server.compute.vmware.vmware_vm.VMwareVM.adapter_add_nio_binding') as mock:
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert mock.called
args, kwgars = mock.call_args
assert args[0] == 0
@@ -156,7 +156,7 @@ async def test_vmware_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict
# @pytest.mark.asyncio
# async def test_vmware_nio_update_udp(app: FastAPI, client: AsyncClient, vm):
# async def test_vmware_nio_update_udp(app: FastAPI, compute_client: AsyncClient, vm):
#
# params = {
# "type": "nio_udp",
@@ -169,12 +169,12 @@ async def test_vmware_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict
# with asyncio_patch('gns3server.compute.vmware.vmware_vm.VMwareVM._ubridge_send'):
# with asyncio_patch('gns3server.compute.vmware.vmware_vm.VMwareVM.ethernet_adapters'):
# with patch('gns3server.compute.vmware.vmware_vm.VMwareVM._get_vnet') as mock:
# response = await client.put("/projects/{project_id}/vmware/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
# response = await compute_client.put("/projects/{project_id}/vmware/nodes/{node_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], node_id=vm["node_id"]), params)
# assert response.status_code == status.HTTP_201_CREATED
# assert response.json()["type"] == "nio_udp"
async def test_vmware_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vmware_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("compute:delete_vmware_node_nio",
project_id=vm["project_id"],
@@ -183,7 +183,7 @@ async def test_vmware_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) ->
port_number="0")
with asyncio_patch('gns3server.compute.vmware.vmware_vm.VMwareVM.adapter_remove_nio_binding') as mock:
response = await client.delete(url)
response = await compute_client.delete(url)
assert mock.called
args, kwgars = mock.call_args
assert args[0] == 0
@@ -191,14 +191,14 @@ async def test_vmware_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) ->
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vmware_update(app: FastAPI, client: AsyncClient, vm: dict, free_console_port: int) -> None:
async def test_vmware_update(app: FastAPI, compute_client: AsyncClient, vm: dict, free_console_port: int) -> None:
params = {
"name": "test",
"console": free_console_port
}
response = await client.put(app.url_path_for("compute:update_vmware_node",
response = await compute_client.put(app.url_path_for("compute:update_vmware_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
@@ -206,7 +206,7 @@ async def test_vmware_update(app: FastAPI, client: AsyncClient, vm: dict, free_c
assert response.json()["console"] == free_console_port
async def test_vmware_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vmware_start_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"capture_file_name": "test.pcap",
@@ -222,13 +222,13 @@ async def test_vmware_start_capture(app: FastAPI, client: AsyncClient, vm: dict)
with patch("gns3server.compute.vmware.vmware_vm.VMwareVM.is_running", return_value=True):
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.start_capture") as mock:
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_200_OK
assert mock.called
assert "test.pcap" in response.json()["pcap_file_path"]
async def test_vmware_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vmware_stop_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("compute:stop_vmware_node_capture",
project_id=vm["project_id"],
@@ -238,15 +238,15 @@ async def test_vmware_stop_capture(app: FastAPI, client: AsyncClient, vm: dict)
with patch("gns3server.compute.vmware.vmware_vm.VMwareVM.is_running", return_value=True):
with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.stop_capture") as mock:
response = await client.post(url)
response = await compute_client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
# @pytest.mark.asyncio
# async def test_vmware_pcap(app: FastAPI, client: AsyncClient, vm, compute_project):
# async def test_vmware_pcap(app: FastAPI, compute_client: AsyncClient, vm, compute_project):
#
# with asyncio_patch("gns3server.compute.vmware.vmware_vm.VMwareVM.get_nio"):
# with asyncio_patch("gns3server.compute.vmware.VMware.stream_pcap_file"):
# response = await client.get("/projects/{project_id}/vmware/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True)
# response = await compute_client.get("/projects/{project_id}/vmware/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True)
# assert response.status_code == status.HTTP_200_OK
tests/api/routes/compute/test_vpcs_nodes.py
+37 -37
View File
@@ -28,47 +28,47 @@ pytestmark = pytest.mark.asyncio
@pytest.fixture
async def vm(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None:
params = {"name": "PC TEST 1"}
response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
response = await compute_client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
return response.json()
async def test_vpcs_create(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
async def test_vpcs_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None:
params = {"name": "PC TEST 1"}
response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
response = await compute_client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
async def test_vpcs_get(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
async def test_vpcs_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None:
response = await client.get(app.url_path_for("compute:get_vpcs_node", project_id=vm["project_id"], node_id=vm["node_id"]))
response = await compute_client.get(app.url_path_for("compute:get_vpcs_node", project_id=vm["project_id"], node_id=vm["node_id"]))
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
assert response.json()["status"] == "stopped"
async def test_vpcs_create_startup_script(app: FastAPI, client: AsyncClient, compute_project: Project) -> None:
async def test_vpcs_create_startup_script(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None:
params = {
"name": "PC TEST 1",
"startup_script": "ip 192.168.1.2\necho TEST"
}
response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
response = await compute_client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
async def test_vpcs_create_port(app: FastAPI,
client: AsyncClient,
compute_client: AsyncClient,
compute_project: Project,
free_console_port: int) -> None:
@@ -77,14 +77,14 @@ async def test_vpcs_create_port(app: FastAPI,
"console": free_console_port
}
response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
response = await compute_client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "PC TEST 1"
assert response.json()["project_id"] == compute_project.id
assert response.json()["console"] == free_console_port
async def test_vpcs_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vpcs_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -100,12 +100,12 @@ async def test_vpcs_nio_create_udp(app: FastAPI, client: AsyncClient, vm: dict)
port_number="0")
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.add_ubridge_udp_connection"):
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
async def test_vpcs_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vpcs_nio_update_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -121,7 +121,7 @@ async def test_vpcs_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict)
port_number="0")
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.add_ubridge_udp_connection"):
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
params["filters"] = {}
@@ -130,12 +130,12 @@ async def test_vpcs_nio_update_udp(app: FastAPI, client: AsyncClient, vm: dict)
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
response = await client.put(url, json=params)
response = await compute_client.put(url, json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
async def test_vpcs_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vpcs_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"type": "nio_udp",
@@ -150,73 +150,73 @@ async def test_vpcs_delete_nio(app: FastAPI, client: AsyncClient, vm: dict) -> N
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
await client.post(url, json=params)
await compute_client.post(url, json=params)
url = app.url_path_for("compute:delete_vpcs_node_nio",
project_id=vm["project_id"],
node_id=vm["node_id"],
adapter_number="0",
port_number="0")
response = await client.delete(url)
response = await compute_client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vpcs_start(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vpcs_start(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.start", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:start_vpcs_node",
response = await compute_client.post(app.url_path_for("compute:start_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vpcs_stop(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vpcs_stop(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.stop", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:stop_vpcs_node",
response = await compute_client.post(app.url_path_for("compute:stop_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vpcs_reload(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vpcs_reload(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.reload", return_value=True) as mock:
response = await client.post(app.url_path_for("compute:reload_vpcs_node",
response = await compute_client.post(app.url_path_for("compute:reload_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vpcs_delete(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vpcs_delete(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
with asyncio_patch("gns3server.compute.vpcs.VPCS.delete_node", return_value=True) as mock:
response = await client.delete(app.url_path_for("compute:delete_vpcs_node",
response = await compute_client.delete(app.url_path_for("compute:delete_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_vpcs_duplicate(app: FastAPI, client: AsyncClient, compute_project: Project, vm: dict) -> None:
async def test_vpcs_duplicate(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict) -> None:
# create destination node first
params = {"name": "PC TEST 1"}
response = await client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
response = await compute_client.post(app.url_path_for("compute:create_vpcs_node", project_id=compute_project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
params = {"destination_node_id": response.json()["node_id"]}
response = await client.post(app.url_path_for("compute:duplicate_vpcs_node",
response = await compute_client.post(app.url_path_for("compute:duplicate_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_201_CREATED
async def test_vpcs_update(app: FastAPI, client: AsyncClient, vm: dict, free_console_port: int) -> None:
async def test_vpcs_update(app: FastAPI, compute_client: AsyncClient, vm: dict, free_console_port: int) -> None:
console_port = free_console_port
params = {
@@ -224,7 +224,7 @@ async def test_vpcs_update(app: FastAPI, client: AsyncClient, vm: dict, free_con
"console": console_port
}
response = await client.put(app.url_path_for("compute:update_vpcs_node",
response = await compute_client.put(app.url_path_for("compute:update_vpcs_node",
project_id=vm["project_id"],
node_id=vm["node_id"]), json=params)
assert response.status_code == status.HTTP_200_OK
@@ -232,7 +232,7 @@ async def test_vpcs_update(app: FastAPI, client: AsyncClient, vm: dict, free_con
assert response.json()["console"] == console_port
async def test_vpcs_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vpcs_start_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
params = {
"capture_file_name": "test.pcap",
@@ -247,13 +247,13 @@ async def test_vpcs_start_capture(app: FastAPI, client: AsyncClient, vm: dict) -
with patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.is_running", return_value=True):
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.start_capture") as mock:
response = await client.post(url, json=params)
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_200_OK
assert mock.called
assert "test.pcap" in response.json()["pcap_file_path"]
async def test_vpcs_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) -> None:
async def test_vpcs_stop_capture(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
url = app.url_path_for("compute:stop_vpcs_node_capture",
project_id=vm["project_id"],
@@ -263,15 +263,15 @@ async def test_vpcs_stop_capture(app: FastAPI, client: AsyncClient, vm: dict) ->
with patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.is_running", return_value=True):
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.stop_capture") as mock:
response = await client.post(url)
response = await compute_client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
# @pytest.mark.asyncio
# async def test_vpcs_pcap(app: FastAPI, client: AsyncClient, vm, compute_project: Project):
# async def test_vpcs_pcap(app: FastAPI, compute_client: AsyncClient, vm, compute_project: Project):
#
# with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.get_nio"):
# with asyncio_patch("gns3server.compute.vpcs.VPCS.stream_pcap_file"):
# response = await client.get("/projects/{project_id}/vpcs/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True)
# response = await compute_client.get("/projects/{project_id}/vpcs/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]), raw=True)
# assert response.status_code == status.HTTP_200_OK
tests/api/routes/controller/test_computes.py
-21
View File
@@ -190,24 +190,3 @@ class TestComputeFeatures:
# response = await client.post(app.url_path_for("autoidlepc", compute_id=compute_id) + "/auto_idlepc", json=params)
# assert mock.called
# assert response.status_code == status.HTTP_200_OK
# FIXME
# @pytest.mark.asyncio
# async def test_compute_endpoint(controller_api):
#
# params = {
# "compute_id": "my_compute",
# "protocol": "http",
# "host": "localhost",
# "port": 84,
# "user": "julien",
# "password": "secure"
# }
#
# response = await controller_api.post("/computes", params)
# assert response.status_code == 201
#
# response = await controller_api.get("/computes/endpoint/my_compute/qemu/images")
# assert response.status_code == 200
# assert response.json['endpoint'] == 'http://localhost:84/v2/compute/qemu/images'