Add tests and coverage pragmas for client_secret_basic auth support

- Add tests for Basic auth error cases (invalid base64, no colon, client_id mismatch)
- Add test for 'none' auth method with public clients
- Add test for explicit token_endpoint_auth_method setting
- Add pragma: no cover for defensive error handling paths

Author: pcarleton

src/mcp/client/auth/oauth2.py

@@ -186,10 +186,10 @@ def prepare_token_auth(
             Tuple of (updated_data, updated_headers)
         """
         if headers is None:
-            headers = {}
+            headers = {}  # pragma: no cover
 
         if not self.client_info:
-            return data, headers
+            return data, headers  # pragma: no cover
 
         auth_method = self.client_info.token_endpoint_auth_method
 

src/mcp/server/auth/handlers/revoke.py

@@ -41,7 +41,7 @@ async def handle(self, request: Request) -> Response:
         """
         try:
             client = await self.client_authenticator.authenticate_request(request)
-        except AuthenticationError as e:
+        except AuthenticationError as e:  # pragma: no cover
             return PydanticJSONResponse(
                 status_code=401,
                 content=RevocationErrorResponse(

src/mcp/server/auth/handlers/token.py

@@ -110,7 +110,7 @@ async def handle(self, request: Request):
         try:
             form_data = await request.form()
             token_request = TokenRequest.model_validate(dict(form_data)).root
-        except ValidationError as validation_error:
+        except ValidationError as validation_error:  # pragma: no cover
             return self.response(
                 TokenErrorResponse(
                     error="invalid_request",

src/mcp/server/auth/middleware/client_auth.py

@@ -93,7 +93,9 @@ async def authenticate_request(self, request: Request) -> OAuthClientInformation
         elif client.token_endpoint_auth_method == "none":
             request_client_secret = None
         else:
-            raise AuthenticationError(f"Unsupported auth method: {client.token_endpoint_auth_method}")
+            raise AuthenticationError(  # pragma: no cover
+                f"Unsupported auth method: {client.token_endpoint_auth_method}"
+            )
 
         # If client from the store expects a secret, validate that the request provides
         # that secret

tests/client/test_auth.py

@@ -601,6 +601,39 @@ async def test_register_client_skip_if_registered(self, oauth_provider: OAuthCli
         request = await oauth_provider._register_client()
         assert request is None
 
+    @pytest.mark.anyio
+    async def test_register_client_explicit_auth_method(self, mock_storage: MockTokenStorage):
+        """Test that explicitly set token_endpoint_auth_method is used without auto-selection."""
+
+        async def redirect_handler(url: str) -> None:
+            pass  # pragma: no cover
+
+        async def callback_handler() -> tuple[str, str | None]:
+            return "test_auth_code", "test_state"  # pragma: no cover
+
+        # Create client metadata with explicit auth method
+        explicit_metadata = OAuthClientMetadata(
+            client_name="Test Client",
+            client_uri=AnyHttpUrl("https://example.com"),
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+            scope="read write",
+            token_endpoint_auth_method="client_secret_basic",
+        )
+        provider = OAuthClientProvider(
+            server_url="https://api.example.com/v1/mcp",
+            client_metadata=explicit_metadata,
+            storage=mock_storage,
+            redirect_handler=redirect_handler,
+            callback_handler=callback_handler,
+        )
+
+        request = await provider._register_client()
+        assert request is not None
+
+        body = json.loads(request.content)
+        # Should use the explicitly set method, not auto-select
+        assert body["token_endpoint_auth_method"] == "client_secret_basic"
+
     @pytest.mark.anyio
     async def test_register_client_none_auth_method_with_server_metadata(self, oauth_provider: OAuthClientProvider):
         """Test that token_endpoint_auth_method=None selects from server's supported methods."""
@@ -611,7 +644,7 @@ async def test_register_client_none_auth_method_with_server_metadata(self, oauth
             token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
             token_endpoint_auth_methods_supported=["client_secret_post"],
         )
-        # Ensure client_metadata has None for token_endpoint_auth_method is None
+        # Ensure client_metadata has None for token_endpoint_auth_method
 
         request = await oauth_provider._register_client()
         assert request is not None

tests/server/fastmcp/auth/test_auth_integration.py

@@ -1117,6 +1117,189 @@ async def test_basic_auth_without_header_fails(
         assert error_response["error"] == "unauthorized_client"
         assert "Missing or invalid Basic authentication" in error_response["error_description"]
 
+    @pytest.mark.anyio
+    async def test_basic_auth_invalid_base64_fails(
+        self, test_client: httpx.AsyncClient, mock_oauth_provider: MockOAuthProvider, pkce_challenge: dict[str, str]
+    ):
+        """Test that invalid base64 in Basic auth header fails."""
+        client_metadata = {
+            "redirect_uris": ["https://client.example.com/callback"],
+            "client_name": "Basic Auth Client",
+            "token_endpoint_auth_method": "client_secret_basic",
+            "grant_types": ["authorization_code", "refresh_token"],
+        }
+
+        response = await test_client.post("/register", json=client_metadata)
+        assert response.status_code == 201
+        client_info = response.json()
+
+        auth_code = f"code_{int(time.time())}"
+        mock_oauth_provider.auth_codes[auth_code] = AuthorizationCode(
+            code=auth_code,
+            client_id=client_info["client_id"],
+            code_challenge=pkce_challenge["code_challenge"],
+            redirect_uri=AnyUrl("https://client.example.com/callback"),
+            redirect_uri_provided_explicitly=True,
+            scopes=["read", "write"],
+            expires_at=time.time() + 600,
+        )
+
+        # Send invalid base64
+        response = await test_client.post(
+            "/token",
+            headers={"Authorization": "Basic !!!invalid-base64!!!"},
+            data={
+                "grant_type": "authorization_code",
+                "client_id": client_info["client_id"],
+                "code": auth_code,
+                "code_verifier": pkce_challenge["code_verifier"],
+                "redirect_uri": "https://client.example.com/callback",
+            },
+        )
+        assert response.status_code == 401
+        error_response = response.json()
+        assert error_response["error"] == "unauthorized_client"
+        assert "Invalid Basic authentication header" in error_response["error_description"]
+
+    @pytest.mark.anyio
+    async def test_basic_auth_no_colon_fails(
+        self, test_client: httpx.AsyncClient, mock_oauth_provider: MockOAuthProvider, pkce_challenge: dict[str, str]
+    ):
+        """Test that Basic auth without colon separator fails."""
+        client_metadata = {
+            "redirect_uris": ["https://client.example.com/callback"],
+            "client_name": "Basic Auth Client",
+            "token_endpoint_auth_method": "client_secret_basic",
+            "grant_types": ["authorization_code", "refresh_token"],
+        }
+
+        response = await test_client.post("/register", json=client_metadata)
+        assert response.status_code == 201
+        client_info = response.json()
+
+        auth_code = f"code_{int(time.time())}"
+        mock_oauth_provider.auth_codes[auth_code] = AuthorizationCode(
+            code=auth_code,
+            client_id=client_info["client_id"],
+            code_challenge=pkce_challenge["code_challenge"],
+            redirect_uri=AnyUrl("https://client.example.com/callback"),
+            redirect_uri_provided_explicitly=True,
+            scopes=["read", "write"],
+            expires_at=time.time() + 600,
+        )
+
+        # Send base64 without colon (invalid format)
+        import base64
+
+        invalid_creds = base64.b64encode(b"no-colon-here").decode()
+        response = await test_client.post(
+            "/token",
+            headers={"Authorization": f"Basic {invalid_creds}"},
+            data={
+                "grant_type": "authorization_code",
+                "client_id": client_info["client_id"],
+                "code": auth_code,
+                "code_verifier": pkce_challenge["code_verifier"],
+                "redirect_uri": "https://client.example.com/callback",
+            },
+        )
+        assert response.status_code == 401
+        error_response = response.json()
+        assert error_response["error"] == "unauthorized_client"
+        assert "Invalid Basic authentication header" in error_response["error_description"]
+
+    @pytest.mark.anyio
+    async def test_basic_auth_client_id_mismatch_fails(
+        self, test_client: httpx.AsyncClient, mock_oauth_provider: MockOAuthProvider, pkce_challenge: dict[str, str]
+    ):
+        """Test that client_id mismatch between body and Basic auth fails."""
+        client_metadata = {
+            "redirect_uris": ["https://client.example.com/callback"],
+            "client_name": "Basic Auth Client",
+            "token_endpoint_auth_method": "client_secret_basic",
+            "grant_types": ["authorization_code", "refresh_token"],
+        }
+
+        response = await test_client.post("/register", json=client_metadata)
+        assert response.status_code == 201
+        client_info = response.json()
+
+        auth_code = f"code_{int(time.time())}"
+        mock_oauth_provider.auth_codes[auth_code] = AuthorizationCode(
+            code=auth_code,
+            client_id=client_info["client_id"],
+            code_challenge=pkce_challenge["code_challenge"],
+            redirect_uri=AnyUrl("https://client.example.com/callback"),
+            redirect_uri_provided_explicitly=True,
+            scopes=["read", "write"],
+            expires_at=time.time() + 600,
+        )
+
+        # Send different client_id in Basic auth header
+        import base64
+
+        wrong_creds = base64.b64encode(f"wrong-client-id:{client_info['client_secret']}".encode()).decode()
+        response = await test_client.post(
+            "/token",
+            headers={"Authorization": f"Basic {wrong_creds}"},
+            data={
+                "grant_type": "authorization_code",
+                "client_id": client_info["client_id"],  # Correct client_id in body
+                "code": auth_code,
+                "code_verifier": pkce_challenge["code_verifier"],
+                "redirect_uri": "https://client.example.com/callback",
+            },
+        )
+        assert response.status_code == 401
+        error_response = response.json()
+        assert error_response["error"] == "unauthorized_client"
+        assert "Client ID mismatch" in error_response["error_description"]
+
+    @pytest.mark.anyio
+    async def test_none_auth_method_public_client(
+        self, test_client: httpx.AsyncClient, mock_oauth_provider: MockOAuthProvider, pkce_challenge: dict[str, str]
+    ):
+        """Test that 'none' authentication method works for public clients."""
+        client_metadata = {
+            "redirect_uris": ["https://client.example.com/callback"],
+            "client_name": "Public Client",
+            "token_endpoint_auth_method": "none",
+            "grant_types": ["authorization_code", "refresh_token"],
+        }
+
+        response = await test_client.post("/register", json=client_metadata)
+        assert response.status_code == 201
+        client_info = response.json()
+        assert client_info["token_endpoint_auth_method"] == "none"
+        # Public clients should not have a client_secret
+        assert "client_secret" not in client_info or client_info.get("client_secret") is None
+
+        auth_code = f"code_{int(time.time())}"
+        mock_oauth_provider.auth_codes[auth_code] = AuthorizationCode(
+            code=auth_code,
+            client_id=client_info["client_id"],
+            code_challenge=pkce_challenge["code_challenge"],
+            redirect_uri=AnyUrl("https://client.example.com/callback"),
+            redirect_uri_provided_explicitly=True,
+            scopes=["read", "write"],
+            expires_at=time.time() + 600,
+        )
+
+        # Token request without any client secret
+        response = await test_client.post(
+            "/token",
+            data={
+                "grant_type": "authorization_code",
+                "client_id": client_info["client_id"],
+                "code": auth_code,
+                "code_verifier": pkce_challenge["code_verifier"],
+                "redirect_uri": "https://client.example.com/callback",
+            },
+        )
+        assert response.status_code == 200
+        token_response = response.json()
+        assert "access_token" in token_response
+
 
 class TestAuthorizeEndpointErrors:
     """Test error handling in the OAuth authorization endpoint."""