gns3-server/tests/agent/mcp/test_tool_params.py

"""
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

REPO_ROOT = Path(__file__).resolve().parents[3]
MCP_DIR = REPO_ROOT / "gns3server" / "agent" / "mcp"
# Node/link handlers live in the shared REST client layer, not MCP_DIR.
API_HANDLERS_FILE = "gns3server/agent/gns3_copilot/gns3_client/api_handlers.py"
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": API_HANDLERS_FILE,
    "get_node_handler": API_HANDLERS_FILE,
    "start_node_handler": API_HANDLERS_FILE,
    "stop_node_handler": API_HANDLERS_FILE,
    "suspend_node_handler": API_HANDLERS_FILE,
    "create_node_handler": API_HANDLERS_FILE,
    "delete_node_handler": API_HANDLERS_FILE,
    "update_node_handler": API_HANDLERS_FILE,
    "get_node_console_info_handler": API_HANDLERS_FILE,
    "list_node_files_handler": API_HANDLERS_FILE,
    "get_node_file_handler": API_HANDLERS_FILE,
    "write_node_file_handler": API_HANDLERS_FILE,
    "delete_node_file_handler": API_HANDLERS_FILE,
    "start_all_nodes_handler": API_HANDLERS_FILE,
    "stop_all_nodes_handler": API_HANDLERS_FILE,
    "suspend_all_nodes_handler": API_HANDLERS_FILE,
    "duplicate_node_handler": API_HANDLERS_FILE,
    "isolate_node_handler": API_HANDLERS_FILE,
    "unisolate_node_handler": API_HANDLERS_FILE,
    "get_node_links_handler": API_HANDLERS_FILE,
    "available_filters_handler": API_HANDLERS_FILE,
    "get_links_handler": API_HANDLERS_FILE,
    "get_link_handler": API_HANDLERS_FILE,
    "create_link_handler": API_HANDLERS_FILE,
    "delete_link_handler": API_HANDLERS_FILE,
    "update_link_handler": API_HANDLERS_FILE,
    "reset_link_handler": API_HANDLERS_FILE,
    "start_capture_handler": API_HANDLERS_FILE,
    "stop_capture_handler": API_HANDLERS_FILE,
    "download_capture_file_handler": API_HANDLERS_FILE,
    "link_marker_handler": API_HANDLERS_FILE,
    "marker_definition_handler": API_HANDLERS_FILE,
    "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 if not filename.startswith('gns3server/') else REPO_ROOT) / 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, ast.AsyncFunctionDef)):
            continue
        if node.name != handler_name:
            continue
        # A handler that forwards params.items() generically (e.g. building the
        # request body from all params) accepts any key — return a wildcard and
        # let the caller skip static consistency checks for it.
        for sub in ast.walk(node):
            if (isinstance(sub, ast.Call) and isinstance(sub.func, ast.Attribute)
                    and sub.func.attr == "items"
                    and isinstance(sub.func.value, ast.Name)
                    and sub.func.value.id in ("params", "params_data")):
                return {"*"}
        # 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 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 dispatch calls
    tree = ast.parse(TOOL_FILE.read_text())
    handlers_found = set()
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        handler_name, _ = _dispatch_args(node)
        if handler_name:
            handlers_found.add(handler_name)

    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 _dict_literal_keys(d):
    """String keys of an ast.Dict literal (non-constant keys like **spread are skipped)."""
    keys = set()
    for k in d.keys:
        if isinstance(k, ast.Constant) and isinstance(k.value, str):
            keys.add(k.value)
    return keys


def _initial_params_keys(fn_node):
    """Keys of the initial 'params = {...}' dict literal inside a tool function, or None.

    Conditional additions (params["k"] = v) after the literal are not included.
    """
    for stmt in ast.walk(fn_node):
        if isinstance(stmt, ast.Assign):
            if (len(stmt.targets) == 1 and isinstance(stmt.targets[0], ast.Name)
                    and stmt.targets[0].id == "params" and isinstance(stmt.value, ast.Dict)):
                return _dict_literal_keys(stmt.value)
    return None


def _dispatch_args(node):
    """Extract (handler_name, payload_arg) from a handler dispatch call.

    Matches both forms used by tools:
      - _run_handler_sync(handler, payload)
      - asyncio.to_thread(_run_handler_sync, handler, payload)
    Returns (None, None) if the call is neither.
    """
    if isinstance(node.func, ast.Name) and node.func.id == "_run_handler_sync":
        args = node.args
    elif (isinstance(node.func, ast.Attribute) and node.func.attr == "to_thread"
            and node.args and isinstance(node.args[0], ast.Name) and node.args[0].id == "_run_handler_sync"):
        args = node.args[1:]
    else:
        return None, None
    if len(args) < 2 or not isinstance(args[0], ast.Name):
        return None, None
    return args[0].id, args[1]


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())

    # Group dispatch calls by (tool, handler): a tool may dispatch the same
    # handler from multiple branches (e.g. node_create single/batch modes),
    # each passing only its branch's keys — the union covers all of them.
    groups = {}
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        handler_name, second_arg = _dispatch_args(node)
        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, ast.AsyncFunctionDef)):
                for child in ast.walk(parent):
                    if child is node:
                        tool_name = parent.name
                        break
        if not tool_name:
            continue

        exact = False
        if isinstance(second_arg, ast.Dict):
            passed_keys = _dict_literal_keys(second_arg)
            exact = True
        elif isinstance(second_arg, ast.Name) and second_arg.id == "params":
            # Tool builds 'params' as a variable — resolve its initial dict literal.
            fn = next((n for n in ast.walk(tree)
                       if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) and n.name == tool_name), None)
            passed_keys = _initial_params_keys(fn) if fn is not None else None
        else:
            passed_keys = None
        if not passed_keys:
            continue

        group = groups.setdefault((tool_name, handler_name), {"passed": set(), "exact": False})
        group["passed"] |= passed_keys
        group["exact"] = group["exact"] or exact

    for (tool_name, handler_name), group in groups.items():
        handler_params = _get_handler_params(handler_name)
        if handler_params is None or "*" in handler_params:
            continue

        passed_keys = group["passed"]

        # 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}"
        )

        if not group["exact"]:
            # The initial literal underestimates what the tool passes (keys may be
            # added conditionally), so only the extra-passed direction is checked.
            continue

        # 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}"
            )