Add project/node/link handler tests: 39 total, covering list/get/create/delete/start/stop/suspend/reload/console/update/fields

This commit is contained in:
YueGuobin 2026-06-18 00:07:14 +08:00
parent 5cc826baff
commit 471e65e48c
No known key found for this signature in database

View File

@ -15,37 +15,115 @@ def _mock_conn(json_result=None):
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"}
class TestGetNodes:
# ── Project ─────────────────────────────────────────────────────────────
def test_fields_filter(self, ctx):
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("gns3server.api.routes.mcp.nodes._get_connector") as m:
conn = _mock_conn([
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},
])
m.return_value = conn
result = get_nodes_handler({"project_id": "p1", "fields": ["name", "status"]}, ctx)
assert result == {"nodes": [{"name": "R1", "status": "started"}], "count": 1}
def test_invalid_fields(self, ctx):
def test_list_invalid_fields(self, ctx):
from gns3server.api.routes.mcp.nodes import get_nodes_handler
with patch("gns3server.api.routes.mcp.nodes._get_connector") as m:
conn = _mock_conn([])
m.return_value = conn
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"
class TestCreateNode:
def test_single_passes_name(self, ctx):
def test_create_single_passes_name(self, ctx):
from gns3server.api.routes.mcp.nodes import create_node_handler
with patch("gns3server.api.routes.mcp.nodes._get_connector") as m:
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({
@ -58,20 +136,19 @@ class TestCreateNode:
)
assert result == {"node_id": "n1", "name": "MyRouter"}
def test_fields_filter(self, ctx):
def test_create_fields_filter(self, ctx):
from gns3server.api.routes.mcp.nodes import create_node_handler
with patch("gns3server.api.routes.mcp.nodes._get_connector") as m:
conn = _mock_conn({"node_id": "n1", "name": "R1", "status": "started", "node_type": "qemu", "console": 5000})
m.return_value = conn
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_fields_validation(self, ctx):
def test_create_fields_validation(self, ctx):
from gns3server.api.routes.mcp.nodes import create_node_handler
with patch("gns3server.api.routes.mcp.nodes._get_connector") as m:
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn()
m.return_value = conn
result = create_node_handler({
@ -81,27 +158,87 @@ class TestCreateNode:
assert "fields must be a list" in result["error"]
conn.http_call.assert_not_called()
def test_batch_inherits_template_id(self, ctx):
def test_create_batch_inherits_template_id(self, ctx):
from gns3server.api.routes.mcp.nodes import create_node_handler
with patch("gns3server.api.routes.mcp.nodes._get_connector") as m:
conn = _mock_conn({"node_id": "n1", "name": "R1"})
m.return_value = conn
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_missing_project_id(self, ctx):
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
class TestCreateLink:
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_compact_format(self, ctx):
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_reload_batch(self, ctx):
from gns3server.api.routes.mcp.nodes import reload_node_handler
with patch(f"{BASE}.{self.mod}._get_connector") as m:
m.return_value = _mock_conn({"status": "started"})
result = reload_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("gns3server.api.routes.mcp.links._get_connector") as m:
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({
@ -110,19 +247,16 @@ class TestCreateLink:
}, 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},
]
},
json_data={"nodes": [
{"node_id": "n1", "adapter_number": 0, "port_number": 0},
{"node_id": "n2", "adapter_number": 0, "port_number": 0},
]},
)
def test_standard_format(self, ctx):
def test_create_standard_format(self, ctx):
from gns3server.api.routes.mcp.links import create_link_handler
with patch("gns3server.api.routes.mcp.links._get_connector") as m:
conn = _mock_conn({"link_id": "l1"})
m.return_value = conn
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": [
@ -132,9 +266,9 @@ class TestCreateLink:
}, ctx)
assert result["link_id"] == "l1"
def test_fields_validation(self, ctx):
def test_create_fields_validation(self, ctx):
from gns3server.api.routes.mcp.links import create_link_handler
with patch("gns3server.api.routes.mcp.links._get_connector") as m:
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn()
m.return_value = conn
result = create_link_handler({
@ -145,20 +279,40 @@ class TestCreateLink:
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("gns3server.api.routes.mcp.appliances._get_connector") as m:
conn = _mock_conn({"appliance_id": "a1", "name": "Cisco ISE"})
m.return_value = conn
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("gns3server.api.routes.mcp.appliances._get_connector") as m:
with patch(f"{BASE}.{self.mod}._get_connector") as m:
conn = _mock_conn({"status": "installed"})
m.return_value = conn
result = install_appliance_handler({
@ -169,24 +323,46 @@ class TestAppliance:
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
class TestTemplates:
# ── Template ────────────────────────────────────────────────────────────
class TestTemplate:
mod = "templates"
def test_list_fields(self, ctx):
from gns3server.api.routes.mcp.templates import list_templates_handler
with patch("gns3server.api.routes.mcp.templates._get_connector") as m:
conn = _mock_conn([
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}"},
])
m.return_value = conn
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("gns3server.api.routes.mcp.templates._get_connector") as m:
conn = _mock_conn()
m.return_value = conn
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()