test: add tests

Author: xiaoyijun

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/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)
 

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/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}),

tests/middleware/create_bearer_auth_test.py

@@ -18,7 +18,9 @@
     MCPAuthAuthServerException,
     MCPAuthConfigException,
     MCPAuthTokenVerificationExceptionCode,
+    MCPAuthBearerAuthException,
 )
+import pydantic
 
 
 class TestHandleBearerAuth:
@@ -58,6 +60,16 @@ def test_should_throw_error_if_issuer_is_not_a_valid_url(
                 auth_info,
             )
 
+    def test_should_throw_error_if_issuer_is_not_string_or_callable(
+        self, auth_info: ContextVar[AuthInfo | None]
+    ):
+        with pytest.raises(pydantic.ValidationError):
+            create_bearer_auth(
+                lambda _: None,  # type: ignore
+                BearerAuthConfig(issuer=123),  # type: ignore
+                auth_info,
+            )
+
 
 @pytest.mark.asyncio
 class TestHandleBearerAuthMiddleware:
@@ -287,6 +299,7 @@ async def test_should_respond_with_error_if_audience_does_not_match(
         ],
     ):
         mock_verify = MagicMock()
+        assert isinstance(auth_config[1].issuer, str)
         mock_verify.return_value = AuthInfo(
             issuer=auth_config[1].issuer,
             client_id="client-id",
@@ -333,6 +346,7 @@ async def test_should_respond_with_error_if_audience_does_not_match_array_case(
         ],
     ):
         mock_verify = MagicMock()
+        assert isinstance(auth_config[1].issuer, str)
         mock_verify.return_value = AuthInfo(
             issuer=auth_config[1].issuer,
             client_id="client-id",
@@ -379,6 +393,7 @@ async def test_should_respond_with_error_if_required_scopes_are_not_present(
         ],
     ):
         mock_verify = MagicMock()
+        assert isinstance(auth_config[1].issuer, str)
         mock_verify.return_value = AuthInfo(
             issuer=auth_config[1].issuer,
             client_id="client-id",
@@ -643,3 +658,99 @@ async def test_should_show_error_details_for_bearer_auth_error(self):
             },
         }
         mock_verify.assert_called_once_with("valid-token")
+
+    @pytest.mark.asyncio
+    async def test_should_include_resource_metadata_on_401_bearer_error(
+        self,
+        auth_info: ContextVar[AuthInfo | None],
+    ):
+        """
+        Test that the WWW-Authenticate header includes the resource_metadata URI when a
+        Bearer auth error (401) occurs and a resource is specified.
+        """
+        config = BearerAuthConfig(
+            issuer="https://correct-issuer.com",
+            resource="https://my-api.com",
+        )
+        mock_verify = MagicMock()
+        mock_verify.return_value = AuthInfo(
+            issuer="https://wrong-issuer.com",
+            client_id="client-id",
+            scopes=[],
+            token="valid-token",
+            subject="subject-id",
+            audience=None,
+            claims={},
+        )
+
+        MiddlewareClass = create_bearer_auth(mock_verify, config, auth_info)
+        middleware = MiddlewareClass(app=MagicMock())
+
+        request = Request(
+            scope={
+                "type": "http",
+                "headers": [(b"authorization", b"Bearer valid-token")],
+                "method": "GET",
+                "path": "/",
+            }
+        )
+
+        response = await middleware.dispatch(request, MagicMock())
+
+        assert response.status_code == 401
+        assert "WWW-Authenticate" in response.headers
+        assert (
+            'resource_metadata="https://my-api.com/.well-known/oauth-protected-resource"'
+            in response.headers["WWW-Authenticate"]
+        )
+
+    async def test_should_respond_with_error_if_callable_issuer_fails(
+        self,
+        auth_config: tuple[
+            VerifyAccessTokenFunction, BearerAuthConfig, ContextVar[AuthInfo | None]
+        ],
+    ):
+        """Test that an error is returned if a callable issuer fails validation."""
+        mock_verify = MagicMock()
+        mock_verify.return_value = AuthInfo(
+            issuer="https://some-issuer.com",
+            client_id="client-id",
+            scopes=[],
+            token="valid-token",
+            audience=None,
+            subject="subject-id",
+            claims={},
+        )
+
+        def failing_issuer_validator(issuer: str):
+            raise MCPAuthBearerAuthException(BearerAuthExceptionCode.INVALID_ISSUER)
+
+        config = BearerAuthConfig(
+            issuer=failing_issuer_validator,
+            resource="https://my-api.com",
+        )
+
+        MiddlewareClass = create_bearer_auth(mock_verify, config, auth_config[2])
+        middleware = MiddlewareClass(app=MagicMock())
+
+        request = Request(
+            scope={
+                "type": "http",
+                "headers": [(b"authorization", b"Bearer valid-token")],
+                "method": "GET",
+                "path": "/",
+            }
+        )
+
+        response = await middleware.dispatch(request, MagicMock())
+
+        assert response.status_code == 401
+        assert "WWW-Authenticate" in response.headers
+        assert (
+            f'error="{BearerAuthExceptionCode.INVALID_ISSUER.value}"'
+            in response.headers["WWW-Authenticate"]
+        )
+        assert (
+            'resource_metadata="https://my-api.com/.well-known/oauth-protected-resource"'
+            in response.headers["WWW-Authenticate"]
+        )