diff --git a/tests/server/auth/test_protected_resource.py b/tests/server/auth/test_protected_resource.py new file mode 100644 index 000000000..0dc34625d --- /dev/null +++ b/tests/server/auth/test_protected_resource.py @@ -0,0 +1,52 @@ +""" +Integration tests for MCP Oauth Protected Resource. +""" + +import httpx +import pytest +from inline_snapshot import snapshot +from pydantic import AnyHttpUrl +from starlette.applications import Starlette + +from mcp.server.auth.routes import create_protected_resource_routes + + +@pytest.fixture +def test_app(): + """Fixture to create protected resource routes for testing.""" + + # Create the protected resource routes + 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): + """Fixture to create an HTTP client for the protected resource app.""" + 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(test_client: httpx.AsyncClient): + """Test the OAuth 2.0 Protected Resource metadata endpoint.""" + + response = await test_client.get("/.well-known/oauth-protected-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"], + } + ) diff --git a/tests/server/fastmcp/auth/test_auth_integration.py b/tests/server/fastmcp/auth/test_auth_integration.py index e4a8f3f4c..a09a9ea71 100644 --- a/tests/server/fastmcp/auth/test_auth_integration.py +++ b/tests/server/fastmcp/auth/test_auth_integration.py @@ -196,7 +196,7 @@ def mock_oauth_provider(): @pytest.fixture -def auth_app(mock_oauth_provider): +def auth_app(mock_oauth_provider: MockOAuthProvider): # Create auth router auth_routes = create_auth_routes( mock_oauth_provider, @@ -217,7 +217,7 @@ def auth_app(mock_oauth_provider): @pytest.fixture -async def test_client(auth_app): +async def test_client(auth_app: Starlette): async with httpx.AsyncClient(transport=httpx.ASGITransport(app=auth_app), base_url="https://mcptest.com") as client: yield client @@ -335,11 +335,8 @@ class TestAuthEndpoints: @pytest.mark.anyio async def test_metadata_endpoint(self, test_client: httpx.AsyncClient): """Test the OAuth 2.0 metadata endpoint.""" - print("Sending request to metadata endpoint") + response = await test_client.get("/.well-known/oauth-authorization-server") - print(f"Got response: {response.status_code}") - if response.status_code != 200: - print(f"Response content: {response.content}") assert response.status_code == 200 metadata = response.json() @@ -387,9 +384,7 @@ async def test_token_invalid_auth_code(self, test_client, registered_client, pkc "redirect_uri": "https://client.example.com/callback", }, ) - print(f"Status code: {response.status_code}") - print(f"Response body: {response.content}") - print(f"Response JSON: {response.json()}") + assert response.status_code == 400 error_response = response.json() assert error_response["error"] == "invalid_grant"