FabledGrimoire/tests/test_mcp_endpoint.py

"""End-to-end tests for the /mcp HTTP endpoint mounted on the Quart app.

These exercise the ASGI dispatch + auth middleware by driving the app's ASGI
callable directly. We avoid Quart's test_client() because it expects Quart's
request pipeline to set `app._preserved_context` as a side effect, but our
ASGI middleware forwards /mcp requests to MCPServer without touching the Quart
pipeline (correct production behavior), which causes test_client to fail on
teardown.

The api_keys lookup is mocked so the tests don't require a database, matching
the pattern in test_api_keys.py.
"""
import json
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from grimoire.app import create_app


async def _send_request(
    app, method: str, path: str,
    headers: dict | None = None, body: bytes | None = None,
) -> tuple[int, bytes]:
    """Drive the app's ASGI callable directly and collect the response."""
    raw_headers = [(k.lower().encode(), v.encode()) for k, v in (headers or {}).items()]
    if body is not None:
        raw_headers.append((b"content-length", str(len(body)).encode()))
    scope = {
        "type": "http",
        "asgi": {"version": "3.0", "spec_version": "2.3"},
        "http_version": "1.1",
        "method": method,
        "scheme": "http",
        "path": path,
        "raw_path": path.encode(),
        "query_string": b"",
        "root_path": "",
        "headers": raw_headers,
        "client": ("testclient", 50000),
        "server": ("testserver", 80),
    }
    receive_messages = [{
        "type": "http.request", "body": body or b"", "more_body": False,
    }]
    sent: list[dict] = []

    async def receive():
        if receive_messages:
            return receive_messages.pop(0)
        return {"type": "http.disconnect"}

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

    await app.asgi_app(scope, receive, send)

    start = next(m for m in sent if m["type"] == "http.response.start")
    body_chunks = b"".join(
        m.get("body", b"") for m in sent if m["type"] == "http.response.body"
    )
    return start["status"], body_chunks


@pytest.mark.asyncio
async def test_mcp_endpoint_unauthenticated_returns_401():
    app = create_app()
    status, _ = await _send_request(app, "POST", "/mcp")
    assert status == 401


@pytest.mark.asyncio
async def test_mcp_endpoint_invalid_token_returns_401():
    app = create_app()
    with patch(
        "grimoire.mcp.auth.lookup_key",
        AsyncMock(return_value=None),
    ):
        status, _ = await _send_request(
            app, "POST", "/mcp",
            headers={"Authorization": "Bearer fmcp_invalid"},
        )
    assert status == 401


@pytest.mark.asyncio
async def test_mcp_endpoint_valid_token_passes_auth():
    """With a valid Bearer, the request must successfully reach MCPServer's
    initialize handler. Asserting `!= 401` is too weak: it lets a 404 from
    a path-mismatch (the original bug) through. MCPServer responds 200 to a
    well-formed initialize handshake.

    MCPServer's session manager normally starts via Quart's @before_serving
    hook in production. This raw-ASGI test doesn't go through Quart's
    serving lifecycle, so we manually enter the session manager."""
    fake_key = MagicMock()
    fake_key.user_id = 7
    fake_key.scope = "write"
    app = create_app()
    initialize_body = json.dumps({
        "jsonrpc": "2.0", "id": 1, "method": "initialize",
        "params": {
            "protocolVersion": "2024-11-05",
            "capabilities": {},
            "clientInfo": {"name": "test", "version": "0"},
        },
    }).encode()
    async with app.mcp_instance.session_manager.run():
        with patch(
            "grimoire.mcp.auth.lookup_key",
            AsyncMock(return_value=fake_key),
        ):
            status, _ = await _send_request(
                app, "POST", "/mcp",
                headers={
                    # MCPServer's transport_security module enforces a Host
                    # header (DNS-rebinding protection); without it the
                    # request gets a 421 Misdirected Request.
                    "Host": "testserver",
                    "Authorization": "Bearer fmcp_valid",
                    "Content-Type": "application/json",
                    "Accept": "application/json, text/event-stream",
                },
                body=initialize_body,
            )
    assert status == 200, f"expected 200 from MCPServer initialize, got {status}"


# Note: there's no explicit "non-/mcp paths bypass the middleware" test here
# because driving Quart's full request pipeline through a hand-rolled ASGI
# scope (no lifespan startup, no hypercorn state) doesn't produce a response.
# The bypass behavior is implicit: if the middleware ate non-/mcp requests,
# the rest of the test suite (~250 tests hitting /api/*) would break.