Add test for ProtectedResourceMetadataParsing

tests/server/fastmcp/auth/test_auth_integration.py

@@ -11,6 +11,7 @@
 
 import httpx
 import pytest
+from inline_snapshot import snapshot
 from pydantic import AnyHttpUrl
 from starlette.applications import Starlette
 
@@ -26,6 +27,7 @@
     ClientRegistrationOptions,
     RevocationOptions,
     create_auth_routes,
+    create_protected_resource_routes,
 )
 from mcp.shared.auth import (
     OAuthClientInformationFull,
@@ -196,7 +198,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 +219,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
 
@@ -249,6 +251,32 @@ async def registered_client(test_client: httpx.AsyncClient, request):
     return client_info
 
 
+@pytest.fixture
+def protected_resource_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 protected_resource_test_client(protected_resource_app: Starlette):
+    """Fixture to create an HTTP client for the protected resource app."""
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=protected_resource_app), base_url="https://mcptest.com"
+    ) as client:
+        yield client
+
+
 @pytest.fixture
 def pkce_challenge():
     """Create a PKCE challenge with code_verifier and code_challenge."""
@@ -335,11 +363,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 +412,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"
@@ -1201,3 +1224,24 @@ async def test_authorize_invalid_scope(self, test_client: httpx.AsyncClient, reg
         # State should be preserved
         assert "state" in query_params
         assert query_params["state"][0] == "test_state"
+
+
+class TestProtectedResourceMetadata:
+    """Test the Protected Resource Metadata model."""
+
+    @pytest.mark.anyio
+    async def test_metadata_endpoint(self, protected_resource_test_client: httpx.AsyncClient):
+        """Test the OAuth 2.0 Protected Resource metadata endpoint."""
+
+        response = await protected_resource_test_client.get("/.well-known/oauth-protected-resource")
+        metadata = response.json()
+        assert metadata == 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"],
+            }
+        )

tests/shared/test_auth.py

@@ -1,6 +1,8 @@
 """Tests for OAuth 2.0 shared code."""
 
-from mcp.shared.auth import OAuthMetadata
+import pytest
+
+from mcp.shared.auth import OAuthMetadata, ProtectedResourceMetadata
 
 
 class TestOAuthMetadata:
@@ -37,3 +39,84 @@ def test_oidc(self):
                 "userinfo_endpoint": "https://example.com/oauth2/userInfo",
             }
         )
+
+
+class TestProtectedResourceMetadataInvalid:
+    """Tests for ProtectedResourceMetadata parsing."""
+
+    def test_invalid_metadata(self):
+        """Should throw when parsing invalid metadata."""
+        with pytest.raises(ValueError):
+            ProtectedResourceMetadata.model_validate(
+                {
+                    "resource": "Not a valid URL",
+                    "authorization_servers": ["https://example.com/oauth2/authorize"],
+                    "scopes_supported": ["read", "write"],
+                    "bearer_methods_supported": ["header"],
+                }
+            )
+
+    def test_valid_metadata(self):
+        """Should not throw when parsing protected resource metadata."""
+
+        ProtectedResourceMetadata.model_validate(
+            {
+                "resource": "https://example.com/resource",
+                "authorization_servers": ["https://example.com/oauth2/authorize"],
+                "scopes_supported": ["read", "write"],
+                "bearer_methods_supported": ["header"],
+            }
+        )
+
+    def test_valid_with_resource_metadata(self):
+        """Should not throw when parsing metadata with resource_name and resource_documentation."""
+
+        ProtectedResourceMetadata.model_validate(
+            {
+                "resource": "https://example.com/resource",
+                "authorization_servers": ["https://example.com/oauth2/authorize"],
+                "scopes_supported": ["read", "write"],
+                "bearer_methods_supported": ["header"],
+                "resource_name": "Example Resource",
+                "resource_documentation": "https://example.com/resource/documentation",
+            }
+        )
+
+    def test_valid_with_invalid_resource_documentation(self):
+        """Should throw when parsing metadata with resource_name and invalid resource_documentation."""
+        with pytest.raises(ValueError):
+            ProtectedResourceMetadata.model_validate(
+                {
+                    "resource": "https://example.com/resource",
+                    "authorization_servers": ["https://example.com/oauth2/authorize"],
+                    "scopes_supported": ["read", "write"],
+                    "bearer_methods_supported": ["header"],
+                    "resource_name": "Example Resource",
+                    "resource_documentation": "Not a valid URL",
+                }
+            )
+
+    def test_valid_full_protected_resource_metadata(self):
+        """Should not throw when parsing full metadata."""
+
+        ProtectedResourceMetadata.model_validate(
+            {
+                "resource": "https://example.com/resource",
+                "authorization_servers": ["https://example.com/oauth2/authorize"],
+                "jwks_uri": "https://example.com/.well-known/jwks.json",
+                "scopes_supported": ["read", "write"],
+                "bearer_methods_supported": ["header"],
+                "resource_signing_alg_values_supported": ["RS256"],
+                "resource_name": "Example Resource",
+                "resource_documentation": "https://example.com/resource/documentation",
+                "resource_policy_uri": "https://example.com/resource/policy",
+                "resource_tos_uri": "https://example.com/resource/tos",
+                "tls_client_certificate_bound_access_tokens": True,
+                # authorization_details_types_supported is a complex type
+                # so we use an empty list for simplicity
+                # see RFC9396
+                "authorization_details_types_supported": [],
+                "dpop_signing_alg_values_supported": ["RS256", "ES256"],
+                "dpop_signing_access_tokens": True,
+            }
+        )