test: add tests

Author: xiaoyijun

mcpauth/auth/authorization_server_handler.py

@@ -90,4 +90,4 @@ def get_token_verifier(self, resource: str) -> TokenVerifier:
         This is a dummy implementation that ignores the resource, as there is only
         one `TokenVerifier` in the authorization server mode.
         """
-        return self.token_verifier 
+        return self.token_verifier 

mcpauth/auth/mcp_auth_handler.py

@@ -17,12 +17,12 @@ def create_metadata_route(self) -> Router:
         Returns a router for serving either the legacy OAuth 2.0 Authorization Server Metadata or
         the OAuth 2.0 Protected Resource Metadata, depending on the configuration.
         """
-        ...
+        ...  # pragma: no cover
 
     @abstractmethod
     def get_token_verifier(self, resource: str) -> TokenVerifier:
         """
         Resolves the appropriate TokenVerifier based on the provided resource.
         :param resource: The resource identifier for verifier lookup.
         """
-        ... 
+        ...  # pragma: no cover 

mcpauth/auth/resource_server_handler.py

@@ -116,4 +116,4 @@ def _validate_config(self, resource_configs: List[ResourceServerConfig]):
                             cause={"error_description": f"The authorization server ('{issuer}') for resource '{resource}' is duplicated."},
                         )
                     unique_auth_servers.add(issuer)
-                    validate_server_config(auth_server) 
+                    validate_server_config(auth_server) 

mcpauth/auth/token_verifier.py

@@ -85,4 +85,4 @@ def verify_jwt(token: str) -> AuthInfo:
             verify_function = create_verify_jwt(jwks_uri, leeway=leeway)
             return verify_function(token)
 
-        return verify_jwt 
+        return verify_jwt 

mcpauth/middleware/create_bearer_auth.py

@@ -106,7 +106,7 @@ def _handle_error(
     www_authenticate_header = BearerWWWAuthenticateHeader()
 
     if isinstance(error, (MCPAuthTokenVerificationException, MCPAuthBearerAuthException)):
-        www_authenticate_header.set_parameter_if_value_exists("error", error.code)
+        www_authenticate_header.set_parameter_if_value_exists("error", error.code.value)
         if error.message:
             www_authenticate_header.set_parameter_if_value_exists("error_description", error.message)
 

mcpauth/utils/_transpile_resource_metadata.py

@@ -31,4 +31,4 @@ def transpile_resource_metadata(
     if auth_servers:
         model_data["authorization_servers"] = auth_servers
 
-    return ProtectedResourceMetadata(**model_data) 
+    return ProtectedResourceMetadata(**model_data) 

samples/server/todo-manager/server.py

@@ -24,7 +24,7 @@
     MCPAuthBearerAuthException,
     BearerAuthExceptionCode,
 )
-from mcpauth.types import AuthInfo
+from mcpauth.types import AuthInfo, ResourceServerConfig, ResourceServerMetadata
 from mcpauth.utils import fetch_server_config
 from .service import TodoService
 
@@ -44,7 +44,22 @@
     )
 
 auth_server_config = fetch_server_config(auth_issuer, AuthServerType.OIDC)
-mcp_auth = MCPAuth(server=auth_server_config)
+resource_id = "https://todo-manager.mcp-auth.com/resource1"
+mcp_auth = MCPAuth(
+    protected_resources=[
+        ResourceServerConfig(
+            metadata=ResourceServerMetadata(
+                resource=resource_id,
+                authorization_servers=[auth_server_config],
+                scopes_supported=[
+                    "create:todos",
+                    "read:todos",
+                    "delete:todos",
+                ],
+            )
+        )
+    ]
+)
 
 def assert_user_id(auth_info: Optional[AuthInfo]) -> str:
     """Assert that auth_info contains a valid user ID and return it."""
@@ -122,12 +137,11 @@ def delete_todo(id: str) -> dict[str, Any]:
         return {"error": "Failed to delete todo"}
 
 # Create the middleware and app
-bearer_auth = Middleware(mcp_auth.bearer_auth_middleware('jwt'))
+bearer_auth = Middleware(mcp_auth.bearer_auth_middleware('jwt', resource=resource_id))
 app = Starlette(
     routes=[
-        # Add the metadata route (`/.well-known/oauth-authorization-server`)
-        mcp_auth.metadata_route(), # pyright: ignore[reportDeprecated]
         # Protect the MCP server with the Bearer auth middleware
+        *mcp_auth.resource_metadata_router().routes,
         Mount("/", app=mcp.sse_app(), middleware=[bearer_auth]),
     ],
 ) 

tests/__init__test.py

