qonto-assistant/tests/test_mcp_server.py

from pathlib import Path

import pytest
from mcp.server.fastmcp.exceptions import ToolError

from qonto_assistant.audit import AuditLogger
from qonto_assistant.config import Settings
from qonto_assistant.mcp_server import create_mcp_server
from qonto_assistant.policy import PolicyEngine
from qonto_assistant.qonto_client import FixtureQontoClient
from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter
from qonto_assistant.service import CapabilityService

FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "qonto"
POLICY_FILE = Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml"


def _settings() -> Settings:
    return Settings.from_env()


def _service() -> tuple[CapabilityService, list[dict[str, object]]]:
    events: list[dict[str, object]] = []
    policy = PolicyEngine.from_file(POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False)
    service = CapabilityService(
        client=FixtureQontoClient(fixture_dir=FIXTURE_DIR),
        policy=policy,
        audit_logger=AuditLogger(sink=events.append),
        rate_limiter=RateLimiter(limit=20, window_seconds=60),
        concurrency_limiter=ConcurrencyLimiter(limit=4),
    )
    return service, events


async def test_mcp_server_starts_and_lists_smoke_tool() -> None:
    server = create_mcp_server(settings=_settings())
    tools = await server.list_tools()
    tool_names = {tool.name for tool in tools}
    assert "qonto_ping" in tool_names


async def test_mcp_server_without_service_only_exposes_smoke_tool() -> None:
    server = create_mcp_server(settings=_settings())
    tools = await server.list_tools()
    tool_names = {tool.name for tool in tools}
    assert tool_names == {"qonto_ping"}


async def test_mcp_server_lists_capability_tools_when_service_wired() -> None:
    service, _ = _service()
    server = create_mcp_server(settings=_settings(), service=service)
    tools = await server.list_tools()
    tool_names = {tool.name for tool in tools}
    assert tool_names == {
        "qonto_ping",
        "qonto_org_summary",
        "qonto_list_transactions",
        "qonto_cost_run_rate_hints",
    }


async def test_qonto_org_summary_tool_returns_redacted_balances() -> None:
    service, events = _service()
    server = create_mcp_server(settings=_settings(), service=service)

    result = await server.call_tool("qonto_org_summary", {})

    assert result[1]["organization"]["name"] == "Binky Hedgehog GmbH"
    assert "iban" not in result[1]["accounts"][0]
    assert events[-1]["capability"] == "org_summary"
    assert events[-1]["protocol"] == "mcp"


async def test_qonto_list_transactions_tool_denies_oversized_page_size() -> None:
    service, events = _service()
    server = create_mcp_server(settings=_settings(), service=service)

    with pytest.raises(ToolError):
        await server.call_tool("qonto_list_transactions", {"page_size": 101})

    assert events[-1]["decision"] == "deny"
    assert events[-1]["deny_reason"] == "arg_constraint"
    assert events[-1]["protocol"] == "mcp"


async def test_qonto_cost_run_rate_hints_tool_returns_recurring_debits() -> None:
    service, events = _service()
    server = create_mcp_server(settings=_settings(), service=service)

    result = await server.call_tool("qonto_cost_run_rate_hints", {"window_days": 90})

    hints = result[1]["cost_run_rate_hints"]["recurring_debits"]
    assert len(hints) > 0
    assert all("amount" in hint for hint in hints)
    assert events[-1]["capability"] == "cost_run_rate_hints"


async def test_capability_tools_are_absent_without_service() -> None:
    server = create_mcp_server(settings=_settings())
    with pytest.raises(ToolError):
        await server.call_tool("qonto_org_summary", {})