86 lines
3.6 KiB
Python
86 lines
3.6 KiB
Python
|
"""Unit tests for MCPAuthMiddleware.
|
||
|
|
|
||
|
|
Wraps a minimal dummy ASGI app (not the real MCP app) so these tests exercise
|
||
|
|
only the auth gate, not the MCP protocol itself — keeps the test fast and
|
||
|
|
independent of FastMCP internals.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import sys
|
||
|
|
from unittest.mock import MagicMock, patch
|
||
|
|
|
||
|
|
# app.mcp.server imports app.shared.bootstrap at module scope (needed for
|
||
|
|
# get_agent_conversation_service/get_jwt_handler), which in turn eagerly
|
||
|
|
# imports several Postgres store modules that do `import psycopg2` at their
|
||
|
|
# own module scope. Since this is the only test file in backend/tests/mcp/
|
||
|
|
# that imports app.mcp.server at module scope (not inside a test function),
|
||
|
|
# it is the first thing to trigger that chain during pytest collection —
|
||
|
|
# guard psycopg2 here the same way backend/tests/observability/
|
||
|
|
# test_model_usage_bootstrap.py and test_model_usage_persistence.py already
|
||
|
|
# do, so a real (network-connecting) psycopg2 never gets bound first.
|
||
|
|
mock_psycopg2 = MagicMock()
|
||
|
|
mock_psycopg2.extras = MagicMock()
|
||
|
|
sys.modules.setdefault("psycopg2", mock_psycopg2)
|
||
|
|
sys.modules.setdefault("psycopg2.extras", mock_psycopg2.extras)
|
||
|
|
sys.modules.setdefault("psycopg2.pool", MagicMock())
|
||
|
|
|
||
|
|
from starlette.applications import Starlette
|
||
|
|
from starlette.responses import PlainTextResponse
|
||
|
|
from starlette.routing import Route
|
||
|
|
from starlette.testclient import TestClient
|
||
|
|
|
||
|
|
from app.mcp.server import MCPAuthMiddleware
|
||
|
|
|
||
|
|
|
||
|
|
def _dummy_app() -> Starlette:
|
||
|
|
"""Build a minimal Starlette app that MCPAuthMiddleware can wrap."""
|
||
|
|
async def _ok(request):
|
||
|
|
"""Return a fixed 200 response so tests can assert pass-through."""
|
||
|
|
return PlainTextResponse("ok")
|
||
|
|
|
||
|
|
app = Starlette(routes=[Route("/ping", _ok)])
|
||
|
|
app.add_middleware(MCPAuthMiddleware)
|
||
|
|
return app
|
||
|
|
|
||
|
|
|
||
|
|
def test_missing_token_rejected_when_auth_enabled():
|
||
|
|
"""No Authorization header + auth_enabled=True -> 401."""
|
||
|
|
with patch("app.mcp.server.settings") as fake_settings:
|
||
|
|
fake_settings.auth_enabled = True
|
||
|
|
client = TestClient(_dummy_app())
|
||
|
|
response = client.get("/ping")
|
||
|
|
assert response.status_code == 401
|
||
|
|
|
||
|
|
|
||
|
|
def test_invalid_token_rejected_when_auth_enabled():
|
||
|
|
"""A token that fails decode_token() -> 401, request never reaches the app."""
|
||
|
|
fake_handler = type("H", (), {"decode_token": lambda self, t: (_ for _ in ()).throw(ValueError("bad token"))})()
|
||
|
|
with patch("app.mcp.server.settings") as fake_settings, \
|
||
|
|
patch("app.mcp.server.get_jwt_handler", return_value=fake_handler):
|
||
|
|
fake_settings.auth_enabled = True
|
||
|
|
client = TestClient(_dummy_app())
|
||
|
|
response = client.get("/ping", headers={"Authorization": "Bearer garbage"})
|
||
|
|
assert response.status_code == 401
|
||
|
|
|
||
|
|
|
||
|
|
def test_valid_token_passes_through_when_auth_enabled():
|
||
|
|
"""A token that decodes successfully -> request reaches the wrapped app."""
|
||
|
|
fake_handler = type("H", (), {"decode_token": lambda self, t: object()})()
|
||
|
|
with patch("app.mcp.server.settings") as fake_settings, \
|
||
|
|
patch("app.mcp.server.get_jwt_handler", return_value=fake_handler):
|
||
|
|
fake_settings.auth_enabled = True
|
||
|
|
client = TestClient(_dummy_app())
|
||
|
|
response = client.get("/ping", headers={"Authorization": "Bearer good"})
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert response.text == "ok"
|
||
|
|
|
||
|
|
|
||
|
|
def test_auth_disabled_always_passes_through():
|
||
|
|
"""auth_enabled=False (dev mode) -> no token needed, matches get_current_user's dev bypass."""
|
||
|
|
with patch("app.mcp.server.settings") as fake_settings:
|
||
|
|
fake_settings.auth_enabled = False
|
||
|
|
client = TestClient(_dummy_app())
|
||
|
|
response = client.get("/ping")
|
||
|
|
assert response.status_code == 200
|