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

Merge branch '3.1' into base-configs-3.0

Browse Source
This commit is contained in:
Jeremy Grossmann authored and GitHub committed 2026-08-08 22:11:39 +02:00
2 parents 12347b24aa 1d6218a3f1
commit 909ccf8fcd
177 files changed
+16013 -1745
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/api/routes/compute/test_ethernet_switch_nodes.py
+186 -224
View File
@@ -21,34 +21,48 @@ import pytest_asyncio
from fastapi import FastAPI, status
from httpx import AsyncClient
from tests.utils import asyncio_patch, AsyncioMagicMock
from unittest.mock import call
from unittest.mock import call, MagicMock
from gns3server.compute.project import Project
# The builtin Ethernet switch talks to uBridge (brctl/bridge modules) instead of
# the Dynamips hypervisor. These are the seams we stub so the routes can be
# exercised without launching a real uBridge / creating kernel interfaces.
_NODE = "gns3server.compute.builtin.nodes.ethernet_switch.EthernetSwitch"
pytestmark = pytest.mark.asyncio
class TestEthernetSwitchNodesRoutes:
@pytest_asyncio.fixture(autouse=True)
async def stub_ubridge(self):
"""Keep uBridge from really starting and capture every command."""
with asyncio_patch(f"{_NODE}._start_ubridge"), asyncio_patch(f"{_NODE}._stop_ubridge"), \
asyncio_patch(f"{_NODE}._ubridge_send"):
yield
@pytest_asyncio.fixture
async def ethernet_switch(self, app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> dict:
params = {"name": "Ethernet Switch"}
with asyncio_patch("gns3server.compute.dynamips.nodes.ethernet_switch.EthernetSwitch.create") as mock:
response = await compute_client.post(
app.url_path_for("compute:create_ethernet_switch", project_id=compute_project.id),
json=params
)
assert mock.called
assert response.status_code == status.HTTP_201_CREATED
response = await compute_client.post(
app.url_path_for("compute:create_ethernet_switch", project_id=compute_project.id),
json=params
)
assert response.status_code == status.HTTP_201_CREATED
json_response = response.json()
node = compute_project.get_node(json_response["node_id"])
node._hypervisor = AsyncioMagicMock()
node._hypervisor.send = AsyncioMagicMock()
node._hypervisor.version = "0.2.16"
# Pretend uBridge is up so the is_running() guards in remove/close pass.
node._ubridge_hypervisor = MagicMock()
node._ubridge_hypervisor.is_running.return_value = True
node._ubridge_send.reset_mock()
return json_response
@staticmethod
def _udp_params() -> dict:
return {"type": "nio_udp", "lport": 4242, "rport": 4343, "rhost": "127.0.0.1"}
async def test_ethernet_switch_create(
self, app: FastAPI,
@@ -57,16 +71,24 @@ class TestEthernetSwitchNodesRoutes:
) -> None:
params = {"name": "Ethernet Switch 1"}
with asyncio_patch("gns3server.compute.dynamips.nodes.ethernet_switch.EthernetSwitch.create") as mock:
response = await compute_client.post(
app.url_path_for("compute:create_ethernet_switch", project_id=compute_project.id),
json=params
)
assert mock.called
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "Ethernet Switch 1"
assert response.json()["project_id"] == compute_project.id
response = await compute_client.post(
app.url_path_for("compute:create_ethernet_switch", project_id=compute_project.id),
json=params
)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["name"] == "Ethernet Switch 1"
assert response.json()["project_id"] == compute_project.id
assert response.json()["status"] == "started"
# creation stands up the kernel bridge with VLAN filtering
node = compute_project.get_node(response.json()["node_id"])
br = node._bridge_name
node._ubridge_send.assert_has_calls([
call(f'brctl delete "{br}"'),
call(f'brctl create "{br}"'),
call(f'link set "{br}" up'),
call(f'brctl vlanfiltering "{br}" on'),
])
async def test_ethernet_switch_get(
self, app: FastAPI,
@@ -87,7 +109,6 @@ class TestEthernetSwitchNodesRoutes:
assert response.json()["project_id"] == compute_project.id
assert response.json()["status"] == "started"
async def test_ethernet_switch_duplicate(
self,
app: FastAPI,
@@ -98,15 +119,11 @@ class TestEthernetSwitchNodesRoutes:
# create destination switch first
params = {"name": "Ethernet Switch 2"}
with asyncio_patch("gns3server.compute.dynamips.nodes.ethernet_switch.EthernetSwitch.create") as mock:
response = await compute_client.post(
app.url_path_for(
"compute:create_ethernet_switch",
project_id=compute_project.id),
json=params
)
assert mock.called
assert response.status_code == status.HTTP_201_CREATED
response = await compute_client.post(
app.url_path_for("compute:create_ethernet_switch", project_id=compute_project.id),
json=params
)
assert response.status_code == status.HTTP_201_CREATED
params = {"destination_node_id": response.json()["node_id"]}
response = await compute_client.post(
@@ -117,7 +134,6 @@ class TestEthernetSwitchNodesRoutes:
)
assert response.status_code == status.HTTP_201_CREATED
async def test_ethernet_switch_update(
self,
app: FastAPI,
@@ -126,10 +142,7 @@ class TestEthernetSwitchNodesRoutes:
ethernet_switch: dict
) -> None:
params = {
"name": "test",
"console_type": "telnet"
}
params = {"name": "test", "console_type": "none"}
response = await compute_client.put(
app.url_path_for(
@@ -141,11 +154,12 @@ class TestEthernetSwitchNodesRoutes:
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "test"
# renaming a builtin switch does not touch uBridge (the kernel bridge is
# name-independent); nothing should have been sent.
node = compute_project.get_node(ethernet_switch["node_id"])
node._hypervisor.send.assert_called_with("ethsw rename \"Ethernet Switch\" \"test\"")
node._ubridge_send.assert_not_called()
async def test_ethernet_switch_update_ports(
async def test_ethernet_switch_update_ports_qinq_proto(
self,
app: FastAPI,
compute_client: AsyncClient,
@@ -153,33 +167,11 @@ class TestEthernetSwitchNodesRoutes:
ethernet_switch: dict
) -> None:
# a QinQ port with the 802.1ad ethertype must switch the bridge protocol
port_params = {
"ports_mapping": [
{
"name": "Ethernet0",
"port_number": 0,
"type": "qinq",
"vlan": 1
},
{
"name": "Ethernet1",
"port_number": 1,
"type": "qinq",
"vlan": 2,
"ethertype": "0x88A8"
},
{
"name": "Ethernet2",
"port_number": 2,
"type": "dot1q",
"vlan": 3,
},
{
"name": "Ethernet3",
"port_number": 3,
"type": "access",
"vlan": 4,
}
{"name": "Ethernet0", "port_number": 0, "type": "qinq", "vlan": 2, "ethertype": "0x88A8"},
{"name": "Ethernet1", "port_number": 1, "type": "access", "vlan": 4},
],
}
@@ -192,90 +184,20 @@ class TestEthernetSwitchNodesRoutes:
)
assert response.status_code == status.HTTP_200_OK
nio_params = {
"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"
}
for port_mapping in port_params["ports_mapping"]:
port_number = port_mapping["port_number"]
vlan = port_mapping["vlan"]
port_type = port_mapping["type"]
ethertype = port_mapping.get("ethertype", "")
url = app.url_path_for(
"compute:create_ethernet_switch_nio",
project_id=ethernet_switch["project_id"],
node_id=ethernet_switch["node_id"],
adapter_number="0",
port_number=f"{port_number}"
)
await compute_client.post(url, json=nio_params)
node = compute_project.get_node(ethernet_switch["node_id"])
nio = node.get_nio(port_number)
calls = [
call.send(f'nio create_udp {nio.name} 4242 127.0.0.1 4343'),
call.send(f'ethsw add_nio "Ethernet Switch" {nio.name}'),
call.send(f'ethsw set_{port_type}_port "Ethernet Switch" {nio.name} {vlan} {ethertype}'.strip())
]
node._hypervisor.send.assert_has_calls(calls)
node._hypervisor.send.reset_mock()
node = compute_project.get_node(ethernet_switch["node_id"])
node._ubridge_send.assert_any_call(f'brctl setvlanproto "{node._bridge_name}" 0x88a8')
@pytest.mark.parametrize(
"ports_settings",
(
(
{
"name": "Ethernet0",
"port_number": 0,
"type": "dot42q", # invalid port type
"vlan": 1,
}
),
(
{
"name": "Ethernet0",
"port_number": 0,
"type": "access", # missing vlan field
}
),
(
{
"name": "Ethernet0",
"port_number": 0,
"type": "dot1q",
"vlan": 1,
"ethertype": "0x88A8" # EtherType is only for QinQ
}
),
(
{
"name": "Ethernet0",
"port_number": 0,
"type": "qinq",
"vlan": 1,
"ethertype": "0x4242" # not a valid EtherType
}
),
(
{
"name": "Ethernet0",
"port_number": 0,
"type": "access",
"vlan": 0, # minimum vlan number is 1
}
),
(
{
"name": "Ethernet0",
"port_number": 0,
"type": "access",
"vlan": 4242, # maximum vlan number is 4094
}
),
{"name": "Ethernet0", "port_number": 0, "type": "dot42q", "vlan": 1}, # bad type
{"name": "Ethernet0", "port_number": 0, "type": "access"}, # missing vlan
{"name": "Ethernet0", "port_number": 0, "type": "dot1q", "vlan": 1,
"ethertype": "0x88A8"}, # ethertype only for qinq
{"name": "Ethernet0", "port_number": 0, "type": "qinq", "vlan": 1,
"ethertype": "0x4242"}, # bad ethertype
{"name": "Ethernet0", "port_number": 0, "type": "access", "vlan": 0}, # vlan < 1
{"name": "Ethernet0", "port_number": 0, "type": "access", "vlan": 4242}, # vlan > 4094
)
)
async def test_ethernet_switch_update_ports_invalid(
@@ -286,20 +208,15 @@ class TestEthernetSwitchNodesRoutes:
ports_settings: dict,
) -> None:
port_params = {
"ports_mapping": [ports_settings]
}
response = await compute_client.put(
app.url_path_for(
"compute:update_ethernet_switch",
project_id=ethernet_switch["project_id"],
node_id=ethernet_switch["node_id"]),
json=port_params
json={"ports_mapping": [ports_settings]}
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT
async def test_ethernet_switch_delete(
self, app: FastAPI,
compute_client: AsyncClient,
@@ -315,12 +232,7 @@ class TestEthernetSwitchNodesRoutes:
)
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_ethernet_switch_start(
self, app: FastAPI,
compute_client: AsyncClient,
ethernet_switch: dict
) -> None:
async def test_ethernet_switch_start(self, app: FastAPI, compute_client: AsyncClient, ethernet_switch: dict) -> None:
response = await compute_client.post(
app.url_path_for(
@@ -331,12 +243,7 @@ class TestEthernetSwitchNodesRoutes:
)
assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
async def test_ethernet_switch_stop(
self, app: FastAPI,
compute_client: AsyncClient,
ethernet_switch: dict
) -> None:
async def test_ethernet_switch_stop(self, app: FastAPI, compute_client: AsyncClient, ethernet_switch: dict) -> None:
response = await compute_client.post(
app.url_path_for(
@@ -347,12 +254,7 @@ class TestEthernetSwitchNodesRoutes:
)
assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
async def test_ethernet_switch_suspend(
self, app: FastAPI,
compute_client: AsyncClient,
ethernet_switch: dict
) -> None:
async def test_ethernet_switch_suspend(self, app: FastAPI, compute_client: AsyncClient, ethernet_switch: dict) -> None:
response = await compute_client.post(
app.url_path_for(
@@ -363,12 +265,7 @@ class TestEthernetSwitchNodesRoutes:
)
assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
async def test_ethernet_switch_reload(
self, app: FastAPI,
compute_client: AsyncClient,
ethernet_switch: dict
) -> None:
async def test_ethernet_switch_reload(self, app: FastAPI, compute_client: AsyncClient, ethernet_switch: dict) -> None:
response = await compute_client.post(
app.url_path_for(
@@ -379,8 +276,7 @@ class TestEthernetSwitchNodesRoutes:
)
assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
async def test_ethernet_switch_create_udp(
async def test_ethernet_switch_create_udp_access(
self,
app: FastAPI,
compute_client: AsyncClient,
@@ -388,13 +284,6 @@ class TestEthernetSwitchNodesRoutes:
ethernet_switch: dict
) -> None:
params = {
"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"
}
url = app.url_path_for(
"compute:create_ethernet_switch_nio",
project_id=ethernet_switch["project_id"],
@@ -402,19 +291,66 @@ class TestEthernetSwitchNodesRoutes:
adapter_number="0",
port_number="0"
)
response = await compute_client.post(url, json=params)
response = await compute_client.post(url, json=self._udp_params())
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["type"] == "nio_udp"
node = compute_project.get_node(ethernet_switch["node_id"])
nio = node.get_nio(0)
calls = [
call.send(f'nio create_udp {nio.name} 4242 127.0.0.1 4343'),
call.send(f'ethsw add_nio "Ethernet Switch" {nio.name}'),
call.send(f'ethsw set_access_port "Ethernet Switch" {nio.name} 1')
]
node._hypervisor.send.assert_has_calls(calls)
br = node._bridge_name
tap = f"{br}-0"
relay = f"{node.id}-0"
# access VLAN 1 (default): drop default PVID 1, re-add 1 as PVID/untagged
node._ubridge_send.assert_has_calls([
call(f"bridge create {relay}"),
call(f'bridge add_nio_tap {relay} "{tap}"'),
call(f'brctl addif "{br}" "{tap}"'),
call(f'brctl vlan_del "{br}" "{tap}" 1'),
call(f'brctl vlan_add "{br}" "{tap}" 1 pvid untagged'),
call(f"bridge add_nio_udp {relay} {nio.lport} {nio.rhost} {nio.rport}"),
call(f"bridge reset_packet_filters {relay}"),
call(f"bridge start {relay}"),
])
async def test_ethernet_switch_create_udp_dot1q(
self,
app: FastAPI,
compute_client: AsyncClient,
compute_project: Project,
ethernet_switch: dict
) -> None:
# make port 0 a dot1q trunk with native VLAN 10
await compute_client.put(
app.url_path_for(
"compute:update_ethernet_switch",
project_id=ethernet_switch["project_id"],
node_id=ethernet_switch["node_id"]),
json={"ports_mapping": [
{"name": "Ethernet0", "port_number": 0, "type": "dot1q", "vlan": 10},
]}
)
node = compute_project.get_node(ethernet_switch["node_id"])
node._ubridge_send.reset_mock()
url = app.url_path_for(
"compute:create_ethernet_switch_nio",
project_id=ethernet_switch["project_id"],
node_id=ethernet_switch["node_id"],
adapter_number="0",
port_number="0"
)
response = await compute_client.post(url, json=self._udp_params())
assert response.status_code == status.HTTP_201_CREATED
br = node._bridge_name
tap = f"{br}-0"
# trunk: drop default 1, admit all VIDs tagged, mark native 10 PVID/untagged
node._ubridge_send.assert_has_calls([
call(f'brctl vlan_del "{br}" "{tap}" 1'),
call(f'brctl vlan_add "{br}" "{tap}" 1 vid 4094'),
call(f'brctl vlan_add "{br}" "{tap}" 10 pvid untagged'),
])
async def test_ethernet_switch_delete_nio(
self,
@@ -424,13 +360,6 @@ class TestEthernetSwitchNodesRoutes:
ethernet_switch: dict
) -> None:
params = {
"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"
}
url = app.url_path_for(
"compute:create_ethernet_switch_nio",
project_id=ethernet_switch["project_id"],
@@ -438,11 +367,10 @@ class TestEthernetSwitchNodesRoutes:
adapter_number="0",
port_number="0"
)
await compute_client.post(url, json=params)
await compute_client.post(url, json=self._udp_params())
node = compute_project.get_node(ethernet_switch["node_id"])
node._hypervisor.send.reset_mock()
nio = node.get_nio(0)
node._ubridge_send.reset_mock()
url = app.url_path_for(
"compute:delete_ethernet_switch_nio",
@@ -454,52 +382,86 @@ class TestEthernetSwitchNodesRoutes:
response = await compute_client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
calls = [
call(f'ethsw remove_nio "Ethernet Switch" {nio.name}'),
call(f'nio delete {nio.name}')
]
node._hypervisor.send.assert_has_calls(calls)
br = node._bridge_name
tap = f"{br}-0"
relay = f"{node.id}-0"
node._ubridge_send.assert_has_calls([
call(f'brctl delif "{br}" "{tap}"'),
call(f"bridge delete {relay}"),
])
async def test_ethernet_switch_start_capture(
self,
app: FastAPI,
compute_client: AsyncClient,
compute_project: Project,
ethernet_switch: dict
) -> None:
params = {
"capture_file_name": "test.pcap",
"data_link_type": "DLT_EN10MB"
}
# capture needs a wired port
url = app.url_path_for(
"compute:create_ethernet_switch_nio",
project_id=ethernet_switch["project_id"],
node_id=ethernet_switch["node_id"],
adapter_number="0",
port_number="0"
)
await compute_client.post(url, json=self._udp_params())
node = compute_project.get_node(ethernet_switch["node_id"])
node._ubridge_send.reset_mock()
params = {"capture_file_name": "test.pcap", "data_link_type": "DLT_EN10MB"}
url = app.url_path_for("compute:start_ethernet_switch_capture",
project_id=ethernet_switch["project_id"],
node_id=ethernet_switch["node_id"],
adapter_number="0",
port_number="0")
with asyncio_patch("gns3server.compute.dynamips.nodes.ethernet_switch.EthernetSwitch.start_capture") as mock:
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"]
response = await compute_client.post(url, json=params)
assert response.status_code == status.HTTP_200_OK
assert "test.pcap" in response.json()["pcap_file_path"]
relay = f"{node.id}-0"
node._ubridge_send.assert_any_call(f'bridge start_capture {relay} "{node.get_nio(0).pcap_output_file}"')
async def test_ethernet_switch_stop_capture(
self,
app: FastAPI,
compute_client: AsyncClient,
compute_project: Project,
ethernet_switch: dict
) -> None:
url = app.url_path_for("compute:stop_ethernet_switch_capture",
project_id=ethernet_switch["project_id"],
node_id=ethernet_switch["node_id"],
adapter_number="0",
port_number="0")
# start a capture first
await compute_client.post(
app.url_path_for(
"compute:create_ethernet_switch_nio",
project_id=ethernet_switch["project_id"],
node_id=ethernet_switch["node_id"],
adapter_number="0",
port_number="0"
),
json=self._udp_params()
)
await compute_client.post(
app.url_path_for("compute:start_ethernet_switch_capture",
project_id=ethernet_switch["project_id"],
node_id=ethernet_switch["node_id"],
adapter_number="0",
port_number="0"),
json={"capture_file_name": "test.pcap", "data_link_type": "DLT_EN10MB"}
)
with asyncio_patch("gns3server.compute.dynamips.nodes.ethernet_switch.EthernetSwitch.stop_capture") as mock:
response = await compute_client.post(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert mock.called
node = compute_project.get_node(ethernet_switch["node_id"])
node._ubridge_send.reset_mock()
relay = f"{node.id}-0"
response = await compute_client.post(
app.url_path_for("compute:stop_ethernet_switch_capture",
project_id=ethernet_switch["project_id"],
node_id=ethernet_switch["node_id"],
adapter_number="0",
port_number="0")
)
assert response.status_code == status.HTTP_204_NO_CONTENT
node._ubridge_send.assert_any_call(f"bridge stop_capture {relay}")
tests/api/routes/compute/test_images.py
+49
View File
@@ -0,0 +1,49 @@
#
# Copyright (C) 2026 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
from fastapi import FastAPI, status
from httpx import AsyncClient
from tests.utils import asyncio_patch
pytestmark = pytest.mark.asyncio
class TestImagesRoutes:
async def test_pull_docker_image(self, app: FastAPI, compute_client: AsyncClient) -> None:
with asyncio_patch("gns3server.compute.docker.Docker.pull_image") as mock:
response = await compute_client.post(
app.url_path_for("compute:pull_docker_image"),
json={"image": "nginx:latest"}
)
mock.assert_called_once_with("nginx:latest", force=True)
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.parametrize("image", ["", " ", "nginx latest"])
async def test_pull_docker_image_rejects_invalid_name(
self, app: FastAPI, compute_client: AsyncClient, image: str
) -> None:
with asyncio_patch("gns3server.compute.docker.Docker.pull_image") as mock:
response = await compute_client.post(
app.url_path_for("compute:pull_docker_image"),
json={"image": image}
)
mock.assert_not_called()
assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT
tests/api/routes/controller/test_computes.py
+17 -2
View File
@@ -17,6 +17,7 @@
import uuid
import pytest
from unittest.mock import MagicMock, patch
from fastapi import FastAPI, status
from httpx import AsyncClient
@@ -25,8 +26,7 @@ from gns3server.schemas.controller.computes import Compute
pytestmark = pytest.mark.asyncio
import unittest
from tests.utils import asyncio_patch
from tests.utils import asyncio_patch, AsyncioMagicMock
class TestComputeRoutes:
@@ -128,6 +128,21 @@ class TestComputeFeatures:
mock.assert_called_with("GET", "docker", "images")
assert response.json() == [{"image": "docker1"}, {"image": "docker2"}]
async def test_compute_pull_docker_image(
self, app: FastAPI, client: AsyncClient, test_compute: Compute
) -> None:
compute = MagicMock()
compute.forward = AsyncioMagicMock(return_value={})
with patch("gns3server.api.routes.controller.computes.Controller.instance") as controller:
controller.return_value.get_compute.return_value = compute
response = await client.post(
app.url_path_for("docker_pull_image", compute_id=test_compute.compute_id),
json={"image": "nginx:latest"}
)
compute.forward.assert_called_with("POST", "docker", "images/pull", data={"image": "nginx:latest"})
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_compute_list_virtualbox_vms(self, app: FastAPI, client: AsyncClient) -> None:
params = {
tests/api/routes/controller/test_markers.py
+267
View File
@@ -0,0 +1,267 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2025 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/>.
"""
HTTP-route tests for the traffic-insight marker endpoints: per-link markers,
project-level definitions, and the project-wide aggregation view.
"""
import pytest
from fastapi import FastAPI, status
from httpx import AsyncClient
from tests.utils import asyncio_patch
from gns3server.controller.project import Project
from gns3server.controller.udp_link import UDPLink
pytestmark = pytest.mark.asyncio
def _inherited(link, name="arp"):
"""Inject an inherited marker so the controller's inheritance guard can fire."""
link._markers[f"global-{name}"] = {
"bpf": name, "tag": None, "enabled": True, "color": None,
"highlight_duration": None, "capture_node_id": "node-id",
"inherited_from": name,
}
class TestMarkerRoutes:
# -----------------------------------------------------------------------
# Per-link markers
# -----------------------------------------------------------------------
async def test_create_marker(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = UDPLink(project)
project._links = {link.id: link}
with asyncio_patch("gns3server.controller.udp_link.UDPLink.start_marker") as mock:
response = await client.post(
app.url_path_for("create_marker", project_id=project.id, link_id=link.id),
json={"bpf": "icmp", "tag": 3, "color": "#ff5722", "highlight_duration": 800},
)
assert response.status_code == status.HTTP_201_CREATED
mock.assert_called_once()
_, kwargs = mock.call_args
assert kwargs["bpf"] == "icmp"
assert kwargs["tag"] == 3
assert kwargs["color"] == "#ff5722"
assert kwargs["highlight_duration"] == 800
assert kwargs["name"].startswith("marker-")
async def test_create_marker_with_explicit_name(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = UDPLink(project)
project._links = {link.id: link}
with asyncio_patch("gns3server.controller.udp_link.UDPLink.start_marker") as mock:
response = await client.post(
app.url_path_for("create_marker", project_id=project.id, link_id=link.id),
json={"name": "web", "bpf": "tcp port 80"},
)
assert response.status_code == status.HTTP_201_CREATED
_, kwargs = mock.call_args
assert kwargs["name"] == "web"
async def test_create_marker_global_prefix_rejected(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = UDPLink(project)
project._links = {link.id: link}
with asyncio_patch("gns3server.controller.udp_link.UDPLink.start_marker") as mock:
response = await client.post(
app.url_path_for("create_marker", project_id=project.id, link_id=link.id),
json={"name": "global-x", "bpf": "icmp"},
)
assert response.status_code == status.HTTP_409_CONFLICT
assert not mock.called # rejected before reaching the controller
async def test_create_marker_bad_format_rejected(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = UDPLink(project)
project._links = {link.id: link}
response = await client.post(
app.url_path_for("create_marker", project_id=project.id, link_id=link.id),
json={"name": "bad name!", "bpf": "icmp"},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
async def test_create_marker_name_too_long_rejected(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = UDPLink(project)
project._links = {link.id: link}
response = await client.post(
app.url_path_for("create_marker", project_id=project.id, link_id=link.id),
json={"name": "x" * 33, "bpf": "icmp"}, # max_length is 32
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
async def test_get_markers(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = UDPLink(project)
link._markers["web"] = {"bpf": "tcp port 80", "tag": None, "enabled": True,
"color": None, "highlight_duration": 800, "capture_node_id": "n1"}
project._links = {link.id: link}
response = await client.get(
app.url_path_for("get_markers", project_id=project.id, link_id=link.id)
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["web"]["highlight_duration"] == 800
async def test_update_marker(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = UDPLink(project)
project._links = {link.id: link}
with asyncio_patch("gns3server.controller.udp_link.UDPLink.update_marker") as mock:
response = await client.put(
app.url_path_for("update_marker", project_id=project.id, link_id=link.id, marker_name="web"),
json={"bpf": "udp port 53", "highlight_duration": 1500},
)
assert response.status_code == status.HTTP_200_OK
_, kwargs = mock.call_args
assert kwargs["bpf"] == "udp port 53"
assert kwargs["highlight_duration"] == 1500
async def test_update_inherited_marker_rejected(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = UDPLink(project)
_inherited(link, "arp")
project._links = {link.id: link}
response = await client.put(
app.url_path_for("update_marker", project_id=project.id, link_id=link.id, marker_name="global-arp"),
json={"name": "global-arp", "bpf": "arp"},
)
assert response.status_code == status.HTTP_409_CONFLICT
assert "inherited" in response.json()["message"]
async def test_delete_marker(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = UDPLink(project)
project._links = {link.id: link}
with asyncio_patch("gns3server.controller.udp_link.UDPLink.stop_marker") as mock:
response = await client.delete(
app.url_path_for("delete_marker", project_id=project.id, link_id=link.id, marker_name="web")
)
assert response.status_code == status.HTTP_204_NO_CONTENT
mock.assert_called_once_with("web")
async def test_delete_inherited_marker_rejected(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = UDPLink(project)
_inherited(link, "arp")
project._links = {link.id: link}
response = await client.delete(
app.url_path_for("delete_marker", project_id=project.id, link_id=link.id, marker_name="global-arp")
)
assert response.status_code == status.HTTP_409_CONFLICT
# -----------------------------------------------------------------------
# Project-level marker definitions
# -----------------------------------------------------------------------
async def test_create_marker_definition(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
with asyncio_patch("gns3server.controller.project.Project.create_marker_definition") as mock:
response = await client.post(
app.url_path_for("create_marker_definition", project_id=project.id),
json={"name": "arp", "bpf": "arp", "highlight_duration": 1200},
)
assert response.status_code == status.HTTP_201_CREATED
_, kwargs = mock.call_args
assert kwargs["name"] == "arp"
assert kwargs["bpf"] == "arp"
assert kwargs["highlight_duration"] == 1200
async def test_create_marker_definition_global_prefix_rejected(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
with asyncio_patch("gns3server.controller.project.Project.create_marker_definition") as mock:
response = await client.post(
app.url_path_for("create_marker_definition", project_id=project.id),
json={"name": "global-x", "bpf": "arp"},
)
assert response.status_code == status.HTTP_409_CONFLICT
assert not mock.called
async def test_update_marker_definition(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
with asyncio_patch("gns3server.controller.project.Project.update_marker_definition") as mock:
response = await client.put(
app.url_path_for("update_marker_definition", project_id=project.id, def_name="arp"),
json={"bpf": "arp or rarp", "highlight_duration": 900},
)
assert response.status_code == status.HTTP_200_OK
_, kwargs = mock.call_args
assert kwargs["bpf"] == "arp or rarp"
assert kwargs["highlight_duration"] == 900
async def test_delete_marker_definition(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
with asyncio_patch("gns3server.controller.project.Project.delete_marker_definition") as mock:
response = await client.delete(
app.url_path_for("delete_marker_definition", project_id=project.id, def_name="arp")
)
assert response.status_code == status.HTTP_204_NO_CONTENT
mock.assert_called_once_with("arp")
async def test_get_marker_definitions(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
project._marker_definitions = {
"arp": {"bpf": "arp", "tag": 5, "color": None, "highlight_duration": 1200},
}
project._links = {}
response = await client.get(
app.url_path_for("get_marker_definitions", project_id=project.id)
)
assert response.status_code == status.HTTP_200_OK
body = response.json()
assert body["arp"]["bpf"] == "arp"
assert body["arp"]["highlight_duration"] == 1200
assert body["arp"]["link_ids"] == []
# -----------------------------------------------------------------------
# Aggregation
# -----------------------------------------------------------------------
async def test_get_project_markers(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = UDPLink(project)
link._markers["icmp"] = {"bpf": "icmp", "tag": 1, "enabled": True, "color": "#ff5722",
"highlight_duration": 800, "capture_node_id": "node-1"}
project._links = {link.id: link}
response = await client.get(
app.url_path_for("get_project_markers", project_id=project.id)
)
assert response.status_code == status.HTTP_200_OK
body = response.json()
key = f"{link.id}/icmp"
assert key in body
assert body[key]["highlight_duration"] == 800
assert body[key]["link_id"] == link.id
assert body[key]["node_id"] == "node-1"
tests/api/routes/controller/test_nodes.py
+27 -11
View File
@@ -553,24 +553,33 @@ class TestNodeRoutes:
compute: Compute,
node: Node
) -> None:
response = MagicMock()
response.body = b"world"
response.status = status.HTTP_200_OK
compute.http_query = AsyncioMagicMock(return_value=response)
# Mock the streaming response
async def mock_iter_chunked(chunk_size):
yield b"world"
mock_stream = AsyncioMagicMock()
mock_stream.iter_chunked = mock_iter_chunked
mock_stream.close = MagicMock()
mock_response = AsyncioMagicMock()
mock_response.status = status.HTTP_200_OK
mock_response.content = mock_stream
compute.http_query = AsyncioMagicMock(return_value=mock_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)
stream=True)
response = await client.get(app.url_path_for(
"get_file",
project_id=project.id,
@@ -595,8 +604,15 @@ class TestNodeRoutes:
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)
# Verify http_query was called with stream parameter
compute.http_query.assert_called_once()
call_args = compute.http_query.call_args
assert call_args[0][0] == "POST"
assert call_args[0][1] == "/projects/{project_id}/files/project-files/vpcs/{node_id}/hello".format(project_id=project.id, node_id=node.id)
assert call_args[1]["timeout"] is None
# data should be an async generator from request.stream()
assert hasattr(call_args[1]["data"], "__aiter__")
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
tests/api/routes/controller/test_users.py
+175
View File
@@ -255,6 +255,10 @@ class TestUserLogin:
assert "token_type" in response.json()
assert response.json().get("token_type") == "bearer"
# check that refresh token is returned
assert "refresh_token" in response.json()
assert response.json().get("refresh_token") is not None
@pytest.mark.parametrize(
"username, password, status_code",
(
@@ -311,6 +315,7 @@ class TestUnauthorizedUser:
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")
assert response.json().get("refresh_token") is not None
token = response.json().get("access_token")
response = await unauthorized_client.get(app.url_path_for("statistics"), params={"token": token})
@@ -485,6 +490,176 @@ class TestLogout:
assert response.status_code == status.HTTP_401_UNAUTHORIZED
class TestRefreshToken:
async def test_login_returns_refresh_token(
self,
app: FastAPI,
unauthorized_client: AsyncClient,
test_user: User,
) -> None:
credentials = {"username": test_user.username, "password": "user1_password"}
response = await unauthorized_client.post(app.url_path_for("login"),
data=credentials,
headers={"content-type": "application/x-www-form-urlencoded"})
assert response.status_code == status.HTTP_200_OK
assert "refresh_token" in response.json()
assert response.json().get("refresh_token") is not None
async def test_authenticate_returns_refresh_token(
self,
app: FastAPI,
unauthorized_client: AsyncClient,
test_user: User,
) -> 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 "refresh_token" in response.json()
assert response.json().get("refresh_token") is not None
async def test_refresh_endpoint_returns_new_tokens(
self,
app: FastAPI,
unauthorized_client: AsyncClient,
test_user: User,
) -> None:
# authenticate to get a refresh token
credentials = {"username": test_user.username, "password": "user1_password"}
auth_response = await unauthorized_client.post(app.url_path_for("authenticate"), json=credentials)
refresh_token = auth_response.json()["refresh_token"]
# use the refresh token at /refresh
response = await unauthorized_client.post(
app.url_path_for("refresh_access_token"),
json={"refresh_token": refresh_token},
)
assert response.status_code == status.HTTP_200_OK
assert "access_token" in response.json()
assert response.json().get("token_type") == "bearer"
assert "refresh_token" in response.json()
assert response.json().get("refresh_token") is not None
async def test_new_access_token_from_refresh_works_on_protected_route(
self,
app: FastAPI,
unauthorized_client: AsyncClient,
test_user: User,
) -> None:
credentials = {"username": test_user.username, "password": "user1_password"}
auth_response = await unauthorized_client.post(app.url_path_for("authenticate"), json=credentials)
refresh_token = auth_response.json()["refresh_token"]
# refresh to get a new access token
refresh_response = await unauthorized_client.post(
app.url_path_for("refresh_access_token"),
json={"refresh_token": refresh_token},
)
new_access_token = refresh_response.json()["access_token"]
# the new access token must work on a protected route
response = await unauthorized_client.get(
app.url_path_for("get_logged_in_user"),
headers={"Authorization": f"Bearer {new_access_token}"},
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["username"] == test_user.username
async def test_refresh_with_stale_token_after_logout(
self,
app: FastAPI,
unauthorized_client: AsyncClient,
test_user: User,
) -> None:
# authenticate and get a refresh token
credentials = {"username": test_user.username, "password": "user1_password"}
auth_response = await unauthorized_client.post(app.url_path_for("authenticate"), json=credentials)
refresh_token = auth_response.json()["refresh_token"]
# logout — bumps token_version, invalidating the refresh token
access_token = auth_response.json()["access_token"]
await unauthorized_client.post(
app.url_path_for("logout"),
headers={"Authorization": f"Bearer {access_token}"}
)
# /refresh must now reject the stale refresh token
response = await unauthorized_client.post(
app.url_path_for("refresh_access_token"),
json={"refresh_token": refresh_token},
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
async def test_refresh_rejects_access_token(
self,
app: FastAPI,
unauthorized_client: AsyncClient,
test_user: User,
) -> None:
# an access token presented at /refresh must be rejected
credentials = {"username": test_user.username, "password": "user1_password"}
auth_response = await unauthorized_client.post(app.url_path_for("authenticate"), json=credentials)
access_token = auth_response.json()["access_token"]
response = await unauthorized_client.post(
app.url_path_for("refresh_access_token"),
json={"refresh_token": access_token},
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
async def test_refresh_rejects_expired_token(
self,
app: FastAPI,
unauthorized_client: AsyncClient,
test_user: User,
) -> None:
# a refresh token with an already-expired timestamp
expired_refresh = auth_service.create_refresh_token(test_user.username, expires_in=-1)
response = await unauthorized_client.post(
app.url_path_for("refresh_access_token"),
json={"refresh_token": expired_refresh},
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
async def test_refresh_rejects_invalid_token(
self,
app: FastAPI,
unauthorized_client: AsyncClient,
) -> None:
response = await unauthorized_client.post(
app.url_path_for("refresh_access_token"),
json={"refresh_token": "not-a-valid-token"},
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
async def test_refresh_token_rejected_as_bearer(
self,
app: FastAPI,
unauthorized_client: AsyncClient,
test_user: User,
) -> None:
# a refresh token used as a bearer access token must be rejected
credentials = {"username": test_user.username, "password": "user1_password"}
auth_response = await unauthorized_client.post(app.url_path_for("authenticate"), json=credentials)
refresh_token = auth_response.json()["refresh_token"]
response = await unauthorized_client.get(
app.url_path_for("get_logged_in_user"),
headers={"Authorization": f"Bearer {refresh_token}"},
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
class TestSuperAdmin:
async def test_super_admin_exists(
tests/api/routes/mcp/conftest.py
+12
View File
@@ -0,0 +1,12 @@
"""MCP test fixtures."""
import pytest
@pytest.fixture
def ctx():
"""Standard gns3_ctx for handlers."""
return {
"server_url": "http://192.168.1.3:3080",
"jwt_token": "test-token",
"jwt_username": "admin",
}
tests/api/routes/mcp/test_handlers.py
+515
View File
@@ -0,0 +1,515 @@
"""
MCP handler unit tests with mocked Gns3Connector.
Tests that handlers correctly transform tool parameters into HTTP calls.
"""
import pytest
from unittest.mock import MagicMock, patch
def _mock_conn(json_result=None):
"""Create a mocked Gns3Connector with base_url and http_call."""
conn = MagicMock()
conn.base_url = "http://192.168.1.3:3080/v3"
conn.http_call.return_value.json.return_value = json_result or {"status": "ok"}
return conn
BASE = "gns3server.api.routes.mcp"
@pytest.fixture
def ctx():
return {"server_url": "http://192.168.1.3:3080", "jwt_token": "token", "jwt_username": "admin"}
# ── Project ─────────────────────────────────────────────────────────────
class TestProject:
mod = "projects"
def test_list(self, ctx):
from gns3server.api.routes.mcp.projects import list_projects_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn([{"project_id": "p1", "name": "Test", "status": "opened"}])
result = list_projects_handler({}, ctx)
assert result["count"] == 1
def test_get(self, ctx):
from gns3server.api.routes.mcp.projects import get_project_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"project_id": "p1"})
result = get_project_handler({"project_id": "p1"}, ctx)
assert result["project_id"] == "p1"
def test_get_missing_id(self, ctx):
from gns3server.api.routes.mcp.projects import get_project_handler
assert "error" in get_project_handler({}, ctx)
def test_create(self, ctx):
from gns3server.api.routes.mcp.projects import create_project_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"project_id": "p1"})
result = create_project_handler({"name": "New"}, ctx)
assert result["project_id"] == "p1"
def test_delete(self, ctx):
from gns3server.api.routes.mcp.projects import delete_project_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({})
result = delete_project_handler({"project_id": "p1"}, ctx)
assert "message" in result
def test_open(self, ctx):
from gns3server.api.routes.mcp.projects import open_project_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"status": "opened"})
result = open_project_handler({"project_id": "p1"}, ctx)
assert result["status"] == "opened"
def test_close(self, ctx):
from gns3server.api.routes.mcp.projects import close_project_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"status": "closed"})
result = close_project_handler({"project_id": "p1"}, ctx)
assert "error" not in result
def test_update(self, ctx):
from gns3server.api.routes.mcp.projects import update_project_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"name": "Updated"})
result = update_project_handler({"project_id": "p1", "name": "Updated"}, ctx)
assert result["name"] == "Updated"
def test_stats(self, ctx):
from gns3server.api.routes.mcp.projects import get_project_stats_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"nodes": 5, "links": 3})
result = get_project_stats_handler({"project_id": "p1"}, ctx)
assert result["nodes"] == 5
# ── Node ────────────────────────────────────────────────────────────────
class TestNode:
mod = "nodes"
def test_list_fields(self, ctx):
from gns3server.api.routes.mcp.nodes import get_nodes_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn([
{"node_id": "n1", "name": "R1", "status": "started", "node_type": "qemu", "console": 5000},
])
result = get_nodes_handler({"project_id": "p1", "fields": ["name", "status"]}, ctx)
assert result == {"nodes": [{"name": "R1", "status": "started"}], "count": 1}
def test_list_invalid_fields(self, ctx):
from gns3server.api.routes.mcp.nodes import get_nodes_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn([])
result = get_nodes_handler({"project_id": "p1", "fields": "not-a-list"}, ctx)
assert "error" in result
def test_get(self, ctx):
from gns3server.api.routes.mcp.nodes import get_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"node_id": "n1", "name": "R1"})
result = get_node_handler({"project_id": "p1", "node_id": "n1"}, ctx)
assert result["name"] == "R1"
def test_create_single_passes_name(self, ctx):
from gns3server.api.routes.mcp.nodes import create_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"node_id": "n1", "name": "MyRouter"})
m.return_value = conn
result = create_node_handler({
"project_id": "p1", "template_id": "t1",
"name": "MyRouter", "x": 100, "y": 200,
}, ctx)
conn.http_call.assert_called_with(
"post", "http://192.168.1.3:3080/v3/projects/p1/templates/t1",
json_data={"x": 100, "y": 200, "compute_id": "local", "name": "MyRouter"},
)
assert result == {"node_id": "n1", "name": "MyRouter"}
def test_create_fields_filter(self, ctx):
from gns3server.api.routes.mcp.nodes import create_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"node_id": "n1", "name": "R1", "status": "started"})
result = create_node_handler({
"project_id": "p1", "template_id": "t1",
"fields": ["node_id", "name"],
}, ctx)
assert result == {"node_id": "n1", "name": "R1"}
def test_create_fields_validation(self, ctx):
from gns3server.api.routes.mcp.nodes import create_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn()
m.return_value = conn
result = create_node_handler({
"project_id": "p1", "template_id": "t1", "fields": "not-a-list",
}, ctx)
assert "error" in result
assert "fields must be a list" in result["error"]
conn.http_call.assert_not_called()
def test_create_batch_inherits_template_id(self, ctx):
from gns3server.api.routes.mcp.nodes import create_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"node_id": "n1", "name": "R1"})
result = create_node_handler({
"project_id": "p1", "template_id": "default-tpl",
"nodes": [{"name": "R1", "x": 0, "y": 0}],
}, ctx)
assert result[0]["status"] == "success"
def test_create_missing_project_id(self, ctx):
from gns3server.api.routes.mcp.nodes import create_node_handler
assert create_node_handler({}, ctx) == {"error": "project_id is required"}
def test_delete_batch(self, ctx):
from gns3server.api.routes.mcp.nodes import delete_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({})
result = delete_node_handler({"project_id": "p1", "node_ids": ["n1", "n2"]}, ctx)
assert len(result) == 2
def test_start_batch(self, ctx):
from gns3server.api.routes.mcp.nodes import start_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"status": "started"})
result = start_node_handler({"project_id": "p1", "node_ids": ["n1"]}, ctx)
assert result[0]["status"] == "success"
def test_stop_batch(self, ctx):
from gns3server.api.routes.mcp.nodes import stop_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"status": "stopped"})
result = stop_node_handler({"project_id": "p1", "node_ids": ["n1"]}, ctx)
assert result[0]["status"] == "success"
def test_suspend_batch(self, ctx):
from gns3server.api.routes.mcp.nodes import suspend_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"status": "suspended"})
result = suspend_node_handler({"project_id": "p1", "node_ids": ["n1"]}, ctx)
assert result[0]["status"] == "success"
def test_console(self, ctx):
from gns3server.api.routes.mcp.nodes import get_node_console_info_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"console_url": "ws://host/console"})
result = get_node_console_info_handler({"project_id": "p1", "node_id": "n1"}, ctx)
assert "command" in result
# ── Link ────────────────────────────────────────────────────────────────
class TestLink:
mod = "links"
def test_list(self, ctx):
from gns3server.api.routes.mcp.links import get_links_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn([{"link_id": "l1", "link_type": "ethernet"}])
result = get_links_handler({"project_id": "p1", "fields": ["link_id"]}, ctx)
assert result["links"] == [{"link_id": "l1"}]
def test_get(self, ctx):
from gns3server.api.routes.mcp.links import get_link_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"link_id": "l1", "link_type": "ethernet"})
result = get_link_handler({"project_id": "p1", "link_id": "l1"}, ctx)
assert result["link_id"] == "l1"
def test_create_compact_format(self, ctx):
from gns3server.api.routes.mcp.links import create_link_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"link_id": "l1", "link_type": "ethernet", "nodes": []})
m.return_value = conn
result = create_link_handler({
"project_id": "p1",
"nodes": ["n1", 0, 0, "n2", 0, 0],
}, ctx)
conn.http_call.assert_called_with(
"post", "http://192.168.1.3:3080/v3/projects/p1/links",
json_data={"nodes": [
{"node_id": "n1", "adapter_number": 0, "port_number": 0},
{"node_id": "n2", "adapter_number": 0, "port_number": 0},
]},
)
def test_create_standard_format(self, ctx):
from gns3server.api.routes.mcp.links import create_link_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"link_id": "l1"})
result = create_link_handler({
"project_id": "p1",
"nodes": [
{"node_id": "n1", "adapter_number": 0, "port_number": 0},
{"node_id": "n2", "adapter_number": 0, "port_number": 0},
],
}, ctx)
assert result["link_id"] == "l1"
def test_create_fields_validation(self, ctx):
from gns3server.api.routes.mcp.links import create_link_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn()
m.return_value = conn
result = create_link_handler({
"project_id": "p1", "fields": "bad",
"nodes": ["n1", 0, 0, "n2", 0, 0],
}, ctx)
assert "error" in result
assert "fields must be a list" in result["error"]
conn.http_call.assert_not_called()
def test_delete_batch(self, ctx):
from gns3server.api.routes.mcp.links import delete_link_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({})
result = delete_link_handler({"project_id": "p1", "link_ids": ["l1", "l2"]}, ctx)
assert len(result) == 2
def test_update(self, ctx):
from gns3server.api.routes.mcp.links import update_link_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"link_id": "l1", "suspend": True})
result = update_link_handler({
"project_id": "p1", "link_id": "l1", "suspend": True,
}, ctx)
assert result["suspend"] is True
# ── Appliance ───────────────────────────────────────────────────────────
class TestAppliance:
mod = "appliances"
def test_get(self, ctx):
from gns3server.api.routes.mcp.appliances import get_appliance_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"appliance_id": "a1", "name": "Cisco ISE"})
result = get_appliance_handler({"appliance_id": "a1"}, ctx)
assert result["name"] == "Cisco ISE"
def test_install_with_version(self, ctx):
from gns3server.api.routes.mcp.appliances import install_appliance_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"status": "installed"})
m.return_value = conn
result = install_appliance_handler({
"appliance_id": "a1", "version": "2.7.0.356",
}, ctx)
conn.http_call.assert_called_with(
"post", "http://192.168.1.3:3080/v3/appliances/a1/install",
params={"version": "2.7.0.356"},
)
def test_install_missing_id(self, ctx):
from gns3server.api.routes.mcp.appliances import install_appliance_handler
result = install_appliance_handler({}, ctx)
assert "error" in result
# ── Template ────────────────────────────────────────────────────────────
class TestTemplate:
mod = "templates"
def test_list_fields(self, ctx):
from gns3server.api.routes.mcp.templates import list_templates_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn([
{"template_id": "t1", "name": "Cisco 7200", "template_type": "dynamips",
"category": "router", "default_name_format": "{name}-{0}"},
])
result = list_templates_handler({"fields": ["template_id", "name"]}, ctx)
assert result["templates"] == [{"template_id": "t1", "name": "Cisco 7200"}]
def test_list_invalid_field(self, ctx):
from gns3server.api.routes.mcp.templates import list_templates_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn()
result = list_templates_handler({"fields": ["does_not_exist"]}, ctx)
assert "error" in result
def test_get(self, ctx):
from gns3server.api.routes.mcp.templates import get_template_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"template_id": "t1", "name": "Test"})
result = get_template_handler({"template_id": "t1"}, ctx)
assert result["name"] == "Test"
def test_delete(self, ctx):
from gns3server.api.routes.mcp.templates import delete_template_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({})
result = delete_template_handler({"template_id": "t1"}, ctx)
assert "deleted" in str(result).lower()
# ── Marker (traffic-insight) ────────────────────────────────────────────
class TestLinkMarker:
"""link_marker_handler direction tri-state: omit=preserve, tx/rx=set, both=clear (→ null)."""
mod = "links"
def test_update_direction_both_clears(self, ctx):
from gns3server.api.routes.mcp.links import link_marker_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "icmp"})
m.return_value = conn
link_marker_handler(
{"project_id": "p", "link_id": "l", "action": "update",
"marker_name": "icmp", "direction": "both"}, ctx,
)
conn.http_call.assert_called_with(
"put", "http://192.168.1.3:3080/v3/projects/p/links/l/markers/icmp",
json_data={"direction": None},
)
def test_update_direction_tx_sets(self, ctx):
from gns3server.api.routes.mcp.links import link_marker_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "icmp"})
m.return_value = conn
link_marker_handler(
{"project_id": "p", "link_id": "l", "action": "update",
"marker_name": "icmp", "direction": "tx"}, ctx,
)
conn.http_call.assert_called_with(
"put", "http://192.168.1.3:3080/v3/projects/p/links/l/markers/icmp",
json_data={"direction": "tx"},
)
def test_update_direction_omitted_preserved(self, ctx):
from gns3server.api.routes.mcp.links import link_marker_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "icmp"})
m.return_value = conn
link_marker_handler(
{"project_id": "p", "link_id": "l", "action": "update",
"marker_name": "icmp", "tag": 1}, ctx,
)
conn.http_call.assert_called_with(
"put", "http://192.168.1.3:3080/v3/projects/p/links/l/markers/icmp",
json_data={"tag": 1},
)
def test_create_direction_both_omitted(self, ctx):
from gns3server.api.routes.mcp.links import link_marker_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "icmp"})
m.return_value = conn
link_marker_handler(
{"project_id": "p", "link_id": "l", "action": "create",
"bpf": "icmp", "direction": "both"}, ctx,
)
conn.http_call.assert_called_with(
"post", "http://192.168.1.3:3080/v3/projects/p/links/l/markers",
json_data={"bpf": "icmp"},
)
def test_create_direction_tx(self, ctx):
from gns3server.api.routes.mcp.links import link_marker_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "icmp"})
m.return_value = conn
link_marker_handler(
{"project_id": "p", "link_id": "l", "action": "create",
"bpf": "icmp", "direction": "tx"}, ctx,
)
conn.http_call.assert_called_with(
"post", "http://192.168.1.3:3080/v3/projects/p/links/l/markers",
json_data={"bpf": "icmp", "direction": "tx"},
)
class TestMarkerDefinition:
"""marker_definition_handler build create/update bodies.
A definition has NO direction: it fans out to every link and auto-selects its
capture node on each, so tx/rx (relative to that node) has no consistent
meaning — any direction passed is ignored, never reaching the request body.
"""
mod = "links"
def test_create_builds_body(self, ctx):
from gns3server.api.routes.mcp.links import marker_definition_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "arp"})
m.return_value = conn
marker_definition_handler(
{"project_id": "p", "action": "create",
"bpf": "arp", "tag": 1, "color": "#fff"}, ctx,
)
conn.http_call.assert_called_with(
"post", "http://192.168.1.3:3080/v3/projects/p/marker-definitions",
json_data={"bpf": "arp", "tag": 1, "color": "#fff"},
)
def test_create_ignores_direction(self, ctx):
from gns3server.api.routes.mcp.links import marker_definition_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "arp"})
m.return_value = conn
marker_definition_handler(
{"project_id": "p", "action": "create",
"bpf": "arp", "direction": "tx"}, ctx,
)
conn.http_call.assert_called_with(
"post", "http://192.168.1.3:3080/v3/projects/p/marker-definitions",
json_data={"bpf": "arp"},
)
def test_update_builds_body(self, ctx):
from gns3server.api.routes.mcp.links import marker_definition_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "arp"})
m.return_value = conn
marker_definition_handler(
{"project_id": "p", "action": "update",
"def_name": "arp", "tag": 1}, ctx,
)
conn.http_call.assert_called_with(
"put", "http://192.168.1.3:3080/v3/projects/p/marker-definitions/arp",
json_data={"tag": 1},
)
def test_update_ignores_direction(self, ctx):
from gns3server.api.routes.mcp.links import marker_definition_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"name": "arp"})
m.return_value = conn
marker_definition_handler(
{"project_id": "p", "action": "update",
"def_name": "arp", "tag": 1, "direction": "rx"}, ctx,
)
conn.http_call.assert_called_with(
"put", "http://192.168.1.3:3080/v3/projects/p/marker-definitions/arp",
json_data={"tag": 1},
)
def test_update_requires_a_field(self, ctx):
from gns3server.api.routes.mcp.links import marker_definition_handler
with patch(f"{BASE}.{self.mod}._get_connector"):
result = marker_definition_handler(
{"project_id": "p", "action": "update", "def_name": "arp"}, ctx,
)
assert "error" in result
tests/api/routes/mcp/test_tool_params.py
+254
View File
@@ -0,0 +1,254 @@
"""
MCP tool parameter consistency tests.
Verifies that the parameters defined in each MCP tool function (in __init__.py)
match what the corresponding handler function actually reads via params.get().
This catches issues like:
- A tool parameter is defined but never passed to the handler
- A handler reads a param that was never defined or passed
"""
import ast
import os
import sys
from pathlib import Path
import pytest
MCP_DIR = Path(__file__).resolve().parents[4] / "gns3server" / "api" / "routes" / "mcp"
TOOL_FILE = MCP_DIR / "__init__.py"
HANDLER_FILES = {
"list_projects_handler": "projects.py",
"get_project_handler": "projects.py",
"create_project_handler": "projects.py",
"delete_project_handler": "projects.py",
"open_project_handler": "projects.py",
"close_project_handler": "projects.py",
"get_project_stats_handler": "projects.py",
"update_project_handler": "projects.py",
"duplicate_project_handler": "projects.py",
"get_project_readme_handler": "projects.py",
"update_project_readme_handler": "projects.py",
"lock_project_handler": "projects.py",
"unlock_project_handler": "projects.py",
"get_locked_project_handler": "projects.py",
"load_project_handler": "projects.py",
"get_nodes_handler": "nodes.py",
"get_node_handler": "nodes.py",
"start_node_handler": "nodes.py",
"stop_node_handler": "nodes.py",
"suspend_node_handler": "nodes.py",
"create_node_handler": "nodes.py",
"delete_node_handler": "nodes.py",
"update_node_handler": "nodes.py",
"get_node_console_info_handler": "nodes.py",
"list_node_files_handler": "nodes.py",
"get_node_file_handler": "nodes.py",
"write_node_file_handler": "nodes.py",
"delete_node_file_handler": "nodes.py",
"start_all_nodes_handler": "nodes.py",
"stop_all_nodes_handler": "nodes.py",
"suspend_all_nodes_handler": "nodes.py",
"duplicate_node_handler": "nodes.py",
"isolate_node_handler": "nodes.py",
"unisolate_node_handler": "nodes.py",
"get_node_links_handler": "nodes.py",
"get_links_handler": "links.py",
"get_link_handler": "links.py",
"create_link_handler": "links.py",
"delete_link_handler": "links.py",
"update_link_handler": "links.py",
"reset_link_handler": "links.py",
"start_capture_handler": "links.py",
"stop_capture_handler": "links.py",
"download_capture_file_handler": "links.py",
"list_templates_handler": "templates.py",
"get_template_handler": "templates.py",
"create_template_handler": "templates.py",
"update_template_handler": "templates.py",
"delete_template_handler": "templates.py",
"list_computes_handler": "computes.py",
"get_compute_handler": "computes.py",
"get_compute_images_handler": "computes.py",
"get_snapshots_handler": "snapshots.py",
"create_snapshot_handler": "snapshots.py",
"delete_snapshot_handler": "snapshots.py",
"restore_snapshot_handler": "snapshots.py",
"get_drawings_handler": "drawings.py",
"create_drawing_handler": "drawings.py",
"get_drawing_handler": "drawings.py",
"update_drawing_handler": "drawings.py",
"delete_drawing_handler": "drawings.py",
"get_symbols_handler": "symbols.py",
"get_symbol_handler": "symbols.py",
"get_symbol_dimensions_handler": "symbols.py",
"get_default_symbols_handler": "symbols.py",
"upload_symbol_handler": "symbols.py",
"delete_symbol_handler": "symbols.py",
"get_appliances_handler": "appliances.py",
"get_appliance_handler": "appliances.py",
"install_appliance_handler": "appliances.py",
"get_version_handler": "server.py",
"get_statistics_handler": "server.py",
"get_images_handler": "images.py",
"get_image_handler": "images.py",
"delete_image_handler": "images.py",
"prune_images_handler": "images.py",
"install_images_handler": "images.py",
"device_config_send_handler": "device_config.py",
"device_show_run_handler": "device_config.py",
"vpcs_config_set_handler": "device_config.py",
}
def _get_handler_params(handler_name):
"""Parse handler file and extract all params.get('xxx') calls."""
filename = HANDLER_FILES.get(handler_name)
if not filename:
return None
filepath = MCP_DIR / filename
if not filepath.exists():
return None
tree = ast.parse(filepath.read_text())
params = set()
for node in ast.walk(tree):
if not isinstance(node, ast.FunctionDef):
continue
if node.name != handler_name:
continue
# Found the handler function, search for params.get("xxx")
for sub in ast.walk(node):
if not isinstance(sub, ast.Call):
continue
if not hasattr(sub.func, "attr") or sub.func.attr != "get":
continue
# params.get("xxx") or params_data.get("xxx")
func_obj = sub.func
if (hasattr(func_obj.value, "id") and func_obj.value.id in ("params", "params_data", "link_data", "node_data")) or \
(hasattr(func_obj.value, "attr") and func_obj.value.attr == "get"):
if sub.args and isinstance(sub.args[0], ast.Constant) and isinstance(sub.args[0].value, str):
params.add(sub.args[0].value)
return params
def _get_tool_params(tool_name, tool_file=TOOL_FILE):
"""Parse __init__.py and extract params passed to _run_handler_sync for a given tool.
Returns the dict literal keys from the _run_handler_sync call.
"""
tree = ast.parse(tool_file.read_text())
for node in ast.walk(tree):
if not isinstance(node, ast.FunctionDef):
continue
if node.name != tool_name:
continue
# Search for _run_handler_sync calls inside this function
for sub in ast.walk(node):
if not isinstance(sub, ast.Call):
continue
if not hasattr(sub.func, "id") or sub.func.id != "_run_handler_sync":
continue
# _run_handler_sync(handler, {dict}) or _run_handler_sync(handler, params)
if len(sub.args) >= 2:
second_arg = sub.args[1]
if isinstance(second_arg, ast.Dict):
keys = set()
for k in second_arg.keys:
if isinstance(k, ast.Constant) and isinstance(k.value, str):
keys.add(k.value)
return keys
elif isinstance(second_arg, ast.Name) and second_arg.id == "params":
return {"*params*"} # special marker for all params passed through
return None
def test_handler_params_all_readable():
"""Every handler registered in __init__.py should have a corresponding file."""
# Extract all handler names from __init__.py by looking for _run_handler_sync calls
tree = ast.parse(TOOL_FILE.read_text())
handlers_found = set()
for node in ast.walk(tree):
if isinstance(node, ast.Call) and hasattr(node.func, "id") and node.func.id == "_run_handler_sync":
if node.args and isinstance(node.args[0], ast.Name):
handlers_found.add(node.args[0].id)
unknown = [h for h in handlers_found if h not in HANDLER_FILES]
assert not unknown, f"Handlers not mapped in HANDLER_FILES: {unknown}"
def _get_tool_fn_name(handler_name):
"""Reverse lookup: find which MCP tool function calls this handler."""
tree = ast.parse(TOOL_FILE.read_text())
for node in ast.walk(tree):
if isinstance(node, ast.Call) and hasattr(node.func, "id") and node.func.id == "_run_handler_sync":
if node.args and isinstance(node.args[0], ast.Name) and node.args[0].id == handler_name:
# Find enclosing function
for parent in ast.walk(tree):
if isinstance(parent, ast.FunctionDef):
for child in ast.walk(parent):
if child is node:
return parent.name
return None
def test_tool_handler_param_consistency():
"""For each tool, the params passed to the handler should match what the handler reads."""
tree = ast.parse(TOOL_FILE.read_text())
# Collect all _run_handler_sync calls with dict literals
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
if not hasattr(node.func, "id") or node.func.id != "_run_handler_sync":
continue
if len(node.args) < 2:
continue
handler_name = node.args[0].id if isinstance(node.args[0], ast.Name) else None
if not handler_name:
continue
# Find the tool function name (enclosing function)
tool_name = None
for parent in ast.walk(tree):
if isinstance(parent, ast.FunctionDef):
for child in ast.walk(parent):
if child is node:
tool_name = parent.name
break
if not tool_name:
continue
second_arg = node.args[1]
if isinstance(second_arg, ast.Dict):
passed_keys = set()
for k in second_arg.keys:
if isinstance(k, ast.Constant) and isinstance(k.value, str):
passed_keys.add(k.value)
handler_params = _get_handler_params(handler_name)
if handler_params is None:
continue
# Check: every passed key is read by the handler
extra_passed = passed_keys - handler_params
assert not extra_passed, (
f"[{tool_name}] Params passed to handler '{handler_name}' but not read: {extra_passed}"
)
# Check: every handler param is passed (except common/optional ones)
missing = handler_params - passed_keys
# Filter out well-known optional params that handlers check
known_optional = {"fields", "template", "name", "version", "compute_id",
"x", "y", "link_type", "filters", "suspend", "link_style",
"show_filters_icon", "label"}
truly_missing = missing - known_optional
if truly_missing:
pytest.fail(
f"[{tool_name}] Handler '{handler_name}' reads params not passed: {truly_missing}"
)
tests/api/routes/test_routes.py
+3 -1
View File
@@ -38,10 +38,12 @@ ALLOWED_CONTROLLER_ENDPOINTS = [
("/v3/version", "POST"),
("/v3/access/users/login", "POST"),
("/v3/access/users/authenticate", "POST"),
("/v3/access/users/refresh", "POST"),
("/v3/symbols", "GET"),
("/v3/symbols/{symbol_id:path}/raw", "GET"),
("/v3/symbols/{symbol_id:path}/dimensions", "GET"),
("/v3/symbols/default_symbols", "GET")
("/v3/symbols/default_symbols", "GET"),
("/v3/mcp/", "GET"),
]
class TestRoutes: