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

Refactor tests and upgrade dev package requirements

Browse Source
This commit is contained in:
grossmj committed 2025-01-17 19:12:44 +10:00
1 parent f6725a37dd
commit ba4e0c945d
29 files changed
+5074 -4558
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/api/routes/controller/test_controller.py
+41 -39
View File
@@ -27,43 +27,45 @@ from gns3server.config import Config
pytestmark = pytest.mark.asyncio
async def test_shutdown_local(app: FastAPI, client: AsyncClient, config: Config) -> None:
class TestControllerRoutes:
os.kill = MagicMock()
config.settings.Server.local = True
response = await client.post(app.url_path_for("shutdown"))
assert response.status_code == status.HTTP_204_NO_CONTENT
assert os.kill.called
async def test_shutdown_non_local(app: FastAPI, client: AsyncClient, config: Config) -> None:
response = await client.post(app.url_path_for("shutdown"))
assert response.status_code == status.HTTP_403_FORBIDDEN
# @pytest.mark.asyncio
# async def test_debug(controller_api, config, tmpdir):
#
# config._main_config_file = str(tmpdir / "test.conf")
# config.set("Server", "local", True)
# response = await controller_api.post('/debug')
# assert response.status_code == 201
# debug_dir = os.path.join(config.config_dir, "debug")
# assert os.path.exists(debug_dir)
# assert os.path.exists(os.path.join(debug_dir, "controller.txt"))
#
#
# @pytest.mark.asyncio
# async def test_debug_non_local(controller_api, config, tmpdir):
#
# config._main_config_file = str(tmpdir / "test.conf")
# config.set("Server", "local", False)
# response = await controller_api.post('/debug')
# assert response.status_code == 403
async def test_statistics_output(app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("statistics"))
assert response.status_code == status.HTTP_200_OK
async def test_shutdown_local(self, app: FastAPI, client: AsyncClient, config: Config) -> None:
os.kill = MagicMock()
config.settings.Server.local = True
response = await client.post(app.url_path_for("shutdown"))
assert response.status_code == status.HTTP_204_NO_CONTENT
assert os.kill.called
async def test_shutdown_non_local(self, app: FastAPI, client: AsyncClient, config: Config) -> None:
response = await client.post(app.url_path_for("shutdown"))
assert response.status_code == status.HTTP_403_FORBIDDEN
# @pytest.mark.asyncio
# async def test_debug(controller_api, config, tmpdir):
#
# config._main_config_file = str(tmpdir / "test.conf")
# config.set("Server", "local", True)
# response = await controller_api.post('/debug')
# assert response.status_code == 201
# debug_dir = os.path.join(config.config_dir, "debug")
# assert os.path.exists(debug_dir)
# assert os.path.exists(os.path.join(debug_dir, "controller.txt"))
#
#
# @pytest.mark.asyncio
# async def test_debug_non_local(controller_api, config, tmpdir):
#
# config._main_config_file = str(tmpdir / "test.conf")
# config.set("Server", "local", False)
# response = await controller_api.post('/debug')
# assert response.status_code == 403
async def test_statistics_output(self, app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("statistics"))
assert response.status_code == status.HTTP_200_OK
tests/api/routes/controller/test_drawings.py
+87 -85
View File
@@ -26,89 +26,91 @@ from gns3server.controller.project import Project
pytestmark = pytest.mark.asyncio
async def test_create_drawing(app: FastAPI, client: AsyncClient, project: Project) -> None:
class TestDrawingsRoutes:
params = {
"svg": '<svg height="210" width="500"><line x1="0" y1="0" x2="200" y2="200" style="stroke:rgb(255,0,0);stroke-width:2" /></svg>',
"x": 10,
"y": 20,
"z": 0
}
response = await client.post(app.url_path_for("create_drawing", project_id=project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["drawing_id"] is not None
async def test_get_drawing(app: FastAPI, client: AsyncClient, project: Project) -> None:
params = {
"svg": '<svg height="210" width="500"><line x1="0" y1="0" x2="200" y2="200" style="stroke:rgb(255,0,0);stroke-width:2" /></svg>',
"x": 10,
"y": 20,
"z": 0
}
response = await client.post(app.url_path_for("create_drawing", project_id=project.id), json=params)
response = await client.get(app.url_path_for(
"get_drawing",
project_id=project.id,
drawing_id=response.json()["drawing_id"])
)
assert response.status_code == 200
assert response.json()["x"] == 10
async def test_update_drawing(app: FastAPI, client: AsyncClient, project: Project) -> None:
params = {
"svg": '<svg height="210" width="500"><line x1="0" y1="0" x2="200" y2="200" style="stroke:rgb(255,0,0);stroke-width:2" /></svg>',
"x": 10,
"y": 20,
"z": 0
}
response = await client.post(app.url_path_for("create_drawing", project_id=project.id), json=params)
response = await client.put(app.url_path_for(
"update_drawing",
project_id=project.id,
drawing_id=response.json()["drawing_id"]),
json={"x": 42}
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["x"] == 42
async def test_all_drawings(app: FastAPI, client: AsyncClient, project: Project) -> None:
params = {
"svg": '<svg height="210" width="500"><line x1="0" y1="0" x2="200" y2="200" style="stroke:rgb(255,0,0);stroke-width:2" /></svg>',
"x": 10,
"y": 20,
"z": 0
}
await client.post(app.url_path_for("create_drawing", project_id=project.id), json=params)
response = await client.get(app.url_path_for("get_drawings", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 1
# test listing links from a closed project
await project.close(ignore_notification=True)
response = await client.get(app.url_path_for("get_drawings", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 1
async def test_delete_drawing(app: FastAPI, client: AsyncClient, project: Project) -> None:
drawing = Drawing(project)
project._drawings = {drawing.id: drawing}
response = await client.delete(app.url_path_for(
"delete_drawing",
project_id=project.id,
drawing_id=drawing.id)
)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert drawing.id not in project.drawings
async def test_create_drawing(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
params = {
"svg": '<svg height="210" width="500"><line x1="0" y1="0" x2="200" y2="200" style="stroke:rgb(255,0,0);stroke-width:2" /></svg>',
"x": 10,
"y": 20,
"z": 0
}
response = await client.post(app.url_path_for("create_drawing", project_id=project.id), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["drawing_id"] is not None
async def test_get_drawing(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
params = {
"svg": '<svg height="210" width="500"><line x1="0" y1="0" x2="200" y2="200" style="stroke:rgb(255,0,0);stroke-width:2" /></svg>',
"x": 10,
"y": 20,
"z": 0
}
response = await client.post(app.url_path_for("create_drawing", project_id=project.id), json=params)
response = await client.get(app.url_path_for(
"get_drawing",
project_id=project.id,
drawing_id=response.json()["drawing_id"])
)
assert response.status_code == 200
assert response.json()["x"] == 10
async def test_update_drawing(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
params = {
"svg": '<svg height="210" width="500"><line x1="0" y1="0" x2="200" y2="200" style="stroke:rgb(255,0,0);stroke-width:2" /></svg>',
"x": 10,
"y": 20,
"z": 0
}
response = await client.post(app.url_path_for("create_drawing", project_id=project.id), json=params)
response = await client.put(app.url_path_for(
"update_drawing",
project_id=project.id,
drawing_id=response.json()["drawing_id"]),
json={"x": 42}
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["x"] == 42
async def test_all_drawings(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
params = {
"svg": '<svg height="210" width="500"><line x1="0" y1="0" x2="200" y2="200" style="stroke:rgb(255,0,0);stroke-width:2" /></svg>',
"x": 10,
"y": 20,
"z": 0
}
await client.post(app.url_path_for("create_drawing", project_id=project.id), json=params)
response = await client.get(app.url_path_for("get_drawings", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 1
# test listing links from a closed project
await project.close(ignore_notification=True)
response = await client.get(app.url_path_for("get_drawings", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 1
async def test_delete_drawing(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
drawing = Drawing(project)
project._drawings = {drawing.id: drawing}
response = await client.delete(app.url_path_for(
"delete_drawing",
project_id=project.id,
drawing_id=drawing.id)
)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert drawing.id not in project.drawings
tests/api/routes/controller/test_gns3vm.py
+30 -28
View File
@@ -24,32 +24,34 @@ from tests.utils import asyncio_patch
pytestmark = pytest.mark.asyncio
async def test_list_vms(app: FastAPI, client: AsyncClient) -> None:
class TestGNS3VMRoutes:
with asyncio_patch("gns3server.controller.gns3vm.vmware_gns3_vm.VMwareGNS3VM.list", return_value=[{"vmname": "test"}]):
response = await client.get(app.url_path_for("get_vms", engine="vmware"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == [
{
"vmname": "test"
}
]
async def test_engines(app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("get_engines"))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) > 0
async def test_put_gns3vm(app: FastAPI, client: AsyncClient) -> None:
response = await client.put(app.url_path_for("update_gns3vm_settings"), json={"vmname": "TEST VM"})
assert response.status_code == status.HTTP_200_OK
assert response.json()["vmname"] == "TEST VM"
async def test_get_gns3vm(app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("get_gns3vm_settings"))
assert response.status_code == status.HTTP_200_OK
async def test_list_vms(self, app: FastAPI, client: AsyncClient) -> None:
with asyncio_patch("gns3server.controller.gns3vm.vmware_gns3_vm.VMwareGNS3VM.list", return_value=[{"vmname": "test"}]):
response = await client.get(app.url_path_for("get_vms", engine="vmware"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == [
{
"vmname": "test"
}
]
async def test_engines(self, app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("get_engines"))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) > 0
async def test_put_gns3vm(self, app: FastAPI, client: AsyncClient) -> None:
response = await client.put(app.url_path_for("update_gns3vm_settings"), json={"vmname": "TEST VM"})
assert response.status_code == status.HTTP_200_OK
assert response.json()["vmname"] == "TEST VM"
async def test_get_gns3vm(self, app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("get_gns3vm_settings"))
assert response.status_code == status.HTTP_200_OK
tests/api/routes/controller/test_links.py
+356 -318
View File
@@ -35,30 +35,88 @@ from gns3server.controller.udp_link import UDPLink
pytestmark = pytest.mark.asyncio
@pytest_asyncio.fixture
async def nodes(compute: Compute, project: Project) -> Tuple[Node, Node]:
class TestLinkRoutes:
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
node1 = await project.add_node(compute, "node1", None, node_type="qemu")
node1._ports = [EthernetPort("E0", 0, 0, 3)]
node2 = await project.add_node(compute, "node2", None, node_type="qemu")
node2._ports = [EthernetPort("E0", 0, 2, 4)]
return node1, node2
async def test_create_link(app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]) -> None:
node1, node2 = nodes
filters = {
"latency": [10],
"frequency_drop": [50]
}
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
@pytest_asyncio.fixture
async def nodes(self, compute: Compute, project: Project) -> Tuple[Node, Node]:
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
node1 = await project.add_node(compute, "node1", None, node_type="qemu")
node1._ports = [EthernetPort("E0", 0, 0, 3)]
node2 = await project.add_node(compute, "node2", None, node_type="qemu")
node2._ports = [EthernetPort("E0", 0, 2, 4)]
return node1, node2
async def test_create_link(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
nodes: Tuple[Node, Node]
) -> None:
node1, node2 = nodes
filters = {
"latency": [10],
"frequency_drop": [50]
}
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
response = await client.post(app.url_path_for("create_link", project_id=project.id), json={
"nodes": [
{
"node_id": node1.id,
"adapter_number": 0,
"port_number": 3,
"label": {
"text": "Text",
"x": 42,
"y": 0
}
},
{
"node_id": node2.id,
"adapter_number": 2,
"port_number": 4
}
],
"filters": filters
})
assert mock.called
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["link_id"] is not None
assert len(response.json()["nodes"]) == 2
assert response.json()["nodes"][0]["label"]["x"] == 42
assert len(project.links) == 1
assert list(project.links.values())[0].filters == filters
async def test_create_link_failure(
self,
app: FastAPI,
client: AsyncClient,
compute: Compute,
project: Project
) -> None:
"""
Make sure the link is deleted if we failed to create it.
The failure is triggered by connecting the link to itself
"""
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
node1 = await project.add_node(compute, "node1", None, node_type="qemu")
node1._ports = [EthernetPort("E0", 0, 0, 3), EthernetPort("E0", 0, 0, 4)]
response = await client.post(app.url_path_for("create_link", project_id=project.id), json={
"nodes": [
{
@@ -71,6 +129,168 @@ async def test_create_link(app: FastAPI, client: AsyncClient, project: Project,
"y": 0
}
},
{
"node_id": node1.id,
"adapter_number": 0,
"port_number": 4
}
]
})
assert response.status_code == status.HTTP_409_CONFLICT
assert len(project.links) == 0
async def test_get_link(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
nodes: Tuple[Node, Node]
) -> None:
node1, node2 = nodes
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
response = await client.post(app.url_path_for("create_link", project_id=project.id), json={
"nodes": [
{
"node_id": node1.id,
"adapter_number": 0,
"port_number": 3,
"label": {
"text": "Text",
"x": 42,
"y": 0
}
},
{
"node_id": node2.id,
"adapter_number": 2,
"port_number": 4
}
]
})
assert mock.called
link_id = response.json()["link_id"]
assert response.json()["nodes"][0]["label"]["x"] == 42
response = await client.get(app.url_path_for("get_link", project_id=project.id, link_id=link_id))
assert response.status_code == status.HTTP_200_OK
assert response.json()["nodes"][0]["label"]["x"] == 42
async def test_update_link_suspend(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
nodes: Tuple[Node, Node]
) -> None:
node1, node2 = nodes
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
response = await client.post(app.url_path_for("create_link", project_id=project.id), json={
"nodes": [
{
"node_id": node1.id,
"adapter_number": 0,
"port_number": 3,
"label": {
"text": "Text",
"x": 42,
"y": 0
}
},
{
"node_id": node2.id,
"adapter_number": 2,
"port_number": 4
}
]
})
assert mock.called
link_id = response.json()["link_id"]
assert response.json()["nodes"][0]["label"]["x"] == 42
response = await client.put(app.url_path_for("update_link", project_id=project.id, link_id=link_id), json={
"nodes": [
{
"node_id": node1.id,
"adapter_number": 0,
"port_number": 3,
"label": {
"text": "Hello",
"x": 64,
"y": 0
}
},
{
"node_id": node2.id,
"adapter_number": 2,
"port_number": 4
}
],
"suspend": True
})
assert response.status_code == status.HTTP_200_OK
assert response.json()["nodes"][0]["label"]["x"] == 64
assert response.json()["suspend"]
assert response.json()["filters"] == {}
async def test_update_link(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
nodes: Tuple[Node, Node]
) -> None:
filters = {
"latency": [10],
"frequency_drop": [50]
}
node1, node2 = nodes
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
response = await client.post(app.url_path_for("create_link", project_id=project.id), json={
"nodes": [
{
"node_id": node1.id,
"adapter_number": 0,
"port_number": 3,
"label": {
"text": "Text",
"x": 42,
"y": 0
}
},
{
"node_id": node2.id,
"adapter_number": 2,
"port_number": 4
}
]
})
assert mock.called
link_id = response.json()["link_id"]
assert response.json()["nodes"][0]["label"]["x"] == 42
response = await client.put(app.url_path_for("update_link", project_id=project.id, link_id=link_id), json={
"nodes": [
{
"node_id": node1.id,
"adapter_number": 0,
"port_number": 3,
"label": {
"text": "Hello",
"x": 64,
"y": 0
}
},
{
"node_id": node2.id,
"adapter_number": 2,
@@ -79,309 +299,127 @@ async def test_create_link(app: FastAPI, client: AsyncClient, project: Project,
],
"filters": filters
})
assert mock.called
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["link_id"] is not None
assert len(response.json()["nodes"]) == 2
assert response.json()["nodes"][0]["label"]["x"] == 42
assert len(project.links) == 1
assert list(project.links.values())[0].filters == filters
async def test_create_link_failure(app: FastAPI, client: AsyncClient, compute: Compute, project: Project) -> None:
"""
Make sure the link is deleted if we failed to create it.
The failure is triggered by connecting the link to itself
"""
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
node1 = await project.add_node(compute, "node1", None, node_type="qemu")
node1._ports = [EthernetPort("E0", 0, 0, 3), EthernetPort("E0", 0, 0, 4)]
response = await client.post(app.url_path_for("create_link", project_id=project.id), json={
"nodes": [
assert response.status_code == status.HTTP_200_OK
assert response.json()["nodes"][0]["label"]["x"] == 64
assert list(project.links.values())[0].filters == filters
async def test_list_link(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
nodes: Tuple[Node, Node]
) -> None:
filters = {
"latency": [10],
"frequency_drop": [50]
}
node1, node2 = nodes
nodes = [
{
"node_id": node1.id,
"adapter_number": 0,
"port_number": 3,
"label": {
"text": "Text",
"x": 42,
"y": 0
}
"port_number": 3
},
{
"node_id": node1.id,
"adapter_number": 0,
"node_id": node2.id,
"adapter_number": 2,
"port_number": 4
}
]
})
assert response.status_code == status.HTTP_409_CONFLICT
assert len(project.links) == 0
async def test_get_link(app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]) -> None:
node1, node2 = nodes
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
response = await client.post(app.url_path_for("create_link", project_id=project.id), json={
"nodes": [
{
"node_id": node1.id,
"adapter_number": 0,
"port_number": 3,
"label": {
"text": "Text",
"x": 42,
"y": 0
}
},
{
"node_id": node2.id,
"adapter_number": 2,
"port_number": 4
}
]
})
assert mock.called
link_id = response.json()["link_id"]
assert response.json()["nodes"][0]["label"]["x"] == 42
response = await client.get(app.url_path_for("get_link", project_id=project.id, link_id=link_id))
assert response.status_code == status.HTTP_200_OK
assert response.json()["nodes"][0]["label"]["x"] == 42
async def test_update_link_suspend(app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]) -> None:
node1, node2 = nodes
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
response = await client.post(app.url_path_for("create_link", project_id=project.id), json={
"nodes": [
{
"node_id": node1.id,
"adapter_number": 0,
"port_number": 3,
"label": {
"text": "Text",
"x": 42,
"y": 0
}
},
{
"node_id": node2.id,
"adapter_number": 2,
"port_number": 4
}
]
})
assert mock.called
link_id = response.json()["link_id"]
assert response.json()["nodes"][0]["label"]["x"] == 42
response = await client.put(app.url_path_for("update_link", project_id=project.id, link_id=link_id), json={
"nodes": [
{
"node_id": node1.id,
"adapter_number": 0,
"port_number": 3,
"label": {
"text": "Hello",
"x": 64,
"y": 0
}
},
{
"node_id": node2.id,
"adapter_number": 2,
"port_number": 4
}
],
"suspend": True
})
assert response.status_code == status.HTTP_200_OK
assert response.json()["nodes"][0]["label"]["x"] == 64
assert response.json()["suspend"]
assert response.json()["filters"] == {}
async def test_update_link(app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]) -> None:
filters = {
"latency": [10],
"frequency_drop": [50]
}
node1, node2 = nodes
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
response = await client.post(app.url_path_for("create_link", project_id=project.id), json={
"nodes": [
{
"node_id": node1.id,
"adapter_number": 0,
"port_number": 3,
"label": {
"text": "Text",
"x": 42,
"y": 0
}
},
{
"node_id": node2.id,
"adapter_number": 2,
"port_number": 4
}
]
})
assert mock.called
link_id = response.json()["link_id"]
assert response.json()["nodes"][0]["label"]["x"] == 42
response = await client.put(app.url_path_for("update_link", project_id=project.id, link_id=link_id), json={
"nodes": [
{
"node_id": node1.id,
"adapter_number": 0,
"port_number": 3,
"label": {
"text": "Hello",
"x": 64,
"y": 0
}
},
{
"node_id": node2.id,
"adapter_number": 2,
"port_number": 4
}
],
"filters": filters
})
assert response.status_code == status.HTTP_200_OK
assert response.json()["nodes"][0]["label"]["x"] == 64
assert list(project.links.values())[0].filters == filters
async def test_list_link(app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]) -> None:
filters = {
"latency": [10],
"frequency_drop": [50]
}
node1, node2 = nodes
nodes = [
{
"node_id": node1.id,
"adapter_number": 0,
"port_number": 3
},
{
"node_id": node2.id,
"adapter_number": 2,
"port_number": 4
}
]
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
await client.post(app.url_path_for("create_link", project_id=project.id), json={
"nodes": nodes,
"filters": filters
})
assert mock.called
response = await client.get(app.url_path_for("get_links", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 1
assert response.json()[0]["filters"] == filters
# test listing links from a closed project
await project.close(ignore_notification=True)
response = await client.get(app.url_path_for("get_links", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 1
assert response.json()[0]["filters"] == filters
async def test_reset_link(app: FastAPI, client: AsyncClient, project: Project) -> None:
link = UDPLink(project)
project._links = {link.id: link}
with asyncio_patch("gns3server.controller.udp_link.UDPLink.delete") as delete_mock:
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as create_mock:
response = await client.post(app.url_path_for("reset_link", project_id=project.id, link_id=link.id))
assert delete_mock.called
assert create_mock.called
assert response.status_code == status.HTTP_200_OK
async def test_start_capture(app: FastAPI, client: AsyncClient, project: Project) -> None:
link = Link(project)
project._links = {link.id: link}
with asyncio_patch("gns3server.controller.link.Link.start_capture") as mock:
response = await client.post(app.url_path_for("start_capture", project_id=project.id, link_id=link.id), json={})
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
await client.post(app.url_path_for("create_link", project_id=project.id), json={
"nodes": nodes,
"filters": filters
})
assert mock.called
assert response.status_code == status.HTTP_201_CREATED
async def test_stop_capture(app: FastAPI, client: AsyncClient, project: Project) -> None:
link = Link(project)
project._links = {link.id: link}
with asyncio_patch("gns3server.controller.link.Link.stop_capture") as mock:
response = await client.post(app.url_path_for("stop_capture", project_id=project.id, link_id=link.id))
response = await client.get(app.url_path_for("get_links", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 1
assert response.json()[0]["filters"] == filters
# test listing links from a closed project
await project.close(ignore_notification=True)
response = await client.get(app.url_path_for("get_links", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 1
assert response.json()[0]["filters"] == filters
async def test_reset_link(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = UDPLink(project)
project._links = {link.id: link}
with asyncio_patch("gns3server.controller.udp_link.UDPLink.delete") as delete_mock:
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as create_mock:
response = await client.post(app.url_path_for("reset_link", project_id=project.id, link_id=link.id))
assert delete_mock.called
assert create_mock.called
assert response.status_code == status.HTTP_200_OK
async def test_start_capture(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = Link(project)
project._links = {link.id: link}
with asyncio_patch("gns3server.controller.link.Link.start_capture") as mock:
response = await client.post(app.url_path_for("start_capture", project_id=project.id, link_id=link.id), json={})
assert mock.called
assert response.status_code == status.HTTP_201_CREATED
async def test_stop_capture(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = Link(project)
project._links = {link.id: link}
with asyncio_patch("gns3server.controller.link.Link.stop_capture") as mock:
response = await client.post(app.url_path_for("stop_capture", project_id=project.id, link_id=link.id))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
# async def test_pcap(controller_api, http_client, project):
#
# async def pcap_capture():
# async with http_client.get(controller_api.get_url("/projects/{}/links/{}/pcap".format(project.id, link.id))) as response:
# response.body = await response.content.read(5)
# print("READ", response.body)
# return response
#
# with asyncio_patch("gns3server.controller.link.Link.capture_node") as mock:
# link = Link(project)
# link._capture_file_name = "test"
# link._capturing = True
# with open(link.capture_file_path, "w+") as f:
# f.write("hello")
# project._links = {link.id: link}
# response = await pcap_capture()
# assert mock.called
# assert response.status_code == 200
# assert b'hello' == response.body
async def test_delete_link(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = Link(project)
project._links = {link.id: link}
with asyncio_patch("gns3server.controller.link.Link.delete") as mock:
response = await client.delete(app.url_path_for("delete_link", project_id=project.id, link_id=link.id))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
# async def test_pcap(controller_api, http_client, project):
#
# async def pcap_capture():
# async with http_client.get(controller_api.get_url("/projects/{}/links/{}/pcap".format(project.id, link.id))) as response:
# response.body = await response.content.read(5)
# print("READ", response.body)
# return response
#
# with asyncio_patch("gns3server.controller.link.Link.capture_node") as mock:
# link = Link(project)
# link._capture_file_name = "test"
# link._capturing = True
# with open(link.capture_file_path, "w+") as f:
# f.write("hello")
# project._links = {link.id: link}
# response = await pcap_capture()
# assert mock.called
# assert response.status_code == 200
# assert b'hello' == response.body
async def test_delete_link(app: FastAPI, client: AsyncClient, project: Project) -> None:
link = Link(project)
project._links = {link.id: link}
with asyncio_patch("gns3server.controller.link.Link.delete") as mock:
response = await client.delete(app.url_path_for("delete_link", project_id=project.id, link_id=link.id))
assert mock.called
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_list_filters(app: FastAPI, client: AsyncClient, project: Project) -> None:
link = Link(project)
project._links = {link.id: link}
with patch("gns3server.controller.link.Link.available_filters", return_value=FILTERS) as mock:
response = await client.get(app.url_path_for("get_filters", project_id=project.id, link_id=link.id))
assert mock.called
assert response.status_code == status.HTTP_200_OK
assert response.json() == FILTERS
async def test_list_filters(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = Link(project)
project._links = {link.id: link}
with patch("gns3server.controller.link.Link.available_filters", return_value=FILTERS) as mock:
response = await client.get(app.url_path_for("get_filters", project_id=project.id, link_id=link.id))
assert mock.called
assert response.status_code == status.HTTP_200_OK
assert response.json() == FILTERS
tests/api/routes/controller/test_nodes.py
+533 -412
View File
@@ -31,416 +31,537 @@ from gns3server.controller.compute import Compute
pytestmark = pytest.mark.asyncio
@pytest.fixture
def node(project: Project, compute: Compute) -> Node:
node = Node(project, compute, "test", node_type="vpcs")
project._nodes[node.id] = node
return node
async def test_create_node(app: FastAPI, client: AsyncClient, project: Project, compute: Compute) -> None:
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
response = await client.post(app.url_path_for("create_node", project_id=project.id), json={
"name": "test",
"node_type": "vpcs",
"compute_id": "example.com",
"properties": {
"startup_script": "echo test"
}
})
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "test"
assert "name" not in response.json()["properties"]
async def test_list_node(app: FastAPI, client: AsyncClient, project: Project, compute: Compute) -> None:
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
await client.post(app.url_path_for("create_node", project_id=project.id), json={
"name": "test",
"node_type": "vpcs",
"compute_id": "example.com",
"properties": {
"startup_script": "echo test"
}
})
response = await client.get(app.url_path_for("get_nodes", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert response.json()[0]["name"] == "test"
# test listing nodes from a closed project
await project.close(ignore_notification=True)
response = await client.get(app.url_path_for("get_nodes", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert response.json()[0]["name"] == "test"
async def test_get_node(app: FastAPI, client: AsyncClient, project: Project, compute: Compute) -> None:
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
response = await client.post(app.url_path_for("create_node", project_id=project.id), json={
"name": "test",
"node_type": "vpcs",
"compute_id": "example.com",
"properties": {
"startup_script": "echo test"
}
})
response = await client.get(app.url_path_for("get_node", project_id=project.id, node_id=response.json()["node_id"]))
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "test"
async def test_update_node(app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node) -> None:
response = MagicMock()
response.json = {"console": 2048}
compute.put = AsyncioMagicMock(return_value=response)
response = await client.put(app.url_path_for("update_node", project_id=project.id, node_id=node.id), json={
"name": "test",
"node_type": "vpcs",
"compute_id": "example.com",
"properties": {
"startup_script": "echo test"
}
})
assert response.status_code == 200
assert response.json()["name"] == "test"
assert "name" not in response.json()["properties"]
async def test_start_all_nodes(app: FastAPI, client: AsyncClient, project: Project, compute: Compute) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("start_all_nodes", project_id=project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_stop_all_nodes(app: FastAPI, client: AsyncClient, project: Project, compute: Compute) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("stop_all_nodes", project_id=project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_suspend_all_nodes(app: FastAPI, client: AsyncClient, project: Project, compute: Compute) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("suspend_all_nodes", project_id=project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_reload_all_nodes(app: FastAPI, client: AsyncClient, project: Project, compute: Compute) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("reload_all_nodes", project_id=project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_reset_console_all_nodes(app: FastAPI, client: AsyncClient, project: Project, compute: Compute) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("reset_console_all_nodes", project_id=project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_start_node(app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("start_node", project_id=project.id, node_id=node.id), json={})
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_stop_node(app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("stop_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_suspend_node(app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("suspend_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_reload_node(app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node):
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("reload_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_isolate_node(app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node):
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("isolate_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_unisolate_node(app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node):
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("unisolate_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_duplicate_node(
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node) -> None:
response = MagicMock()
response.json({"console": 2035})
compute.post = AsyncioMagicMock(return_value=response)
response = await client.post(app.url_path_for("duplicate_node", project_id=project.id, node_id=node.id),
json={"x": 10, "y": 5, "z": 0})
assert response.status_code == status.HTTP_201_CREATED
async def test_delete_node(app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node) -> None:
compute.post = AsyncioMagicMock()
response = await client.delete(app.url_path_for("delete_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_dynamips_idle_pc(
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = MagicMock()
response.json = {"idlepc": "0x60606f54"}
compute.get = AsyncioMagicMock(return_value=response)
node._node_type = "dynamips" # force Dynamips node type
response = await client.get(app.url_path_for("auto_idlepc", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_200_OK
assert response.json()["idlepc"] == "0x60606f54"
async def test_dynamips_idle_pc_wrong_node_type(
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = await client.get(app.url_path_for("auto_idlepc", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_400_BAD_REQUEST
async def test_dynamips_idlepc_proposals(
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = MagicMock()
response.json = ["0x60606f54", "0x33805a22"]
compute.get = AsyncioMagicMock(return_value=response)
node._node_type = "dynamips" # force Dynamips node type
response = await client.get(app.url_path_for("idlepc_proposals", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_200_OK
assert response.json() == ["0x60606f54", "0x33805a22"]
async def test_dynamips_idlepc_proposals_wrong_node_type(
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = await client.get(app.url_path_for("idlepc_proposals", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_400_BAD_REQUEST
async def test_qemu_disk_image_create(
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = MagicMock()
compute.post = AsyncioMagicMock(return_value=response)
node._node_type = "qemu" # force Qemu node type
response = await client.post(
app.url_path_for("create_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2"),
json={"format": "qcow2", "size": 30}
)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_qemu_disk_image_create_wrong_node_type(
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = await client.post(
app.url_path_for("create_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2"),
json={"format": "qcow2", "size": 30}
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
async def test_qemu_disk_image_update(
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = MagicMock()
compute.put = AsyncioMagicMock(return_value=response)
node._node_type = "qemu" # force Qemu node type
response = await client.put(
app.url_path_for("update_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2"),
json={"extend": 10}
)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_qemu_disk_image_update_wrong_node_type(
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = await client.put(
app.url_path_for("update_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2"),
json={"extend": 10}
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
async def test_qemu_disk_image_delete(
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = MagicMock()
compute.delete = AsyncioMagicMock(return_value=response)
node._node_type = "qemu" # force Qemu node type
response = await client.delete(
app.url_path_for("delete_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2")
)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_qemu_disk_image_delete_wrong_node_type(
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = await client.delete(
app.url_path_for("delete_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2")
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
async def test_get_file(app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node) -> None:
response = MagicMock()
response.body = b"world"
response.status = status.HTTP_200_OK
compute.http_query = AsyncioMagicMock(return_value=response)
response = await client.get(app.url_path_for("get_file", project_id=project.id, node_id=node.id, file_path="hello"))
assert response.status_code == status.HTTP_200_OK
assert response.content == b'world'
compute.http_query.assert_called_with(
"GET",
"/projects/{project_id}/files/project-files/vpcs/{node_id}/hello".format(
class TestNodeRoutes:
@pytest.fixture
def node(self, project: Project, compute: Compute) -> Node:
node = Node(project, compute, "test", node_type="vpcs")
project._nodes[node.id] = node
return node
async def test_create_node(self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute) -> None:
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
response = await client.post(app.url_path_for("create_node", project_id=project.id), json={
"name": "test",
"node_type": "vpcs",
"compute_id": "example.com",
"properties": {
"startup_script": "echo test"
}
})
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "test"
assert "name" not in response.json()["properties"]
async def test_list_node(self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute) -> None:
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
await client.post(app.url_path_for("create_node", project_id=project.id), json={
"name": "test",
"node_type": "vpcs",
"compute_id": "example.com",
"properties": {
"startup_script": "echo test"
}
})
response = await client.get(app.url_path_for("get_nodes", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert response.json()[0]["name"] == "test"
# test listing nodes from a closed project
await project.close(ignore_notification=True)
response = await client.get(app.url_path_for("get_nodes", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert response.json()[0]["name"] == "test"
async def test_get_node(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute
) -> None:
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
response = await client.post(app.url_path_for("create_node", project_id=project.id), json={
"name": "test",
"node_type": "vpcs",
"compute_id": "example.com",
"properties": {
"startup_script": "echo test"
}
})
response = await client.get(app.url_path_for("get_node", project_id=project.id, node_id=response.json()["node_id"]))
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "test"
async def test_update_node(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = MagicMock()
response.json = {"console": 2048}
compute.put = AsyncioMagicMock(return_value=response)
response = await client.put(app.url_path_for("update_node", project_id=project.id, node_id=node.id), json={
"name": "test",
"node_type": "vpcs",
"compute_id": "example.com",
"properties": {
"startup_script": "echo test"
}
})
assert response.status_code == 200
assert response.json()["name"] == "test"
assert "name" not in response.json()["properties"]
async def test_start_all_nodes(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute
) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("start_all_nodes", project_id=project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_stop_all_nodes(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute
) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("stop_all_nodes", project_id=project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_suspend_all_nodes(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute
) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("suspend_all_nodes", project_id=project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_reload_all_nodes(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute
) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("reload_all_nodes", project_id=project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_reset_console_all_nodes(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute
) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("reset_console_all_nodes", project_id=project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_start_node(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("start_node", project_id=project.id, node_id=node.id), json={})
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_stop_node(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("stop_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_suspend_node(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("suspend_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_reload_node(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
):
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("reload_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_isolate_node(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
):
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("isolate_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_unisolate_node(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("unisolate_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_duplicate_node(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = MagicMock()
response.json({"console": 2035})
compute.post = AsyncioMagicMock(return_value=response)
response = await client.post(app.url_path_for("duplicate_node", project_id=project.id, node_id=node.id),
json={"x": 10, "y": 5, "z": 0})
assert response.status_code == status.HTTP_201_CREATED
async def test_delete_node(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
compute.post = AsyncioMagicMock()
response = await client.delete(app.url_path_for("delete_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_dynamips_idle_pc(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = MagicMock()
response.json = {"idlepc": "0x60606f54"}
compute.get = AsyncioMagicMock(return_value=response)
node._node_type = "dynamips" # force Dynamips node type
response = await client.get(app.url_path_for("auto_idlepc", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_200_OK
assert response.json()["idlepc"] == "0x60606f54"
async def test_dynamips_idle_pc_wrong_node_type(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = await client.get(app.url_path_for("auto_idlepc", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_400_BAD_REQUEST
async def test_dynamips_idlepc_proposals(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = MagicMock()
response.json = ["0x60606f54", "0x33805a22"]
compute.get = AsyncioMagicMock(return_value=response)
node._node_type = "dynamips" # force Dynamips node type
response = await client.get(app.url_path_for("idlepc_proposals", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_200_OK
assert response.json() == ["0x60606f54", "0x33805a22"]
async def test_dynamips_idlepc_proposals_wrong_node_type(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = await client.get(app.url_path_for("idlepc_proposals", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_400_BAD_REQUEST
async def test_qemu_disk_image_create(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = MagicMock()
compute.post = AsyncioMagicMock(return_value=response)
node._node_type = "qemu" # force Qemu node type
response = await client.post(
app.url_path_for("create_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2"),
json={"format": "qcow2", "size": 30}
)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_qemu_disk_image_create_wrong_node_type(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = await client.post(
app.url_path_for("create_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2"),
json={"format": "qcow2", "size": 30}
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
async def test_qemu_disk_image_update(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = MagicMock()
compute.put = AsyncioMagicMock(return_value=response)
node._node_type = "qemu" # force Qemu node type
response = await client.put(
app.url_path_for("update_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2"),
json={"extend": 10}
)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_qemu_disk_image_update_wrong_node_type(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = await client.put(
app.url_path_for("update_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2"),
json={"extend": 10}
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
async def test_qemu_disk_image_delete(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = MagicMock()
compute.delete = AsyncioMagicMock(return_value=response)
node._node_type = "qemu" # force Qemu node type
response = await client.delete(
app.url_path_for("delete_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2")
)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_qemu_disk_image_delete_wrong_node_type(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = await client.delete(
app.url_path_for("delete_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2")
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
async def test_get_file(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
response = MagicMock()
response.body = b"world"
response.status = status.HTTP_200_OK
compute.http_query = AsyncioMagicMock(return_value=response)
response = await client.get(app.url_path_for("get_file", project_id=project.id, node_id=node.id, file_path="hello"))
assert response.status_code == status.HTTP_200_OK
assert response.content == b'world'
compute.http_query.assert_called_with(
"GET",
"/projects/{project_id}/files/project-files/vpcs/{node_id}/hello".format(
project_id=project.id,
node_id=node.id),
timeout=None,
raw=True)
response = await client.get(app.url_path_for(
"get_file",
project_id=project.id,
node_id=node.id),
timeout=None,
raw=True)
response = await client.get(app.url_path_for(
"get_file",
project_id=project.id,
node_id=node.id,
file_path="../hello"))
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_post_file(app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node) -> None:
compute.http_query = AsyncioMagicMock()
response = await client.post(app.url_path_for(
"post_file",
project_id=project.id,
node_id=node.id,
file_path="hello"), content=b"hello")
assert response.status_code == status.HTTP_201_CREATED
compute.http_query.assert_called_with("POST", "/projects/{project_id}/files/project-files/vpcs/{node_id}/hello".format(project_id=project.id, node_id=node.id), data=b'hello', timeout=None, raw=True)
response = await client.get("/projects/{project_id}/nodes/{node_id}/files/../hello".format(project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_404_NOT_FOUND
# @pytest.mark.asyncio
# async def test_get_and_post_with_nested_paths_normalization(controller_api, project, node, compute):
#
# response = MagicMock()
# response.body = b"world"
# compute.http_query = AsyncioMagicMock(return_value=response)
# response = await controller_api.get("/projects/{project_id}/nodes/{node_id}/files/hello\\nested".format(project_id=project.id, node_id=node.id))
# assert response.status_code == 200
# assert response.content == b'world'
#
# compute.http_query.assert_called_with("GET", "/projects/{project_id}/files/project-files/vpcs/{node_id}/hello/nested".format(project_id=project.id, node_id=node.id), timeout=None, raw=True)
#
# compute.http_query = AsyncioMagicMock()
# response = await controller_api.post("/projects/{project_id}/nodes/{node_id}/files/hello\\nested".format(project_id=project.id, node_id=node.id), body=b"hello", raw=True)
# assert response.status_code == 201
#
# compute.http_query.assert_called_with("POST", "/projects/{project_id}/files/project-files/vpcs/{node_id}/hello/nested".format(project_id=project.id, node_id=node.id), data=b'hello', timeout=None, raw=True)
node_id=node.id,
file_path="../hello"))
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_post_file(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
node: Node
) -> None:
compute.http_query = AsyncioMagicMock()
response = await client.post(app.url_path_for(
"post_file",
project_id=project.id,
node_id=node.id,
file_path="hello"), content=b"hello")
assert response.status_code == status.HTTP_201_CREATED
compute.http_query.assert_called_with("POST", "/projects/{project_id}/files/project-files/vpcs/{node_id}/hello".format(project_id=project.id, node_id=node.id), data=b'hello', timeout=None, raw=True)
response = await client.get("/projects/{project_id}/nodes/{node_id}/files/../hello".format(project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_404_NOT_FOUND
# @pytest.mark.asyncio
# async def test_get_and_post_with_nested_paths_normalization(controller_api, project, node, compute):
#
# response = MagicMock()
# response.body = b"world"
# compute.http_query = AsyncioMagicMock(return_value=response)
# response = await controller_api.get("/projects/{project_id}/nodes/{node_id}/files/hello\\nested".format(project_id=project.id, node_id=node.id))
# assert response.status_code == 200
# assert response.content == b'world'
#
# compute.http_query.assert_called_with("GET", "/projects/{project_id}/files/project-files/vpcs/{node_id}/hello/nested".format(project_id=project.id, node_id=node.id), timeout=None, raw=True)
#
# compute.http_query = AsyncioMagicMock()
# response = await controller_api.post("/projects/{project_id}/nodes/{node_id}/files/hello\\nested".format(project_id=project.id, node_id=node.id), body=b"hello", raw=True)
# assert response.status_code == 201
#
# compute.http_query.assert_called_with("POST", "/projects/{project_id}/files/project-files/vpcs/{node_id}/hello/nested".format(project_id=project.id, node_id=node.id), data=b'hello', timeout=None, raw=True)
tests/api/routes/controller/test_projects.py
+529 -476
View File
File diff suppressed because it is too large. Load diff
tests/api/routes/controller/test_snapshots.py
+45 -43
View File
@@ -30,47 +30,49 @@ from gns3server.controller.snapshot import Snapshot
pytestmark = pytest.mark.asyncio
@pytest_asyncio.fixture
async def project(app: FastAPI, client: AsyncClient, controller: Controller) -> Project:
class TestSnapshotRoutes:
u = str(uuid.uuid4())
params = {"name": "test", "project_id": u}
await client.post(app.url_path_for("create_project"), json=params)
project = controller.get_project(u)
return project
@pytest_asyncio.fixture
async def snapshot(project: Project):
snapshot = await project.snapshot("test")
return snapshot
async def test_list_snapshots(app: FastAPI, client: AsyncClient, project: Project, snapshot: Snapshot) -> None:
assert snapshot.name == "test"
response = await client.get(app.url_path_for("get_snapshots", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 1
async def test_delete_snapshot(app: FastAPI, client: AsyncClient, project: Project, snapshot: Snapshot) -> None:
response = await client.delete(app.url_path_for("delete_snapshot", project_id=project.id, snapshot_id=snapshot.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
assert not os.path.exists(snapshot.path)
async def test_restore_snapshot(app: FastAPI, client: AsyncClient, project: Project, snapshot: Snapshot) -> None:
response = await client.post(app.url_path_for("restore_snapshot", project_id=project.id, snapshot_id=snapshot.id))
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == project.name
async def test_create_snapshot(app: FastAPI, client: AsyncClient, project: Project) -> None:
response = await client.post(app.url_path_for("create_snapshot", project_id=project.id), json={"name": "snap1"})
assert response.status_code == status.HTTP_201_CREATED
assert len(os.listdir(os.path.join(project.path, "snapshots"))) == 1
@pytest_asyncio.fixture
async def project(self, app: FastAPI, client: AsyncClient, controller: Controller) -> Project:
u = str(uuid.uuid4())
params = {"name": "test", "project_id": u}
await client.post(app.url_path_for("create_project"), json=params)
controller_project = controller.get_project(u)
return controller_project
@pytest_asyncio.fixture
async def snapshot(self, project: Project):
controller_snapshot = await project.snapshot("test")
return controller_snapshot
async def test_list_snapshots(self, app: FastAPI, client: AsyncClient, project: Project, snapshot: Snapshot) -> None:
assert snapshot.name == "test"
response = await client.get(app.url_path_for("get_snapshots", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 1
async def test_delete_snapshot(self, app: FastAPI, client: AsyncClient, project: Project, snapshot: Snapshot) -> None:
response = await client.delete(app.url_path_for("delete_snapshot", project_id=project.id, snapshot_id=snapshot.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
assert not os.path.exists(snapshot.path)
async def test_restore_snapshot(self, app: FastAPI, client: AsyncClient, project: Project, snapshot: Snapshot) -> None:
response = await client.post(app.url_path_for("restore_snapshot", project_id=project.id, snapshot_id=snapshot.id))
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == project.name
async def test_create_snapshot(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
response = await client.post(app.url_path_for("create_snapshot", project_id=project.id), json={"name": "snap1"})
assert response.status_code == status.HTTP_201_CREATED
assert len(os.listdir(os.path.join(project.path, "snapshots"))) == 1
tests/api/routes/controller/test_symbols.py
+39 -37
View File
@@ -28,41 +28,43 @@ from gns3server.controller import Controller
pytestmark = pytest.mark.asyncio
async def test_symbols(app: FastAPI, client: AsyncClient) -> None:
class TestSymbolRoutes:
response = await client.get(app.url_path_for("get_symbols"))
assert response.status_code == status.HTTP_200_OK
assert {
'symbol_id': ':/symbols/classic/firewall.svg',
'filename': 'firewall.svg',
'builtin': True,
'theme': 'Classic'
} in response.json()
async def test_get(app: FastAPI, client: AsyncClient, controller: Controller) -> None:
controller.symbols.theme = "Classic"
url = app.url_path_for("get_symbol", symbol_id=urllib.parse.quote(':/symbols/classic/firewall.svg'))
response = await client.get(url)
assert response.status_code == status.HTTP_200_OK
assert response.headers['CONTENT-TYPE'] == 'image/svg+xml'
assert response.headers['CONTENT-LENGTH'] == '9381'
assert '</svg>' in response.text
# Reply with the default symbol
response = await client.get(app.url_path_for("get_symbol", symbol_id="404.png"))
assert response.status_code == status.HTTP_200_OK
async def test_upload(app: FastAPI, client: AsyncClient, symbols_dir: str) -> None:
response = await client.post(app.url_path_for("upload_symbol", symbol_id="test2"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
with open(os.path.join(symbols_dir, "test2")) as f:
assert f.read() == "TEST"
response = await client.get(app.url_path_for("get_symbol", symbol_id="test2"))
assert response.status_code == status.HTTP_200_OK
async def test_symbols(self, app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("get_symbols"))
assert response.status_code == status.HTTP_200_OK
assert {
'symbol_id': ':/symbols/classic/firewall.svg',
'filename': 'firewall.svg',
'builtin': True,
'theme': 'Classic'
} in response.json()
async def test_get(self, app: FastAPI, client: AsyncClient, controller: Controller) -> None:
controller.symbols.theme = "Classic"
url = app.url_path_for("get_symbol", symbol_id=urllib.parse.quote(':/symbols/classic/firewall.svg'))
response = await client.get(url)
assert response.status_code == status.HTTP_200_OK
assert response.headers['CONTENT-TYPE'] == 'image/svg+xml'
assert response.headers['CONTENT-LENGTH'] == '9381'
assert '</svg>' in response.text
# Reply with the default symbol
response = await client.get(app.url_path_for("get_symbol", symbol_id="404.png"))
assert response.status_code == status.HTTP_200_OK
async def test_upload(self, app: FastAPI, client: AsyncClient, symbols_dir: str) -> None:
response = await client.post(app.url_path_for("upload_symbol", symbol_id="test2"), content=b"TEST")
assert response.status_code == status.HTTP_204_NO_CONTENT
with open(os.path.join(symbols_dir, "test2")) as f:
assert f.read() == "TEST"
response = await client.get(app.url_path_for("get_symbol", symbol_id="test2"))
assert response.status_code == status.HTTP_200_OK
tests/api/routes/controller/test_users.py
+15 -29
View File
@@ -254,22 +254,6 @@ class TestUserLogin:
assert "token_type" in response.json()
assert response.json().get("token_type") == "bearer"
async def test_user_can_authenticate_using_json(
self,
app: FastAPI,
unauthorized_client: AsyncClient,
test_user: User,
config: Config
) -> None:
credentials = {
"username": test_user.username,
"password": "user1_password",
}
response = await unauthorized_client.post(app.url_path_for("authenticate"), json=credentials)
assert response.status_code == status.HTTP_200_OK
assert response.json().get("access_token")
@pytest.mark.parametrize(
"username, password, status_code",
(
@@ -299,7 +283,19 @@ class TestUserLogin:
assert response.status_code == status_code
assert "access_token" not in response.json()
async def test_user_can_use_token_as_url_param(
class TestUnauthorizedUser:
async def test_user_cannot_access_own_data_if_not_authenticated(
self, app: FastAPI,
unauthorized_client: AsyncClient,
test_user: User,
) -> None:
response = await unauthorized_client.get(app.url_path_for("get_logged_in_user"))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
async def test_user_can_authenticate_using_json(
self,
app: FastAPI,
unauthorized_client: AsyncClient,
@@ -311,15 +307,14 @@ class TestUserLogin:
"username": test_user.username,
"password": "user1_password",
}
response = await unauthorized_client.post(app.url_path_for("authenticate"), json=credentials)
assert response.status_code == status.HTTP_200_OK
token = response.json().get("access_token")
assert response.json().get("access_token")
token = response.json().get("access_token")
response = await unauthorized_client.get(app.url_path_for("statistics"), params={"token": token})
assert response.status_code == status.HTTP_200_OK
class TestUserMe:
async def test_authenticated_user_can_retrieve_own_data(
@@ -336,15 +331,6 @@ class TestUserMe:
assert user.email == test_user.email
assert user.user_id == test_user.user_id
async def test_user_cannot_access_own_data_if_not_authenticated(
self, app: FastAPI,
unauthorized_client: AsyncClient,
test_user: User,
) -> None:
response = await unauthorized_client.get(app.url_path_for("get_logged_in_user"))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
async def test_authenticated_user_can_update_own_data(
self,
app: FastAPI,
tests/api/routes/controller/test_version.py
+35 -32
View File
@@ -25,38 +25,41 @@ from gns3server.version import __version__
pytestmark = pytest.mark.asyncio
async def test_version_output(app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("get_version"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == {'controller_host': '127.0.0.1', 'local': False, 'version': __version__}
class TestVersionRoutes:
async def test_version_input(app: FastAPI, client: AsyncClient) -> None:
async def test_version_output(self, app: FastAPI, client: AsyncClient) -> None:
params = {'version': __version__}
response = await client.post(app.url_path_for("check_version"), json=params)
assert response.status_code == status.HTTP_200_OK
assert response.json() == {'version': __version__}
async def test_version_invalid_input(app: FastAPI, client: AsyncClient) -> None:
params = {'version': "0.4.2"}
response = await client.post(app.url_path_for("check_version"), json=params)
assert response.status_code == status.HTTP_409_CONFLICT
assert response.json() == {'message': 'Client version 0.4.2 is not the same as server version {}'.format(__version__)}
async def test_version_invalid_input_schema(app: FastAPI, client: AsyncClient) -> None:
params = {'version': "0.4.2", "bla": "blu"}
response = await client.post(app.url_path_for("check_version"), json=params)
assert response.status_code == status.HTTP_409_CONFLICT
async def test_version_invalid_json(app: FastAPI, client: AsyncClient) -> None:
params = "BOUM"
response = await client.post(app.url_path_for("check_version"), json=params)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
response = await client.get(app.url_path_for("get_version"))
assert response.status_code == status.HTTP_200_OK
assert response.json() == {'controller_host': '127.0.0.1', 'local': False, 'version': __version__}
async def test_version_input(self, app: FastAPI, client: AsyncClient) -> None:
params = {'version': __version__}
response = await client.post(app.url_path_for("check_version"), json=params)
assert response.status_code == status.HTTP_200_OK
assert response.json() == {'version': __version__}
async def test_version_invalid_input(self, app: FastAPI, client: AsyncClient) -> None:
params = {'version': "0.4.2"}
response = await client.post(app.url_path_for("check_version"), json=params)
assert response.status_code == status.HTTP_409_CONFLICT
assert response.json() == {'message': 'Client version 0.4.2 is not the same as server version {}'.format(__version__)}
async def test_version_invalid_input_schema(self, app: FastAPI, client: AsyncClient) -> None:
params = {'version': "0.4.2", "bla": "blu"}
response = await client.post(app.url_path_for("check_version"), json=params)
assert response.status_code == status.HTTP_409_CONFLICT
async def test_version_invalid_json(self, app: FastAPI, client: AsyncClient) -> None:
params = "BOUM"
response = await client.post(app.url_path_for("check_version"), json=params)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY