From 38265906f1a85dd05342d1fc9de54ae79c35ddf4 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 26 May 2026 19:54:46 -0400 Subject: [PATCH] test(mcp): drive ASGI app directly, skip Quart test_client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Quart's test_client expects its request pipeline to populate app._preserved_context. Our /mcp middleware deliberately bypasses that pipeline (forwarding straight to FastMCP), so test_client's teardown blew up with AttributeError. The middleware is correct; the test harness was wrong. Build raw ASGI scope/receive/send and call app.asgi_app directly — which is what production hypercorn does anyway. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_mcp_endpoint.py | 129 ++++++++++++++++++++++++------------- 1 file changed, 84 insertions(+), 45 deletions(-) diff --git a/tests/test_mcp_endpoint.py b/tests/test_mcp_endpoint.py index e1577cb..c903d59 100644 --- a/tests/test_mcp_endpoint.py +++ b/tests/test_mcp_endpoint.py @@ -1,9 +1,16 @@ """End-to-end tests for the /mcp HTTP endpoint mounted on the Quart app. -These exercise the ASGI dispatch + auth middleware. The api_keys lookup is -mocked so the tests don't require a database, matching the pattern in -test_api_keys.py. +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 FastMCP 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 @@ -11,15 +18,55 @@ import pytest from fabledassistant.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() - async with app.test_client() as client: - resp = await client.post( - "/mcp", - json={"jsonrpc": "2.0", "id": 1, "method": "ping"}, - ) - assert resp.status_code == 401 + status, _ = await _send_request(app, "POST", "/mcp") + assert status == 401 @pytest.mark.asyncio @@ -29,56 +76,48 @@ async def test_mcp_endpoint_invalid_token_returns_401(): "fabledassistant.mcp.auth.lookup_key", AsyncMock(return_value=None), ): - async with app.test_client() as client: - resp = await client.post( - "/mcp", - json={"jsonrpc": "2.0", "id": 1, "method": "ping"}, - headers={"Authorization": "Bearer fmcp_invalid"}, - ) - assert resp.status_code == 401 + 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 reaches FastMCP. FastMCP may return - 200 (initialize handshake), 400, or a streaming response — anything other - than 401 confirms auth passed and the request was handed off. - """ + """With a valid Bearer, the request reaches FastMCP. Anything other than 401 + confirms auth passed and the request was handed off to the sub-app.""" 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() with patch( "fabledassistant.mcp.auth.lookup_key", AsyncMock(return_value=fake_key), ): - async with app.test_client() as client: - resp = await client.post( - "/mcp", - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "initialize", - "params": { - "protocolVersion": "2024-11-05", - "capabilities": {}, - "clientInfo": {"name": "test", "version": "0"}, - }, - }, - headers={ - "Authorization": "Bearer fmcp_valid", - "Accept": "application/json, text/event-stream", - }, - ) - assert resp.status_code != 401 + status, _ = await _send_request( + app, "POST", "/mcp", + headers={ + "Authorization": "Bearer fmcp_valid", + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + }, + body=initialize_body, + ) + assert status != 401 @pytest.mark.asyncio async def test_non_mcp_paths_bypass_mcp_dispatch(): - """A non-/mcp path must not be routed to the FastMCP sub-app. Hitting an - unknown /api/* route should return 404 from Quart, not 401 from the MCP - middleware.""" + """A non-/mcp path must reach Quart's normal routing, not the MCP dispatch.""" app = create_app() - async with app.test_client() as client: - resp = await client.get("/api/this-route-does-not-exist") - assert resp.status_code != 401 + status, _ = await _send_request(app, "GET", "/api/this-route-does-not-exist") + assert status != 401