mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-08-27 20:40:13 +03:00
4e30b6905a
Dynamips.create_nio is async def while BaseManager.create_nio is a sync def. The previous fix only added the extra 'node' argument but did not await the resulting coroutine, causing 'was never awaited' warnings and passing a coroutine object instead of an NIO instance to the binding dispatch. Add 'await' on the Dynamips branch. Test updated to verify both the async nature and the parameter count.
345 lines
14 KiB
Python
345 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
|
#
|
|
# Copyright (C) 2020 GNS3 Technologies Inc.
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import pytest
|
|
import uuid
|
|
import os
|
|
|
|
from unittest.mock import patch
|
|
from tests.utils import asyncio_patch
|
|
from fastapi import FastAPI, status
|
|
from httpx import AsyncClient
|
|
|
|
from gns3server.compute.project_manager import ProjectManager
|
|
from gns3server.compute.project import Project
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
class TestComputeProjectRoutes:
|
|
|
|
@pytest.fixture
|
|
def base_params(self, tmpdir) -> dict:
|
|
"""Return standard parameters"""
|
|
|
|
params = {
|
|
"name": "test",
|
|
"project_id": str(uuid.uuid4())
|
|
}
|
|
return params
|
|
|
|
|
|
async def test_create_project_without_dir(
|
|
self,
|
|
app: FastAPI,
|
|
compute_client: AsyncClient,
|
|
base_params: dict
|
|
) -> None:
|
|
|
|
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(
|
|
self,
|
|
app: FastAPI,
|
|
compute_client: AsyncClient,
|
|
base_params: dict
|
|
) -> None:
|
|
|
|
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 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
|
|
assert response.json()["project_id"] == base_params["project_id"]
|
|
assert response.json()["name"] == base_params["name"]
|
|
assert response.json()["variables"] is None
|
|
|
|
|
|
async def test_show_project_invalid_uuid(self, app: FastAPI, compute_client: AsyncClient) -> None:
|
|
|
|
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(self, app: FastAPI, compute_client: AsyncClient) -> dict:
|
|
|
|
ProjectManager.instance()._projects = {}
|
|
|
|
params = {"name": "test", "project_id": "51010203-0405-0607-0809-0a0b0c0d0e0f"}
|
|
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 compute_client.post(app.url_path_for("compute:create_compute_project"), json=params)
|
|
assert response.status_code == status.HTTP_201_CREATED
|
|
|
|
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(self, app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None:
|
|
|
|
with asyncio_patch("gns3server.compute.project.Project.delete", return_value=True) as mock:
|
|
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(self, app: FastAPI, compute_client: AsyncClient, base_params: dict) -> None:
|
|
|
|
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 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(self, app: FastAPI, compute_client: AsyncClient) -> None:
|
|
|
|
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(self, app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None:
|
|
|
|
with asyncio_patch("gns3server.compute.project.Project.close", return_value=True) as mock:
|
|
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
|
|
|
|
|
|
# @pytest.mark.asyncio
|
|
# async def test_close_project_two_client_connected(compute_api, compute_project):
|
|
#
|
|
# ProjectHandler._notifications_listening = {compute_project.id: 2}
|
|
# with asyncio_patch("gns3server.compute.project.Project.close", return_value=True) as mock:
|
|
# response = await compute_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(self, app: FastAPI, compute_client: AsyncClient) -> None:
|
|
|
|
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(self, app: FastAPI, compute_client: AsyncClient) -> None:
|
|
|
|
project = ProjectManager.instance().create_project(project_id="01010203-0405-0607-0809-0a0b0c0d0e0b")
|
|
|
|
with open(os.path.join(project.path, "hello"), "w+") as f:
|
|
f.write("world")
|
|
|
|
response = await compute_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 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 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(
|
|
self,
|
|
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 compute_client.get(
|
|
app.url_path_for(
|
|
"compute:get_compute_project_file",
|
|
project_id=project.id,
|
|
file_path=file_path
|
|
)
|
|
)
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
|
|
|
|
async def test_write_file(self, 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 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
|
|
|
|
with open(os.path.join(project.path, "hello")) as f:
|
|
assert f.read() == "world"
|
|
|
|
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(
|
|
self,
|
|
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 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
|
|
|
|
|
|
class TestBatchNIOEdgeCases:
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dynamips_router_dispatch_to_slot_add_nio_binding(self):
|
|
"""_add_nio_binding dispatches Dynamips router to slot_add_nio_binding."""
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from gns3server.api.routes.compute.projects import _add_nio_binding
|
|
|
|
node = MagicMock()
|
|
type(node.manager).__name__ = "Dynamips"
|
|
node.slot_add_nio_binding = AsyncMock()
|
|
nio = MagicMock()
|
|
|
|
await _add_nio_binding(node, 0, 0, nio)
|
|
node.slot_add_nio_binding.assert_called_once_with(0, 0, nio)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dynamips_switch_dispatch_to_add_nio(self):
|
|
"""_add_nio_binding dispatches Dynamips switch to add_nio."""
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from gns3server.api.routes.compute.projects import _add_nio_binding
|
|
|
|
node = MagicMock()
|
|
type(node.manager).__name__ = "Dynamips"
|
|
del node.slot_add_nio_binding # no slot_add_nio → switch path
|
|
node.add_nio = AsyncMock()
|
|
nio = MagicMock()
|
|
|
|
await _add_nio_binding(node, 0, 0, nio)
|
|
node.add_nio.assert_called_once_with(nio, 0)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dynamips_create_nio_is_async_and_needs_await(self):
|
|
"""
|
|
Dynamips.create_nio is async (returns a coroutine) unlike the sync
|
|
base version. The batch handler must await it.
|
|
"""
|
|
import inspect
|
|
import asyncio as _asyncio
|
|
|
|
class _FakeDynamips:
|
|
async def create_nio(self, node, nio_settings):
|
|
return {"type": "nio_udp", "node": node}
|
|
|
|
class _FakeBase:
|
|
def create_nio(self, nio_settings):
|
|
return {"type": "nio_udp"}
|
|
|
|
dyn = _FakeDynamips()
|
|
base = _FakeBase()
|
|
assert len(inspect.signature(dyn.create_nio).parameters) == 2 # Dynamips
|
|
assert len(inspect.signature(base.create_nio).parameters) == 1 # standard
|
|
assert inspect.iscoroutinefunction(dyn.create_nio)
|
|
assert not inspect.iscoroutinefunction(base.create_nio)
|
|
|
|
# Verify the batch logic: 2 params → await, 1 param → no await
|
|
d_result = await dyn.create_nio("r1", {"type": "nio_udp"})
|
|
b_result = base.create_nio({"type": "nio_udp"})
|
|
assert d_result["node"] == "r1"
|
|
assert b_result["type"] == "nio_udp"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_qemu_dispatch_to_adapter_add_nio_binding(self):
|
|
"""_add_nio_binding dispatches Qemu to adapter_add_nio_binding."""
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from gns3server.api.routes.compute.projects import _add_nio_binding
|
|
|
|
node = MagicMock()
|
|
type(node.manager).__name__ = "Qemu"
|
|
node.adapter_add_nio_binding = AsyncMock()
|
|
nio = MagicMock()
|
|
|
|
await _add_nio_binding(node, 0, 0, nio)
|
|
node.adapter_add_nio_binding.assert_called_once_with(0, nio)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_iou_dispatch_to_adapter_add_nio_binding(self):
|
|
"""_add_nio_binding dispatches IOU to adapter_add_nio_binding(adapter, port, nio)."""
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from gns3server.api.routes.compute.projects import _add_nio_binding
|
|
|
|
node = MagicMock()
|
|
type(node.manager).__name__ = "IOU"
|
|
node.adapter_add_nio_binding = AsyncMock()
|
|
nio = MagicMock()
|
|
|
|
await _add_nio_binding(node, 1, 2, nio)
|
|
node.adapter_add_nio_binding.assert_called_once_with(1, 2, nio)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_vpcs_dispatch_to_port_add_nio_binding(self):
|
|
"""_add_nio_binding dispatches VPCS to port_add_nio_binding."""
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from gns3server.api.routes.compute.projects import _add_nio_binding
|
|
|
|
node = MagicMock()
|
|
type(node.manager).__name__ = "VPCS"
|
|
node.port_add_nio_binding = AsyncMock()
|
|
nio = MagicMock()
|
|
|
|
await _add_nio_binding(node, 0, 3, nio)
|
|
node.port_add_nio_binding.assert_called_once_with(3, nio)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_builtin_dispatch_to_add_nio(self):
|
|
"""_add_nio_binding dispatches Builtin nodes to add_nio(nio, port)."""
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from gns3server.api.routes.compute.projects import _add_nio_binding
|
|
|
|
node = MagicMock()
|
|
type(node.manager).__name__ = "Builtin"
|
|
node.add_nio = AsyncMock()
|
|
nio = MagicMock()
|
|
|
|
await _add_nio_binding(node, 0, 0, nio)
|
|
node.add_nio.assert_called_once_with(nio, 0)
|