|
26 | 26 | ClientRegistrationOptions,
|
27 | 27 | RevocationOptions,
|
28 | 28 | create_auth_routes,
|
| 29 | + create_protected_resource_routes, |
29 | 30 | )
|
30 | 31 | from mcp.shared.auth import (
|
31 | 32 | OAuthClientInformationFull,
|
@@ -216,6 +217,25 @@ def auth_app(mock_oauth_provider):
|
216 | 217 | return app
|
217 | 218 |
|
218 | 219 |
|
| 220 | +@pytest.fixture |
| 221 | +def protected_resource_app(auth_app): |
| 222 | + """Fixture to create protected resource routes for testing.""" |
| 223 | + |
| 224 | + print(auth_app.router.routes) |
| 225 | + # Create the protected resource routes |
| 226 | + protected_resource_routes = create_protected_resource_routes( |
| 227 | + resource_url=AnyHttpUrl("https://example.com/resource"), |
| 228 | + authorization_servers=[AnyHttpUrl("https://auth.example.com/authorization")], |
| 229 | + scopes_supported=["read", "write"], |
| 230 | + resource_name="Example Resource", |
| 231 | + resource_documentation=AnyHttpUrl("https://docs.example.com/resource"), |
| 232 | + ) |
| 233 | + |
| 234 | + # add routes to the auth app |
| 235 | + auth_app.router.routes.extend(protected_resource_routes) |
| 236 | + return auth_app |
| 237 | + |
| 238 | + |
219 | 239 | @pytest.fixture
|
220 | 240 | async def test_client(auth_app):
|
221 | 241 | async with httpx.AsyncClient(transport=httpx.ASGITransport(app=auth_app), base_url="https://mcptest.com") as client:
|
@@ -1201,3 +1221,26 @@ async def test_authorize_invalid_scope(self, test_client: httpx.AsyncClient, reg
|
1201 | 1221 | # State should be preserved
|
1202 | 1222 | assert "state" in query_params
|
1203 | 1223 | assert query_params["state"][0] == "test_state"
|
| 1224 | + |
| 1225 | + |
| 1226 | +class TestProtectedResourceMetadata: |
| 1227 | + """Test the Protected Resource Metadata model.""" |
| 1228 | + |
| 1229 | + @pytest.mark.anyio |
| 1230 | + async def test_metadata_endpoint(self, protected_resource_app: Starlette, test_client: httpx.AsyncClient): |
| 1231 | + """Test the OAuth 2.0 Protected Resource metadata endpoint.""" |
| 1232 | + print("Sending request to protected resource metadata endpoint") |
| 1233 | + response = await test_client.get("/.well-known/oauth-protected-resource") |
| 1234 | + print(f"Got response: {response.status_code}") |
| 1235 | + if response.status_code != 200: |
| 1236 | + print(f"Response content: {response.content}") |
| 1237 | + assert response.status_code == 200 |
| 1238 | + |
| 1239 | + metadata = response.json() |
| 1240 | + print(f"Protected Resource Metadata: {metadata}") |
| 1241 | + assert metadata["resource"] == "https://example.com/resource" |
| 1242 | + assert metadata["authorization_servers"] == ["https://auth.example.com/authorization"] |
| 1243 | + assert metadata["scopes_supported"] == ["read", "write"] |
| 1244 | + assert metadata["resource_name"] == "Example Resource" |
| 1245 | + assert metadata["resource_documentation"] == "https://docs.example.com/resource" |
| 1246 | + assert metadata["bearer_methods_supported"] == ["header"] |
0 commit comments