mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-08-28 21:10:14 +03:00
Merge pull request #2791 from yueguobin/fix/mcp-test-issues
Fix MCP testing issues: appliance_install version, handler unit tests, parameter validation
This commit is contained in:
Jeremy Grossmann
GitHub
commit
4ffab7abee
@ -1284,6 +1284,7 @@ async def appliance_get(
|
||||
@mcp.tool()
|
||||
async def appliance_install(
|
||||
appliance_id: Annotated[str, Field(description="UUID of the appliance to install")],
|
||||
version: Annotated[str | None, Field(description="Version to install (e.g. '2.7.0.356'). Required if the appliance has multiple versions. Use appliance_get to see available versions.")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Create a template from a GNS3 appliance definition.
|
||||
|
||||
@ -1294,6 +1295,7 @@ async def appliance_install(
|
||||
"""
|
||||
return await asyncio.to_thread(_run_handler_sync, install_appliance_handler, {
|
||||
"appliance_id": appliance_id,
|
||||
"version": version,
|
||||
})
|
||||
|
||||
|
||||
|
||||
@ -80,5 +80,10 @@ def install_appliance_handler(params: dict[str, Any], gns3_ctx: dict[str, Any])
|
||||
if not appliance_id:
|
||||
return {"error": "appliance_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
result = conn.http_call("post", f"{conn.base_url}/appliances/{appliance_id}/install").json()
|
||||
url = f"{conn.base_url}/appliances/{appliance_id}/install"
|
||||
request_params = {}
|
||||
version = params.get("version")
|
||||
if version:
|
||||
request_params["version"] = version
|
||||
result = conn.http_call("post", url, params=request_params).json()
|
||||
return {"message": f"Appliance {appliance_id} installation requested", "result": result}
|
||||
|
||||
12
tests/api/routes/mcp/conftest.py
Normal file
12
tests/api/routes/mcp/conftest.py
Normal 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",
|
||||
}
|
||||
368
tests/api/routes/mcp/test_handlers.py
Normal file
368
tests/api/routes/mcp/test_handlers.py
Normal file
@ -0,0 +1,368 @@
|
||||
"""
|
||||
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_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(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()
|
||||
256
tests/api/routes/mcp/test_tool_params.py
Normal file
256
tests/api/routes/mcp/test_tool_params.py
Normal file
@ -0,0 +1,256 @@
|
||||
"""
|
||||
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",
|
||||
"reload_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",
|
||||
"reload_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}"
|
||||
)
|
||||
Loading…
x
Reference in New Issue
Block a user