hold-slayer/tests/test_mcp.py

"""
MCP server tests — tool surface, lazy gateway resolution, call safety.

Uses the FastMCP in-memory client (no network, no mounted app).
"""

import pytest
from fastmcp import Client
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool

import db.database as dbmod
from config import Settings, get_settings
from core.dial_plan import is_emergency_number
from core.gateway import AIPSTNGateway
from db.database import Base
from mcp_server.server import create_mcp_server


@pytest.fixture
async def mem_db(monkeypatch):
    engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        poolclass=StaticPool,
        connect_args={"check_same_thread": False},
    )
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    factory = async_sessionmaker(engine, expire_on_commit=False)
    monkeypatch.setattr(dbmod, "_engine", engine)
    monkeypatch.setattr(dbmod, "_session_factory", factory)
    yield factory
    await engine.dispose()

EXPECTED_TOOLS = {
    "make_call",
    "get_call_status",
    "transfer_call",
    "hangup",
    "list_active_calls",
    "get_call_flow",
    "create_call_flow",
    "send_dtmf",
    "get_call_transcript",
    "get_call_recording",
    "get_call_summary",
    "search_call_history",
    "learn_call_flow",
    "list_devices",
    "gateway_status",
}


def _make_gateway(max_calls: int = 4) -> AIPSTNGateway:
    """Unstarted gateway on the in-memory MockSIPEngine — no network, no DB."""
    return AIPSTNGateway(settings=Settings(max_concurrent_calls=max_calls))


class TestToolSurface:
    async def test_tool_listing_matches_expected(self):
        mcp = create_mcp_server(lambda: None)
        async with Client(mcp) as client:
            tools = {t.name for t in await client.list_tools()}
        assert tools == EXPECTED_TOOLS

    async def test_no_fastmcp_auth_configured(self):
        # Auth for /mcp is enforced by the ASGI _owner_only_mcp guard in
        # main.py, not on the FastMCP instance itself.
        assert create_mcp_server(lambda: None).auth is None


class TestMcpOwnerGuard:
    """The ASGI wrapper gates /mcp before the inner app runs."""

    def _wrapped(self):
        import main

        calls = {"inner": 0}

        async def inner(scope, receive, send):
            calls["inner"] += 1
            await send({"type": "http.response.start", "status": 200, "headers": []})
            await send({"type": "http.response.body", "body": b"ok"})

        return main._owner_only_mcp(inner), calls

    async def _run(self, app, headers):
        sent = []

        async def receive():
            return {"type": "http.request", "body": b"", "more_body": False}

        async def send(msg):
            sent.append(msg)

        scope = {
            "type": "http",
            "method": "POST",
            "path": "/mcp/",
            "headers": headers,
            "query_string": b"",
        }
        await app(scope, receive, send)
        status = next(m["status"] for m in sent if m["type"] == "http.response.start")
        return status

    async def test_missing_token_401(self, monkeypatch, mem_db):
        monkeypatch.setattr(get_settings().casdoor, "enabled", True)
        monkeypatch.setattr(get_settings().casdoor, "endpoint", "https://id.example.test")
        monkeypatch.setattr(get_settings(), "owner_name", "owner@example.test")
        app, calls = self._wrapped()
        status = await self._run(app, headers=[])
        assert status == 401
        assert calls["inner"] == 0

    async def test_dev_owner_passes(self, monkeypatch, mem_db):
        monkeypatch.setattr(get_settings().casdoor, "enabled", False)
        app, calls = self._wrapped()
        status = await self._run(app, headers=[])
        assert status == 200
        assert calls["inner"] == 1


class TestGatewayResolution:
    async def test_tool_errors_cleanly_before_gateway_ready(self):
        mcp = create_mcp_server(lambda: None)
        async with Client(mcp) as client:
            with pytest.raises(Exception, match="starting up"):
                await client.call_tool("list_active_calls", {})

    async def test_make_call_happy_path(self):
        gateway = _make_gateway()
        mcp = create_mcp_server(lambda: gateway)
        async with Client(mcp) as client:
            result = await client.call_tool(
                "make_call", {"number": "+15551234567", "mode": "direct"}
            )
        text = result.content[0].text
        assert "initiated" in text
        assert "+15551234567" in text
        assert len(gateway.call_manager.active_calls) == 1


class TestCallSafety:
    def test_emergency_number_detection(self):
        for number in ("911", "9911", "112", "+1911", "+112", " 911 ", "9-1-1"):
            assert is_emergency_number(number), number
        for number in ("+19115551234", "+18005551234", "211", "999"):
            assert not is_emergency_number(number), number

    async def test_gateway_refuses_emergency_numbers(self):
        gateway = _make_gateway()
        with pytest.raises(ValueError, match="emergency"):
            await gateway.make_call("911")
        assert gateway.call_manager.active_calls == {}

    async def test_mcp_make_call_refuses_emergency(self):
        gateway = _make_gateway()
        mcp = create_mcp_server(lambda: gateway)
        async with Client(mcp) as client:
            with pytest.raises(Exception, match="[Ee]mergency"):
                await client.call_tool("make_call", {"number": "911"})

    async def test_concurrent_call_cap(self):
        gateway = _make_gateway(max_calls=1)
        await gateway.make_call("+15551234567")
        with pytest.raises(ValueError, match="limit"):
            await gateway.make_call("+15557654321")