backport: coverage — recognize 'lax no cover'; pragma harness scaffolding + 3.11 dead-zone lines

- pyproject: add 'pragma: lax no cover' to exclude_lines (matches main)
- _connect.py: Protocol stub body, verifier-gate no-branch, SSE return-Response no-cover
- _harness.py: resource_server_url gate no-branch
- test_initialize/test_wire: tg.cancel_scope.cancel() lax-no-cover (cpython#106749, 3.11 only)
- test_sse: nested-async-with no-branch (3.11 arc)

All harness scaffolding; no port bugs. Local ./scripts/test → 100.00%.

Author: maxisbey

pyproject.toml

@@ -213,6 +213,7 @@ ignore_errors = true
 precision = 2
 exclude_lines = [
     "pragma: no cover",
+    "pragma: lax no cover",
     "if TYPE_CHECKING:",
     "@overload",
     "raise NotImplementedError",

tests/interaction/_connect.py

@@ -103,7 +103,7 @@ def __call__(
         logging_callback: LoggingFnT | None = None,
         message_handler: MessageHandlerFnT | None = None,
         client_info: Implementation | None = None,
-    ) -> AbstractAsyncContextManager[ClientSession]: ...
+    ) -> AbstractAsyncContextManager[ClientSession]: ...  # pragma: no cover
 
 
 @asynccontextmanager
@@ -188,7 +188,7 @@ def build_streamable_http_app(
 
     if auth is not None:
         required_scopes = auth.required_scopes or []
-        if verifier is not None:
+        if verifier is not None:  # pragma: no branch  — every auth-bearing caller supplies a provider/verifier
             middleware = [
                 Middleware(AuthenticationMiddleware, backend=BearerAuthBackend(verifier)),
                 Middleware(AuthContextMiddleware),
@@ -437,7 +437,8 @@ def build_sse_app(server: Server[Any] | FastMCP) -> tuple[Starlette, SseServerTr
     async def handle_sse(request: Request) -> Response:
         async with sse.connect_sse(request.scope, request.receive, request._send) as (read, write):  # type: ignore[reportPrivateUsage]
             await lowlevel.run(read, write, lowlevel.create_initialization_options())
-        return Response()
+        # under StreamingASGITransport the request is cancelled on close, so run() never returns
+        return Response()  # pragma: no cover
 
     app = Starlette(
         routes=[

tests/interaction/auth/_harness.py

@@ -477,7 +477,7 @@ async def connect_with_oauth(
     else:
         routes.append(Route("/mcp", endpoint=asgi))
 
-    if settings.resource_server_url:
+    if settings.resource_server_url:  # pragma: no branch  — auth_settings() always sets this
         routes.extend(
             create_protected_resource_routes(
                 resource_url=settings.resource_server_url,

tests/interaction/lowlevel/test_initialize.py

@@ -70,7 +70,7 @@ async def _initialize(server: Server[Any]) -> InitializeResult:
             async with ClientSession(client_read, client_write) as session:
                 with anyio.fail_after(5):
                     initialize_result = await session.initialize()
-            tg.cancel_scope.cancel()
+            tg.cancel_scope.cancel()  # pragma: lax no cover  — python/cpython#106749 (3.11 tracer dead-zone)
     return initialize_result
 
 
@@ -88,7 +88,7 @@ async def _bare_session(server: Server[Any]) -> AsyncIterator[ClientSession]:
             tg.start_soon(lambda: server.run(server_read, server_write, server.create_initialization_options()))
             async with ClientSession(client_read, client_write) as session:
                 yield session
-            tg.cancel_scope.cancel()
+            tg.cancel_scope.cancel()  # pragma: lax no cover  — python/cpython#106749 (3.11 tracer dead-zone)
 
 
 @requirement("lifecycle:initialize:basic")

tests/interaction/lowlevel/test_wire.py

@@ -83,7 +83,7 @@ async def _record(
                     await client.initialize()
                     yield client, recording
             finally:
-                tg.cancel_scope.cancel()
+                tg.cancel_scope.cancel()  # pragma: lax no cover  — python/cpython#106749 (3.11 tracer dead-zone)
 
 
 @requirement("protocol:request-id:unique")

tests/interaction/transports/test_sse.py

@@ -54,7 +54,7 @@ def httpx_client_factory(
         async with sse_client(
             f"{BASE_URL}/sse", httpx_client_factory=httpx_client_factory, on_session_created=captured_session_id.append
         ) as (read, write):
-            async with ClientSession(read, write) as client:
+            async with ClientSession(read, write) as client:  # pragma: no branch
                 await client.initialize()
                 assert len(captured_session_id) == 1
                 assert UUID(hex=captured_session_id[0]) in sse._read_stream_writers