test: add end-to-end OAuth authorization tests with an in-process AS/RS harness

Author: maxisbey

src/mcp/client/auth/oauth2.py

@@ -360,10 +360,10 @@ async def _perform_authorization_code_grant(self) -> tuple[str, str]:
         auth_code, returned_state = await self.context.callback_handler()
 
         if returned_state is None or not secrets.compare_digest(returned_state, state):
-            raise OAuthFlowError(f"State parameter mismatch: {returned_state} != {state}")  # pragma: no cover
+            raise OAuthFlowError(f"State parameter mismatch: {returned_state} != {state}")
 
         if not auth_code:
-            raise OAuthFlowError("No authorization code received")  # pragma: no cover
+            raise OAuthFlowError("No authorization code received")
 
         # Return auth code and code verifier for token exchange
         return auth_code, pkce_params.code_verifier
@@ -452,7 +452,7 @@ async def _refresh_token(self) -> httpx.Request:
 
         return httpx.Request("POST", token_url, data=refresh_data, headers=headers)
 
-    async def _handle_refresh_response(self, response: httpx.Response) -> bool:  # pragma: no cover
+    async def _handle_refresh_response(self, response: httpx.Response) -> bool:
         """Handle token refresh response. Returns True if successful."""
         if response.status_code != 200:
             logger.warning(f"Token refresh failed: {response.status_code}")
@@ -468,12 +468,12 @@ async def _handle_refresh_response(self, response: httpx.Response) -> bool:  # p
             await self.context.storage.set_tokens(token_response)
 
             return True
-        except ValidationError:
+        except ValidationError:  # pragma: no cover
             logger.exception("Invalid refresh response")
             self.context.clear_tokens()
             return False
 
-    async def _initialize(self) -> None:  # pragma: no cover
+    async def _initialize(self) -> None:
         """Load stored tokens and client info."""
         self.context.current_tokens = await self.context.storage.get_tokens()
         self.context.client_info = await self.context.storage.get_client_info()
@@ -507,17 +507,17 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.
         """HTTPX auth flow integration."""
         async with self.context.lock:
             if not self._initialized:
-                await self._initialize()  # pragma: no cover
+                await self._initialize()
 
             # Capture protocol version from request headers
             self.context.protocol_version = request.headers.get(MCP_PROTOCOL_VERSION)
 
             if not self.context.is_token_valid() and self.context.can_refresh_token():
                 # Try to refresh token
-                refresh_request = await self._refresh_token()  # pragma: no cover
-                refresh_response = yield refresh_request  # pragma: no cover
+                refresh_request = await self._refresh_token()
+                refresh_response = yield refresh_request
 
-                if not await self._handle_refresh_response(refresh_response):  # pragma: no cover
+                if not await self._handle_refresh_response(refresh_response):
                     # Refresh failed, need full re-authentication
                     self._initialized = False
 
@@ -612,7 +612,7 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.
                     # Step 5: Perform authorization and complete token exchange
                     token_response = yield await self._perform_authorization()
                     await self._handle_token_response(token_response)
-                except Exception:  # pragma: no cover
+                except Exception:
                     logger.exception("OAuth flow error")
                     raise
 

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

