Max Isbey
1b74b06753
Cut comment and docstring volume roughly in half across src, tests, examples, and docs_src: removed comments that restate the adjacent code, leftover development narration, section banners, and self-evident Args/Returns blocks, and compressed the remaining docstrings to a Google-style summary line plus only the detail that earns its place. Kept (and tightened) the load-bearing content: Raises sections, deprecation and version-availability notes, spec/RFC/issue references, why-comments for non-obvious decisions, and all coverage pragmas. The generated mcp_types.v* wire modules are untouched.
163 lines
5.8 KiB
Python
163 lines
5.8 KiB
Python
"""Integration tests for MCP Oauth Protected Resource."""
|
|
|
|
from urllib.parse import urlparse
|
|
|
|
import httpx
|
|
import pytest
|
|
from inline_snapshot import snapshot
|
|
from pydantic import AnyHttpUrl
|
|
from starlette.applications import Starlette
|
|
|
|
from mcp.server.auth.routes import build_resource_metadata_url, create_protected_resource_routes
|
|
|
|
|
|
@pytest.fixture
|
|
def test_app():
|
|
protected_resource_routes = create_protected_resource_routes(
|
|
resource_url=AnyHttpUrl("https://example.com/resource"),
|
|
authorization_servers=[AnyHttpUrl("https://auth.example.com/authorization")],
|
|
scopes_supported=["read", "write"],
|
|
resource_name="Example Resource",
|
|
resource_documentation=AnyHttpUrl("https://docs.example.com/resource"),
|
|
)
|
|
|
|
app = Starlette(routes=protected_resource_routes)
|
|
return app
|
|
|
|
|
|
@pytest.fixture
|
|
async def test_client(test_app: Starlette):
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=test_app), base_url="https://mcptest.com") as client:
|
|
yield client
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_metadata_endpoint_with_path(test_client: httpx.AsyncClient):
|
|
response = await test_client.get("/.well-known/oauth-protected-resource/resource")
|
|
assert response.json() == snapshot(
|
|
{
|
|
"resource": "https://example.com/resource",
|
|
"authorization_servers": ["https://auth.example.com/authorization"],
|
|
"scopes_supported": ["read", "write"],
|
|
"resource_name": "Example Resource",
|
|
"resource_documentation": "https://docs.example.com/resource",
|
|
"bearer_methods_supported": ["header"],
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_metadata_endpoint_root_path_returns_404(test_client: httpx.AsyncClient):
|
|
response = await test_client.get("/.well-known/oauth-protected-resource")
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.fixture
|
|
def root_resource_app():
|
|
protected_resource_routes = create_protected_resource_routes(
|
|
resource_url=AnyHttpUrl("https://example.com"),
|
|
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
|
scopes_supported=["read"],
|
|
resource_name="Root Resource",
|
|
)
|
|
|
|
app = Starlette(routes=protected_resource_routes)
|
|
return app
|
|
|
|
|
|
@pytest.fixture
|
|
async def root_resource_client(root_resource_app: Starlette):
|
|
async with httpx.AsyncClient(
|
|
transport=httpx.ASGITransport(app=root_resource_app), base_url="https://mcptest.com"
|
|
) as client:
|
|
yield client
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_metadata_endpoint_without_path(root_resource_client: httpx.AsyncClient):
|
|
response = await root_resource_client.get("/.well-known/oauth-protected-resource")
|
|
assert response.status_code == 200
|
|
assert response.json() == snapshot(
|
|
{
|
|
"resource": "https://example.com/",
|
|
"authorization_servers": ["https://auth.example.com/"],
|
|
"scopes_supported": ["read"],
|
|
"resource_name": "Root Resource",
|
|
"bearer_methods_supported": ["header"],
|
|
}
|
|
)
|
|
|
|
|
|
def test_metadata_url_construction_url_without_path():
|
|
resource_url = AnyHttpUrl("https://example.com")
|
|
result = build_resource_metadata_url(resource_url)
|
|
assert str(result) == "https://example.com/.well-known/oauth-protected-resource"
|
|
|
|
|
|
def test_metadata_url_construction_url_with_path_component():
|
|
resource_url = AnyHttpUrl("https://example.com/mcp")
|
|
result = build_resource_metadata_url(resource_url)
|
|
assert str(result) == "https://example.com/.well-known/oauth-protected-resource/mcp"
|
|
|
|
|
|
def test_metadata_url_construction_url_with_trailing_slash_only():
|
|
resource_url = AnyHttpUrl("https://example.com/")
|
|
result = build_resource_metadata_url(resource_url)
|
|
# Trailing slash should be treated as empty path
|
|
assert str(result) == "https://example.com/.well-known/oauth-protected-resource"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"resource_url,expected_url",
|
|
[
|
|
("https://example.com", "https://example.com/.well-known/oauth-protected-resource"),
|
|
("https://example.com/", "https://example.com/.well-known/oauth-protected-resource"),
|
|
("https://example.com/mcp", "https://example.com/.well-known/oauth-protected-resource/mcp"),
|
|
("http://localhost:8001/mcp", "http://localhost:8001/.well-known/oauth-protected-resource/mcp"),
|
|
],
|
|
)
|
|
def test_metadata_url_construction_various_resource_configurations(resource_url: str, expected_url: str):
|
|
result = build_resource_metadata_url(AnyHttpUrl(resource_url))
|
|
assert str(result) == expected_url
|
|
|
|
|
|
def test_route_consistency_route_path_matches_metadata_url():
|
|
resource_url = AnyHttpUrl("https://example.com/mcp")
|
|
|
|
metadata_url = build_resource_metadata_url(resource_url)
|
|
|
|
routes = create_protected_resource_routes(
|
|
resource_url=resource_url,
|
|
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
|
)
|
|
|
|
metadata_path = urlparse(str(metadata_url)).path
|
|
|
|
assert len(routes) == 1
|
|
assert routes[0].path == metadata_path
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"resource_url,expected_path",
|
|
[
|
|
("https://example.com", "/.well-known/oauth-protected-resource"),
|
|
("https://example.com/", "/.well-known/oauth-protected-resource"),
|
|
("https://example.com/mcp", "/.well-known/oauth-protected-resource/mcp"),
|
|
],
|
|
)
|
|
def test_route_consistency_consistent_paths_for_various_resources(resource_url: str, expected_path: str):
|
|
resource_url_obj = AnyHttpUrl(resource_url)
|
|
|
|
metadata_url = build_resource_metadata_url(resource_url_obj)
|
|
url_path = urlparse(str(metadata_url)).path
|
|
|
|
routes = create_protected_resource_routes(
|
|
resource_url=resource_url_obj,
|
|
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
|
)
|
|
route_path = routes[0].path
|
|
|
|
assert url_path == expected_path
|
|
assert route_path == expected_path
|
|
assert url_path == route_path
|