diff --git a/tests/api/routes/mcp/test_handlers.py b/tests/api/routes/mcp/test_handlers.py index 45e1e6054..02b15a989 100644 --- a/tests/api/routes/mcp/test_handlers.py +++ b/tests/api/routes/mcp/test_handlers.py @@ -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()