fix: restore pragmas for lines not covered in CI

Restore # pragma: no cover and # pragma: no branch on lines that are
not actually covered in CI (platform-dependent, branch coverage, or
unreachable code paths).

Author: maxisbey

tests/client/auth/extensions/test_client_credentials.py

@@ -26,7 +26,7 @@ def __init__(self):
     async def get_tokens(self) -> OAuthToken | None:
         return self._tokens
 
-    async def set_tokens(self, tokens: OAuthToken) -> None:
+    async def set_tokens(self, tokens: OAuthToken) -> None:  # pragma: no cover
         self._tokens = tokens
 
     async def get_client_info(self) -> OAuthClientInformationFull | None:  # pragma: no cover

tests/client/test_stdio.py

@@ -481,7 +481,7 @@ def handle_term(sig, frame):
             await anyio.sleep(0.5)
 
             # Verify child is writing
-            if os.path.exists(marker_file):
+            if os.path.exists(marker_file):  # pragma: no branch
                 size1 = os.path.getsize(marker_file)
                 await anyio.sleep(0.3)
                 size2 = os.path.getsize(marker_file)

tests/client/transports/test_memory.py

@@ -40,7 +40,7 @@ def greet(name: str) -> str:
         return f"Hello, {name}!"
 
     @server.resource("test://resource")
-    def test_resource() -> str:
+    def test_resource() -> str:  # pragma: no cover
         """A test resource."""
         return "Test content"
 

tests/issues/test_1027_win_unreachable_cleanup.py

@@ -102,7 +102,7 @@ def echo(text: str) -> str:
 
         # Give server a moment to complete cleanup
         with anyio.move_on_after(5.0):
-            while not Path(cleanup_marker).exists():
+            while not Path(cleanup_marker).exists():  # pragma: no cover
                 await anyio.sleep(0.1)
 
         # Verify cleanup marker was created - this works now that stdio_client
@@ -113,9 +113,9 @@ def echo(text: str) -> str:
     finally:
         # Clean up files
         for path in [server_script, startup_marker, cleanup_marker]:
-            try:
+            try:  # pragma: no cover
                 Path(path).unlink()
-            except FileNotFoundError:
+            except FileNotFoundError:  # pragma: no cover
                 pass
 
 
@@ -224,7 +224,7 @@ def echo(text: str) -> str:
 
         # Check if cleanup ran
         with anyio.move_on_after(5.0):
-            while not Path(cleanup_marker).exists():
+            while not Path(cleanup_marker).exists():  # pragma: no cover
                 await anyio.sleep(0.1)
 
         # Verify the cleanup ran - stdin closure enables graceful shutdown
@@ -234,7 +234,7 @@ def echo(text: str) -> str:
     finally:
         # Clean up files
         for path in [server_script, startup_marker, cleanup_marker]:
-            try:
+            try:  # pragma: no cover
                 Path(path).unlink()
-            except FileNotFoundError:
+            except FileNotFoundError:  # pragma: no cover
                 pass

tests/server/fastmcp/auth/test_auth_integration.py

@@ -172,7 +172,7 @@ async def load_access_token(self, token: str) -> AccessToken | None:
 
     async def revoke_token(self, token: AccessToken | RefreshToken) -> None:
         match token:
-            case RefreshToken():
+            case RefreshToken():  # pragma: no cover
                 # Remove the refresh token
                 del self.refresh_tokens[token.token]
 

tests/server/fastmcp/resources/test_file_resources.py

@@ -18,9 +18,9 @@ def temp_file():
         f.write(content)
         path = Path(f.name).resolve()
     yield path
-    try:
+    try:  # pragma: no cover
         path.unlink()
-    except FileNotFoundError:
+    except FileNotFoundError:  # pragma: no cover
         pass  # File was already deleted by the test
 
 

tests/server/fastmcp/resources/test_resource_manager.py

@@ -18,9 +18,9 @@ def temp_file():
         f.write(content)
         path = Path(f.name).resolve()
     yield path
-    try:
+    try:  # pragma: no cover
         path.unlink()
