test(oauth): cover refresh success, failure, and a skipped re-check

The refresh re-check and the failure flag now sit outside the lock's
exit so branch coverage can see them. A peer that stores a valid token
while this caller waits does not start a second refresh. A non-200
refresh clears the token and sends the original request unauthenticated.

Author: Bartok9

src/mcp/client/auth/oauth2.py

@@ -640,14 +640,23 @@ async def _auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx2.Req
 
         if needs_refresh:
             async with self.context.refresh_lock:
-                refresh_request = None
+                # Re-check outside the inner `async with` so a peer refresh that
+                # landed while this caller waited is a normal branch, not one
+                # hidden in the lock's exit.
+                still_needs_refresh = False
                 async with self.context.lock:
-                    if self._needs_token_refresh():
+                    still_needs_refresh = self._needs_token_refresh()
+                refresh_request = None
+                if still_needs_refresh:
+                    async with self.context.lock:
                         refresh_request = await self._refresh_token()
                 if refresh_request is not None:
                     refresh_response = yield refresh_request
+                    refresh_ok = False
                     async with self.context.lock:
-                        if not await self._handle_refresh_response(refresh_response):
+                        refresh_ok = await self._handle_refresh_response(refresh_response)
+                    if not refresh_ok:
+                        async with self.context.lock:
                             self._initialized = False
 
         async with self.context.lock:

tests/client/test_auth.py

@@ -7,6 +7,7 @@
 from unittest import mock
 from urllib.parse import parse_qs, quote, unquote, urlparse
 
+import anyio
 import httpx2
 import pytest
 from inline_snapshot import Is, snapshot
@@ -3667,3 +3668,144 @@ async def echo_callback() -> AuthorizationCodeResult:
         await auth_flow.asend(httpx2.Response(200, request=final_req))
     except StopAsyncIteration:
         pass
+
+
+def _expired_refreshable(oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken) -> None:
+    """Token is present and refreshable, but past its expiry. Discovery is already done."""
+    oauth_provider._initialized = True
+    oauth_provider.context.current_tokens = valid_tokens
+    oauth_provider.context.token_expiry_time = time.time() - 30
+    oauth_provider.context.client_info = OAuthClientInformationFull(
+        client_id="test_client",
+        client_secret="test_secret",
+        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        token_endpoint_auth_method="client_secret_post",
+    )
+
+
+@pytest.mark.anyio
+async def test_expired_token_is_refreshed_then_the_original_request_is_sent(
+    oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken, mock_storage: MockTokenStorage
+):
+    """A successful refresh stores the new token and the original request goes out with it.
+
+    SDK behaviour: refresh runs outside ``context.lock``, and a 200 refresh response
+    leaves the provider initialized.
+    """
+    _expired_refreshable(oauth_provider, valid_tokens)
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    refresh_request = await auth_flow.__anext__()
+
+    assert refresh_request.method == "POST"
+    assert "grant_type=refresh_token" in refresh_request.content.decode()
+    refresh_response = httpx2.Response(
+        200,
+        json={"access_token": "refreshed-access", "token_type": "Bearer", "expires_in": 3600, "refresh_token": "kept"},
+        request=refresh_request,
+    )
+    primary = await auth_flow.asend(refresh_response)
+
+    assert primary.method == "GET"
+    assert primary.headers["Authorization"] == "Bearer refreshed-access"
+    assert oauth_provider._initialized is True
+    stored = await mock_storage.get_tokens()
+    assert stored is not None
+    assert stored.access_token == "refreshed-access"
+    try:
+        await auth_flow.asend(httpx2.Response(200, request=primary))
+    except StopAsyncIteration:
+        pass
+
+
+@pytest.mark.anyio
+async def test_failed_refresh_clears_the_token_and_sends_the_request_unauthenticated(
+    oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken
+):
+    """A non-200 refresh drops the stored token and the original request goes out without it.
+
+    SDK behaviour: refresh failure marks the provider uninitialized so the next call rediscovers.
+    """
+    _expired_refreshable(oauth_provider, valid_tokens)
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    refresh_request = await auth_flow.__anext__()
+    assert refresh_request.method == "POST"
+    primary = await auth_flow.asend(httpx2.Response(400, json={"error": "invalid_grant"}, request=refresh_request))
+
+    assert primary.method == "GET"
+    assert "Authorization" not in primary.headers
+    assert oauth_provider._initialized is False
+    assert oauth_provider.context.current_tokens is None
+    try:
+        await auth_flow.asend(httpx2.Response(200, request=primary))
+    except StopAsyncIteration:
+        pass
+
+
+@pytest.mark.anyio
+async def test_refresh_waiter_sends_the_token_stored_while_it_waited(
+    oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken
+):
+    """A caller that finds an expired token does not refresh if a peer stored a valid one while it waited.
+
+    The re-check is under ``refresh_lock``. This drives ``async_auth_flow`` and holds that lock
+    so the wait is real; the public client has no other way to observe the single-flight.
+    """
+    _expired_refreshable(oauth_provider, valid_tokens)
+
+    class HoldableLock:
+        """Stand-in for ``refresh_lock`` that signals when a waiter arrives."""
+
+        def __init__(self) -> None:
+            self._inner = anyio.Lock()
+            self.waiting = anyio.Event()
+            self.arm = False
+
+        async def __aenter__(self):
+            if self.arm:
+                self.waiting.set()
+            await self._inner.acquire()
+            return self
+
+        async def __aexit__(self, *_exc: object) -> None:
+            self._inner.release()
+
+        async def acquire(self) -> None:
+            await self._inner.acquire()
+
+        def release(self) -> None:
+            self._inner.release()
+
+        def locked(self) -> bool:
+            return self._inner.locked()
+
+    lock = HoldableLock()
+    oauth_provider.context.refresh_lock = lock
+    await lock.acquire()
+    lock.arm = True
+    sent: list[httpx2.Request] = []
+
+    async def drive() -> None:
+        auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+        request = await auth_flow.__anext__()
+        sent.append(request)
+        try:
+            await auth_flow.asend(httpx2.Response(200, request=request))
+        except StopAsyncIteration:
+            pass
+
+    try:
+        async with anyio.create_task_group() as task_group:
+            task_group.start_soon(drive)
+            with anyio.fail_after(5):
+                await lock.waiting.wait()
+            oauth_provider.context.token_expiry_time = time.time() + 3600
+            lock.release()
+    finally:
+        if lock.locked():
+            lock.release()
+
+    assert len(sent) == 1
+    assert sent[0].method == "GET"
+    assert sent[0].headers["Authorization"] == "Bearer test_access_token"