@@ -124,6 +124,18 @@ def test_bearer_auth_middleware_calls_get_token_verifier_in_resource_mode(
             mock_get_verifier.assert_called_once_with(resource="https://api.example.com")
 
 
+def test_bearer_auth_middleware_throws_for_invalid_mode(
+    valid_server_config: AuthServerConfig,
+):
+    """Test that bearer_auth_middleware throws a ValueError for an invalid mode."""
+    auth = MCPAuth(server=valid_server_config)
+    with pytest.raises(
+        ValueError,
+        match="mode_or_verify must be 'jwt' or a callable function that verifies tokens.",
+    ):
+        auth.bearer_auth_middleware(mode_or_verify="invalid_mode")  # type: ignore
+
+
 @patch("mcpauth.auth.resource_server_handler.validate_server_config")
 def test_metadata_route_throws_in_resource_mode(
     mock_validate: MagicMock, valid_resource_config: ResourceServerConfig
@@ -164,6 +176,26 @@ def test_metadata_route_calls_handler_method(
     mock_create_route.assert_called_once()
 
 
+@patch(
+    "mcpauth.auth.authorization_server_handler.AuthorizationServerHandler.create_metadata_route"
+)
+@patch("mcpauth.auth.authorization_server_handler.validate_server_config")
+def test_metadata_route_throws_if_route_is_not_route_instance(
+    mock_validate: MagicMock,
+    mock_create_route: MagicMock,
+    valid_server_config: AuthServerConfig,
+):
+    """Test that metadata_route throws an error if the created route is not a Route instance."""
+    # Ensure the mock returns a router-like object with a routes attribute
+    # containing something that is not a Route instance
+    mock_create_route.return_value = MagicMock(routes=[MagicMock()])
+    auth = MCPAuth(server=valid_server_config)
+    with pytest.warns(DeprecationWarning):
+        with pytest.raises(IndexError, match="No metadata endpoint route was created"):
+            auth.metadata_route()  # pyright: ignore[reportDeprecated]
+    mock_create_route.assert_called_once()
+
+
 @patch(
     "mcpauth.auth.resource_server_handler.ResourceServerHandler.create_metadata_route"
 )
@@ -176,4 +208,20 @@ def test_resource_metadata_router_calls_handler_method(
     """Test that resource_metadata_router calls the handler's create_metadata_route method."""
     auth = MCPAuth(protected_resources=valid_resource_config)
     auth.resource_metadata_router()
-    mock_create_route.assert_called_once() 
+    mock_create_route.assert_called_once()
+
+
+@patch("mcpauth.middleware.create_bearer_auth.create_bearer_auth")
+def test_bearer_auth_middleware_with_callable_verifier(
+    mock_create_bearer_auth: MagicMock, valid_server_config: AuthServerConfig
+):
+    """Test that bearer_auth_middleware works with a callable verifier."""
+    auth = MCPAuth(server=valid_server_config)
+    verifier = MagicMock()
+    with patch("mcpauth.MCPAuthHandler.get_token_verifier"):
+        auth.bearer_auth_middleware(mode_or_verify=verifier)
+
+    mock_create_bearer_auth.assert_called_once()
+    # Check that the verifier is passed to create_bearer_auth
+    args, _ = mock_create_bearer_auth.call_args
+    assert args[0] == verifier 

tests/auth/authorization_server_handler_test.py

@@ -150,4 +150,4 @@ def test_get_token_verifier(valid_auth_server_config: AuthServerConfig):
             AuthServerModeConfig(server=valid_auth_server_config)
         )
         verifier = handler.get_token_verifier("test-resource")
-        assert verifier == handler.token_verifier 
+        assert verifier == handler.token_verifier 

tests/auth/resource_server_handler_test.py

@@ -91,6 +91,33 @@ def test_init_duplicate_resource_id(mock_validate: Mock, mock_resource_config: R
     assert excinfo.value.code == AuthServerExceptionCode.INVALID_SERVER_CONFIG
 
 
+@patch(
+    "mcpauth.auth.resource_server_handler.validate_server_config",
+    return_value=type("ValidationResult", (), {"is_valid": True}),
+)
+def test_init_duplicate_auth_server(
+    mock_validate: Mock, mock_auth_server: AuthServerConfig
+):
+    """Test that ResourceServerHandler throws an error if an auth server is duplicated for a resource."""
+    config_with_duplicate_auth_server = RSC(
+        metadata=ResourceServerMetadata(
+            resource="https://my-api.com",
+            authorization_servers=[mock_auth_server, mock_auth_server],
+        )
+    )
+    with pytest.raises(MCPAuthAuthServerException) as excinfo:
+        ResourceServerHandler(
+            ResourceServerModeConfig(
+                protected_resources=[config_with_duplicate_auth_server]
+            )
+        )
+    assert excinfo.value.code == AuthServerExceptionCode.INVALID_SERVER_CONFIG
+    assert (
+        excinfo.value.cause["error_description"]  # type: ignore[reportGeneralTypeIssues]
+        == "The authorization server ('https://auth.example.com') for resource 'https://my-api.com' is duplicated."
+    )
+
+
 @patch(
     "mcpauth.auth.resource_server_handler.validate_server_config",
     return_value=type("ValidationResult", (), {"is_valid": True}),
@@ -153,4 +180,4 @@ def test_create_metadata_route(mock_validate: Mock, mock_resource_config: RSC):
     assert response.json()["authorization_servers"] == ["https://auth.example.com"]
 
     response = client.options(endpoint_path)
-    assert response.status_code == 204 
+    assert response.status_code == 204