-    except FileNotFoundError:
+    except FileNotFoundError:  # pragma: no cover
         pass  # File was already deleted by the test
 
 

tests/server/fastmcp/test_elicitation.py

@@ -68,7 +68,7 @@ async def test_stdio_elicitation():
     async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams):
         if params.message == "Tool wants to ask: What is your name?":
             return ElicitResult(action="accept", content={"answer": "Test User"})
-        else:
+        else:  # pragma: no cover
             raise ValueError(f"Unexpected elicitation message: {params.message}")
 
     await call_tool_and_assert(
@@ -100,7 +100,7 @@ def create_validation_tool(name: str, schema_class: type[BaseModel]):
         async def tool(ctx: Context[ServerSession, None]) -> str:
             try:
                 await ctx.elicit(message="This should fail validation", schema=schema_class)
-                return "Should not reach here"
+                return "Should not reach here"  # pragma: no cover
             except TypeError as e:
                 return f"Validation failed as expected: {str(e)}"
 
@@ -191,7 +191,7 @@ class InvalidOptionalSchema(BaseModel):
     async def invalid_optional_tool(ctx: Context[ServerSession, None]) -> str:
         try:
             await ctx.elicit(message="This should fail", schema=InvalidOptionalSchema)
-            return "Should not reach here"
+            return "Should not reach here"  # pragma: no cover
         except TypeError as e:
             return f"Validation failed: {str(e)}"
 

tests/shared/test_streamable_http.py

@@ -79,7 +79,7 @@ def extract_protocol_version_from_sse(response: requests.Response) -> str:
         if line.startswith("data: "):
             init_data = json.loads(line[6:])
             return init_data["result"]["protocolVersion"]
-    raise ValueError("Could not extract protocol version from SSE response")
+    raise ValueError("Could not extract protocol version from SSE response")  # pragma: no cover
 
 
 # Simple in-memory event store for testing
@@ -97,7 +97,7 @@ async def store_event(self, stream_id: StreamId, message: types.JSONRPCMessage |
         self._events.append((stream_id, event_id, message))
         return event_id
 
-    async def replay_events_after(
+    async def replay_events_after(  # pragma: no cover
         self,
         last_event_id: EventId,
         send_callback: EventCallback,
@@ -1154,7 +1154,7 @@ async def test_streamable_http_client_session_termination(basic_server: None, ba
             assert len(tools.tools) == 10
 
     headers: dict[str, str] = {}
-    if captured_session_id:
+    if captured_session_id:  # pragma: no branch
         headers[MCP_SESSION_ID_HEADER] = captured_session_id
 
     async with create_mcp_http_client(headers=headers) as httpx_client:
@@ -1218,7 +1218,7 @@ async def mock_delete(self: httpx.AsyncClient, *args: Any, **kwargs: Any) -> htt
             assert len(tools.tools) == 10
 
     headers: dict[str, str] = {}
-    if captured_session_id:
+    if captured_session_id:  # pragma: no branch
         headers[MCP_SESSION_ID_HEADER] = captured_session_id
 
     async with create_mcp_http_client(headers=headers) as httpx_client:
@@ -1312,9 +1312,9 @@ async def run_tool():
 
     # Now resume the session with the same mcp-session-id and protocol version
     headers: dict[str, Any] = {}
-    if captured_session_id:
+    if captured_session_id:  # pragma: no branch
         headers[MCP_SESSION_ID_HEADER] = captured_session_id
-    if captured_protocol_version:
+    if captured_protocol_version:  # pragma: no branch
         headers[MCP_PROTOCOL_VERSION_HEADER] = captured_protocol_version
 
     async with create_mcp_http_client(headers=headers) as httpx_client:

tests/shared/test_ws.py

@@ -121,7 +121,7 @@ def server(server_port: int) -> Generator[None, None, None]:
     # Signal the server to stop
     proc.kill()
     proc.join(timeout=2)
-    if proc.is_alive():
+    if proc.is_alive():  # pragma: no cover
         print("server process failed to terminate")