DataOps-platform/tests/mcp/test_server_contracts.py

import inspect

from app.core.mcp.servers import (
    CONTEXT_TOOL_NAMES,
    SCHEDULING_TOOL_NAMES,
    create_context_mcp,
    create_scheduling_mcp,
)


class FakeFastMCP:
    def __init__(self, name, **kwargs):
        self.name = name
        self.kwargs = kwargs
        self.tools = {}

    def tool(self):
        def register(function):
            self.tools[function.__name__] = function
            return function

        return register


class Service:
    def __getattr__(self, name):
        return lambda *_args, **_kwargs: {"tool": name}


def test_context_mcp_exposes_only_bounded_read_and_validation_tools():
    server = create_context_mcp(
        Service(),
        identity=object(),
        fastmcp_cls=FakeFastMCP,
    )

    assert set(server.tools) == CONTEXT_TOOL_NAMES
    assert server.kwargs["json_response"] is True
    assert not {"delete_flow", "read_business_rows"} & set(server.tools)


def test_scheduling_mcp_exposes_only_compound_gateway_tools():
    server = create_scheduling_mcp(
        Service(),
        identity=object(),
        fastmcp_cls=FakeFastMCP,
    )

    assert set(server.tools) == SCHEDULING_TOOL_NAMES
    assert "delete_flow" not in server.tools
    assert "create_yaml" not in server.tools
    assert "configure_connection" not in server.tools
    assert "task_tokens" not in inspect.signature(server.tools["run_canary"]).parameters
    assert (
        "canary" not in inspect.signature(server.tools["promote_candidate"]).parameters
    )