You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from dataclasses import replace
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from mcp.server.auth.provider import AccessToken, TokenVerifier
|
|
|
|
from app.config import get_settings
|
|
from app.mcp.server import build_mcp_server
|
|
|
|
|
|
class _StaticTokenVerifier(TokenVerifier):
|
|
async def verify_token(self, token: str) -> AccessToken | None:
|
|
if token != "valid-token":
|
|
return None
|
|
return AccessToken(
|
|
token=token,
|
|
client_id="oauth-client",
|
|
scopes=["availability:read"],
|
|
)
|
|
|
|
|
|
def _oauth_settings():
|
|
return replace(
|
|
get_settings(),
|
|
auth_mode="api_key",
|
|
mcp_auth_mode="oauth",
|
|
mcp_oauth_issuer="https://issuer.example",
|
|
mcp_resource_server_url="https://mcp.example.com/mcp",
|
|
)
|
|
|
|
|
|
def test_mcp_oauth_exposes_protected_resource_metadata() -> None:
|
|
server = build_mcp_server(settings=_oauth_settings(), token_verifier=_StaticTokenVerifier())
|
|
|
|
with TestClient(server.streamable_http_app()) as client:
|
|
response = client.get("/.well-known/oauth-protected-resource/mcp")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["resource"] == "https://mcp.example.com/mcp"
|
|
assert [value.rstrip("/") for value in payload["authorization_servers"]] == [
|
|
"https://issuer.example"
|
|
]
|
|
assert payload.get("scopes_supported") in (None, [])
|
|
|
|
|
|
def test_mcp_oauth_requires_bearer_token_with_challenge() -> None:
|
|
server = build_mcp_server(settings=_oauth_settings(), token_verifier=_StaticTokenVerifier())
|
|
|
|
with TestClient(server.streamable_http_app()) as client:
|
|
response = client.post("/mcp", json={})
|
|
|
|
assert response.status_code == 401
|
|
challenge = response.headers.get("www-authenticate", "")
|
|
assert challenge.startswith("Bearer ")
|
|
assert 'error="invalid_token"' in challenge
|
|
assert "resource_metadata=" in challenge
|
|
assert "/.well-known/oauth-protected-resource/mcp" in challenge
|
|
|
|
|
|
def test_mcp_oauth_mode_requires_resource_server_url() -> None:
|
|
settings = replace(
|
|
get_settings(),
|
|
mcp_auth_mode="oauth",
|
|
mcp_oauth_issuer="https://issuer.example",
|
|
mcp_resource_server_url=None,
|
|
)
|
|
with pytest.raises(ValueError, match="MCP_RESOURCE_SERVER_URL"):
|
|
build_mcp_server(settings=settings, token_verifier=_StaticTokenVerifier())
|
|
|
|
|
|
def test_mcp_server_registers_meeting_interval_tool() -> None:
|
|
server = build_mcp_server(settings=get_settings())
|
|
tool_names = [tool.name for tool in asyncio.run(server.list_tools())]
|
|
|
|
assert "check_availability" in tool_names
|
|
assert "available_meeting_intervals" in tool_names
|