Merge pull request #42 from sacha-development-stuff/codex/fix-coverage-failure-in-tests-pz6y8h

Add tests to cover OAuth token flows

Author: SoldierSacha

tests/unit/client/test_oauth2_providers.py

@@ -705,6 +705,57 @@ async def post(self, url: str, *, data: dict[str, str], headers: dict[str, str])
     assert "resource" not in recorded_client.last_data
 
 
+@pytest.mark.anyio
+async def test_token_exchange_request_token_skips_client_error_and_omits_scope(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    storage = InMemoryStorage()
+    client_metadata = OAuthClientMetadata(redirect_uris=_redirect_uris())
+
+    subject_supplier = AsyncMock(return_value="subject-token")
+
+    provider = TokenExchangeProvider(
+        "https://api.example.com/service",
+        client_metadata,
+        storage,
+        subject_token_supplier=subject_supplier,
+    )
+
+    metadata_without_scopes = _metadata_json()
+    metadata_without_scopes.pop("scopes_supported", None)
+
+    metadata_responses = [
+        _make_response(404),
+        _make_response(200, json_data=metadata_without_scopes),
+    ]
+    registration_response = _make_response(200, json_data=_registration_json())
+
+    class RecordingAsyncClient(DummyAsyncClient):
+        def __init__(self) -> None:
+            super().__init__(post_responses=[_make_response(200, json_data=_token_json())])
+            self.last_data: dict[str, str] | None = None
+
+        async def post(
+            self, url: str, *, data: dict[str, str], headers: dict[str, str]
+        ) -> httpx.Response:
+            self.last_data = data
+            return await super().post(url, data=data, headers=headers)
+
+    clients: list[DummyAsyncClient] = [
+        DummyAsyncClient(send_responses=metadata_responses),
+        DummyAsyncClient(send_responses=[registration_response]),
+        RecordingAsyncClient(),
+    ]
+    monkeypatch.setattr("mcp.client.auth.oauth2.httpx.AsyncClient", AsyncClientFactory(clients))
+
+    await provider._request_token()
+
+    recorded_client = cast(RecordingAsyncClient, clients[-1])
+    assert recorded_client.last_data is not None
+    assert "scope" not in recorded_client.last_data
+    assert provider.client_metadata.scope is None
+
+
 @pytest.mark.anyio
 async def test_token_exchange_request_token_raises_on_failure(monkeypatch: pytest.MonkeyPatch) -> None:
     storage = InMemoryStorage()

tests/unit/server/auth/test_token_handler.py

@@ -52,6 +52,15 @@ async def exchange_client_credentials(self, client_info: object, scopes: list[st
         raise TokenError(error="invalid_client", error_description="bad credentials")
 
 
+class ClientCredentialsProviderSuccess:
+    def __init__(self) -> None:
+        self.last_scopes: list[str] | None = None
+
+    async def exchange_client_credentials(self, client_info: object, scopes: list[str]) -> OAuthToken:
+        self.last_scopes = scopes
+        return OAuthToken(access_token="client-token")
+
+
 class TokenExchangeProviderStub:
     def __init__(self) -> None:
         self.last_call: dict[str, Any] | None = None
@@ -155,6 +164,67 @@ async def test_handle_client_credentials_returns_token_error() -> None:
     assert result.error_description == "bad credentials"
 
 
+@pytest.mark.anyio
+async def test_handle_route_authorization_code_branch() -> None:
+    code_verifier = "a" * 64
+    digest = hashlib.sha256(code_verifier.encode()).digest()
+    code_challenge = base64.urlsafe_b64encode(digest).decode().rstrip("=")
+
+    provider = AuthorizationCodeProvider(expected_code="auth-code", code_challenge=code_challenge)
+    client_info = OAuthClientInformationFull(
+        client_id="client",
+        grant_types=["authorization_code"],
+        scope="alpha",
+    )
+    handler = TokenHandler(
+        provider=cast(OAuthAuthorizationServerProvider[Any, Any, Any], provider),
+        client_authenticator=cast(ClientAuthenticator, DummyAuthenticator(client_info)),
+    )
+
+    request_data = {
+        "grant_type": "authorization_code",
+        "code": "auth-code",
+        "redirect_uri": None,
+        "client_id": "client",
+        "client_secret": "secret",
+        "code_verifier": code_verifier,
+    }
+
+    response = await handler.handle(cast(Request, DummyRequest(request_data)))
+
+    assert response.status_code == 200
+    payload = json.loads(bytes(response.body).decode())
+    assert payload["access_token"] == "auth-token"
+
+
+@pytest.mark.anyio
+async def test_handle_route_client_credentials_branch() -> None:
+    provider = ClientCredentialsProviderSuccess()
+    client_info = OAuthClientInformationFull(
+        client_id="client",
+        grant_types=["client_credentials"],
+        scope="alpha beta",
+    )
+    handler = TokenHandler(
+        provider=cast(OAuthAuthorizationServerProvider[Any, Any, Any], provider),
+        client_authenticator=cast(ClientAuthenticator, DummyAuthenticator(client_info)),
+    )
+
+    request_data = {
+        "grant_type": "client_credentials",
+        "scope": "beta",
+        "client_id": "client",
+        "client_secret": "secret",
+    }
+
+    response = await handler.handle(cast(Request, DummyRequest(request_data)))
+
+    assert response.status_code == 200
+    payload = json.loads(bytes(response.body).decode())
+    assert payload["access_token"] == "client-token"
+    assert provider.last_scopes == ["beta"]
+
+
 @pytest.mark.anyio
 async def test_handle_route_refresh_token_branch() -> None:
     provider = RefreshTokenProvider()