bvandeusen
94f7a6de37
Wires FastMCP's streamable-HTTP ASGI sub-app into the Quart app via asgi_app replacement. Requests under /mcp are stripped, auth-checked against api_keys, and forwarded to FastMCP with fable_user_id set on the ASGI scope. All other paths pass through to the original Quart dispatch unchanged. Tests cover the three auth paths (no header, invalid token, valid token) plus a regression check that non-/mcp paths bypass the MCP dispatch. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
"""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.
|
|
"""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from fabledassistant.app import create_app
|
|
|
|
|
|
@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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mcp_endpoint_invalid_token_returns_401():
|
|
app = create_app()
|
|
with patch(
|
|
"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
|
|
|
|
|
|
@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.
|
|
"""
|
|
fake_key = MagicMock()
|
|
fake_key.user_id = 7
|
|
fake_key.scope = "write"
|
|
app = create_app()
|
|
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
|
|
|
|
|
|
@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."""
|
|
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
|