@@ -117,7 +117,7 @@ async def error_response(
                     pass
 
             # the error response MUST contain the state specified by the client, if any
-            if state is None:  # pragma: no cover
+            if state is None:
                 # make last-ditch effort to load state
                 state = best_effort_extract_string("state", params)
 

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

@@ -95,7 +95,7 @@ async def _send_auth_error(self, send: Send, status_code: int, error: str, descr
         """Send an authentication error response with WWW-Authenticate header."""
         # Build WWW-Authenticate header value
         www_auth_parts = [f'error="{error}"', f'error_description="{description}"']
-        if self.resource_metadata_url:  # pragma: no cover
+        if self.resource_metadata_url:
             www_auth_parts.append(f'resource_metadata="{self.resource_metadata_url}"')
 
         www_authenticate = f"Bearer {', '.join(www_auth_parts)}"

src/mcp/server/lowlevel/server.py

@@ -603,7 +603,7 @@ def streamable_http_app(
         required_scopes: list[str] = []
 
         # Set up auth if configured
-        if auth:  # pragma: no cover
+        if auth:
             required_scopes = auth.required_scopes or []
 
             # Add auth middleware if token verifier is available
@@ -629,10 +629,10 @@ def streamable_http_app(
                 )
 
         # Set up routes with or without auth
-        if token_verifier:  # pragma: no cover
+        if token_verifier:
             # Determine resource metadata URL
             resource_metadata_url = None
-            if auth and auth.resource_server_url:
+            if auth and auth.resource_server_url:  # pragma: no branch
                 # Build compliant metadata URL for WWW-Authenticate header
                 resource_metadata_url = build_resource_metadata_url(auth.resource_server_url)
 
@@ -652,7 +652,7 @@ def streamable_http_app(
             )
 
         # Add protected resource metadata endpoint if configured as RS
-        if auth and auth.resource_server_url:  # pragma: no cover
+        if auth and auth.resource_server_url:
             routes.extend(
                 create_protected_resource_routes(
                     resource_url=auth.resource_server_url,

src/mcp/shared/auth.py

@@ -93,7 +93,7 @@ def validate_scope(self, requested_scope: str | None) -> list[str] | None:
         for scope in requested_scopes:
             if scope not in allowed_scopes:  # pragma: no branch
                 raise InvalidScopeError(f"Client was not registered with scope {scope}")
-        return requested_scopes  # pragma: no cover
+        return requested_scopes
 
     def validate_redirect_uri(self, redirect_uri: AnyUrl | None) -> AnyUrl:
         if redirect_uri is not None:

src/mcp/shared/session.py

@@ -451,7 +451,7 @@ async def _handle_session_message(message: SessionMessage) -> None:
                     try:
                         await stream.send(JSONRPCError(jsonrpc="2.0", id=id, error=error))
                         await stream.aclose()
-                    except Exception:  # pragma: no cover
+                    except Exception:  # pragma: lax no cover
                         # Stream might already be closed
                         pass
                 self._response_streams.clear()

tests/interaction/_connect.py

@@ -11,7 +11,7 @@
 import warnings
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
-from typing import Protocol
+from typing import Any, Protocol
 
 import httpx
 from httpx_sse import ServerSentEvent, aconnect_sse
@@ -25,6 +25,8 @@
 from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamable_http_client
 from mcp.server import Server
+from mcp.server.auth.provider import OAuthAuthorizationServerProvider, TokenVerifier
+from mcp.server.auth.settings import AuthSettings
 from mcp.server.mcpserver import MCPServer
 from mcp.server.sse import SseServerTransport
 from mcp.server.streamable_http import EventStore
@@ -154,6 +156,9 @@ async def mounted_app(
     transport_security: TransportSecuritySettings | None = NO_DNS_REBINDING_PROTECTION,
     on_request: Callable[[httpx.Request], Awaitable[None]] | None = None,
     headers: dict[str, str] | None = None,
+    auth: AuthSettings | None = None,
+    token_verifier: TokenVerifier | None = None,
+    auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
 ) -> AsyncIterator[tuple[httpx.AsyncClient, StreamableHTTPSessionManager]]:
     """Mount the server's streamable HTTP app on the in-process bridge and yield an httpx client.
 
@@ -167,11 +172,15 @@ async def mounted_app(
     DNS-rebinding protection is disabled by default; pass explicit settings (or `None` for the
     localhost auto-enable behaviour) to test the protection itself.
     """
-    app = server.streamable_http_app(
+    lowlevel = server._lowlevel_server if isinstance(server, MCPServer) else server
+    app = lowlevel.streamable_http_app(
         stateless_http=stateless_http,
         event_store=event_store,
         retry_interval=retry_interval,
         transport_security=transport_security,
+        auth=auth,
+        token_verifier=token_verifier,
+        auth_server_provider=auth_server_provider,
     )
     event_hooks = {"request": [on_request]} if on_request is not None else None
     async with server.session_manager.run():