coverage: add pragmas to reach 100% coverage on tests/

Author: maxisbey

tests/cli/test_utils.py

@@ -82,7 +82,7 @@ def test_get_npx_windows(monkeypatch: pytest.MonkeyPatch):
     def fake_run(cmd: list[str], **kw: Any) -> subprocess.CompletedProcess[bytes]:
         if cmd[0] in candidates:
             return subprocess.CompletedProcess(cmd, 0)
-        else:
+        else:  # pragma: no cover
             raise subprocess.CalledProcessError(1, cmd[0])
 
     monkeypatch.setattr(sys, "platform", "win32")

tests/client/auth/extensions/test_client_credentials.py

@@ -15,16 +15,16 @@ def __init__(self):
         self._tokens: OAuthToken | None = None
         self._client_info: OAuthClientInformationFull | None = None
 
-    async def get_tokens(self) -> OAuthToken | None:
+    async def get_tokens(self) -> OAuthToken | None:  # pragma: no cover
         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:
+    async def get_client_info(self) -> OAuthClientInformationFull | None:  # pragma: no cover
         return self._client_info
 
-    async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
+    async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:  # pragma: no cover
         self._client_info = client_info
 
 
@@ -45,11 +45,11 @@ def client_metadata():
 
 @pytest.fixture
 def rfc7523_oauth_provider(client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage):
-    async def redirect_handler(url: str) -> None:
+    async def redirect_handler(url: str) -> None:  # pragma: no cover
         """Mock redirect handler."""
         pass
 
-    async def callback_handler() -> tuple[str, str | None]:
+    async def callback_handler() -> tuple[str, str | None]:  # pragma: no cover
         """Mock callback handler."""
         return "test_auth_code", "test_state"
 

tests/client/conftest.py

@@ -40,23 +40,23 @@ def clear(self) -> None:
         self.client.sent_messages.clear()
         self.server.sent_messages.clear()
 
-    def get_client_requests(self, method: str | None = None) -> list[JSONRPCRequest]:
+    def get_client_requests(self, method: str | None = None) -> list[JSONRPCRequest]:  # pragma: no cover
         """Get client-sent requests, optionally filtered by method."""
         return [
             req.message.root
             for req in self.client.sent_messages
             if isinstance(req.message.root, JSONRPCRequest) and (method is None or req.message.root.method == method)
         ]
 
-    def get_server_requests(self, method: str | None = None) -> list[JSONRPCRequest]:
+    def get_server_requests(self, method: str | None = None) -> list[JSONRPCRequest]:  # pragma: no cover
         """Get server-sent requests, optionally filtered by method."""
-        return [
+        return [  # pragma: no cover
             req.message.root
             for req in self.server.sent_messages
             if isinstance(req.message.root, JSONRPCRequest) and (method is None or req.message.root.method == method)
         ]
 
-    def get_client_notifications(self, method: str | None = None) -> list[JSONRPCNotification]:
+    def get_client_notifications(self, method: str | None = None) -> list[JSONRPCNotification]:  # pragma: no cover
         """Get client-sent notifications, optionally filtered by method."""
         return [
             notif.message.root
@@ -65,7 +65,7 @@ def get_client_notifications(self, method: str | None = None) -> list[JSONRPCNot
             and (method is None or notif.message.root.method == method)
         ]
 
-    def get_server_notifications(self, method: str | None = None) -> list[JSONRPCNotification]:
+    def get_server_notifications(self, method: str | None = None) -> list[JSONRPCNotification]:  # pragma: no cover
         """Get server-sent notifications, optionally filtered by method."""
         return [
             notif.message.root

tests/client/test_auth.py

@@ -22,13 +22,13 @@ def __init__(self):
         self._client_info: OAuthClientInformationFull | None = None
 
     async def get_tokens(self) -> OAuthToken | None:
-        return self._tokens
+        return self._tokens  # pragma: no cover
 
     async def set_tokens(self, tokens: OAuthToken) -> None:
         self._tokens = tokens
 
     async def get_client_info(self) -> OAuthClientInformationFull | None:
-        return self._client_info
+        return self._client_info  # pragma: no cover
 
     async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
         self._client_info = client_info
@@ -64,11 +64,11 @@ def valid_tokens():
 def oauth_provider(client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage):
     async def redirect_handler(url: str) -> None:
         """Mock redirect handler."""
-        pass
+        pass  # pragma: no cover
 
     async def callback_handler() -> tuple[str, str | None]:
         """Mock callback handler."""
-        return "test_auth_code", "test_state"
+        return "test_auth_code", "test_state"  # pragma: no cover
 
     return OAuthClientProvider(
         server_url="https://api.example.com/v1/mcp",
@@ -247,10 +247,10 @@ async def test_build_protected_resource_discovery_urls(
         """Test protected resource metadata discovery URL building with fallback."""
 
         async def redirect_handler(url: str) -> None:
-            pass
+            pass  # pragma: no cover
 
         async def callback_handler() -> tuple[str, str | None]:
-            return "test_auth_code", "test_state"
+            return "test_auth_code", "test_state"  # pragma: no cover
 
         provider = OAuthClientProvider(
             server_url="https://api.example.com",
@@ -663,7 +663,7 @@ async def aread(self):
 
             @property
             def text(self):
-                if not self._aread_called:
+                if not self._aread_called:  # pragma: no cover
                     raise RuntimeError("Response.text accessed before response.aread()")
                 return self._text
 
@@ -846,10 +846,10 @@ async def test_auth_flow_no_unnecessary_retry_after_oauth(
         # In the buggy version, this would yield the request AGAIN unconditionally
         # In the fixed version, this should end the generator
         try:
-            await auth_flow.asend(response)  # extra request
-            request_yields += 1
-            # If we reach here, the bug is present
-            pytest.fail(
+            await auth_flow.asend(response)  # extra request  # pragma: no cover
+            request_yields += 1  # pragma: no cover
+            # If we reach here, the bug is present  # pragma: no cover
+            pytest.fail(  # pragma: no cover
                 f"Unnecessary retry detected! Request was yielded {request_yields} times. "
                 f"This indicates the retry logic bug that caused 2x performance degradation. "
                 f"The request should only be yielded once for successful responses."
@@ -949,7 +949,7 @@ async def mock_callback() -> tuple[str, str | None]:
         success_response = httpx.Response(200, request=final_request)
         try:
             await auth_flow.asend(success_response)
-            pytest.fail("Should have stopped after successful response")
+            pytest.fail("Should have stopped after successful response")  # pragma: no cover
         except StopAsyncIteration:
             pass  # Expected
 
@@ -1269,10 +1269,10 @@ def test_extract_field_from_www_auth_valid_cases(
         """Test extraction of various fields from valid WWW-Authenticate headers."""
 
         async def redirect_handler(url: str) -> None:
-            pass
+            pass  # pragma: no cover
 
         async def callback_handler() -> tuple[str, str | None]:
-            return "test_auth_code", "test_state"
+            return "test_auth_code", "test_state"  # pragma: no cover
 
         provider = OAuthClientProvider(
             server_url="https://api.example.com/v1/mcp",
@@ -1317,10 +1317,10 @@ def test_extract_field_from_www_auth_invalid_cases(
         """Test extraction returns None for invalid cases."""
 
         async def redirect_handler(url: str) -> None:
-            pass
+            pass  # pragma: no cover
 
         async def callback_handler() -> tuple[str, str | None]:
-            return "test_auth_code", "test_state"
+            return "test_auth_code", "test_state"  # pragma: no cover
 
         provider = OAuthClientProvider(
             server_url="https://api.example.com/v1/mcp",

tests/client/test_http_unicode.py

@@ -35,7 +35,7 @@
 }
 
 
-def run_unicode_server(port: int) -> None:
+def run_unicode_server(port: int) -> None:  # pragma: no cover
     """Run the Unicode test server in a separate process."""
     # Import inside the function since this runs in a separate process
     from collections.abc import AsyncGenerator
@@ -167,7 +167,7 @@ def running_unicode_server(unicode_server_port: int) -> Generator[str, None, Non
         # Clean up - try graceful termination first
         proc.terminate()
         proc.join(timeout=2)
-        if proc.is_alive():
+        if proc.is_alive():  # pragma: no cover
             proc.kill()
             proc.join(timeout=1)
 

tests/client/test_list_methods_cursor.py

@@ -19,27 +19,27 @@ async def full_featured_server():
     server = FastMCP("test")
 
     @server.tool(name="test_tool_1")
-    async def test_tool_1() -> str:
+    async def test_tool_1() -> str:  # pragma: no cover
         """First test tool"""
         return "Result 1"
 
     @server.tool(name="test_tool_2")
-    async def test_tool_2() -> str:
+    async def test_tool_2() -> str:  # pragma: no cover
         """Second test tool"""
         return "Result 2"
 
     @server.resource("resource://test/data")
-    async def test_resource() -> str:
+    async def test_resource() -> str:  # pragma: no cover
         """Test resource"""
         return "Test data"
 
     @server.prompt()
-    async def test_prompt(name: str) -> str:
+    async def test_prompt(name: str) -> str:  # pragma: no cover
         """Test prompt"""
         return f"Hello, {name}!"
 
     @server.resource("resource://test/{name}")
-    async def test_template(name: str) -> str:
+    async def test_template(name: str) -> str:  # pragma: no cover
         """Test resource template"""
         return f"Data for {name}"
 
@@ -209,7 +209,7 @@ async def test_list_tools_with_strict_server_validation():
     server = Server("strict_server")
 
     @server.list_tools()
-    async def handle_list_tools(request: ListToolsRequest) -> ListToolsResult:
+    async def handle_list_tools(request: ListToolsRequest) -> ListToolsResult:  # pragma: no cover
         """Strict handler that validates params field exists"""
 
         # Simulate strict server validation

tests/client/test_logging_callback.py

@@ -51,7 +51,7 @@ async def test_tool_with_log(
     async def message_handler(
         message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
     ) -> None:
-        if isinstance(message, Exception):
+        if isinstance(message, Exception):  # pragma: no cover
             raise message
 
     async with create_session(

tests/client/test_notification_response.py

@@ -24,7 +24,7 @@
 from tests.test_helpers import wait_for_server
 
 
-def create_non_sdk_server_app() -> Starlette:
+def create_non_sdk_server_app() -> Starlette:  # pragma: no cover
     """Create a minimal server that doesn't follow SDK conventions."""
 
     async def handle_mcp_request(request: Request) -> Response:
@@ -67,7 +67,7 @@ async def handle_mcp_request(request: Request) -> Response:
     return app
 
 
-def run_non_sdk_server(port: int) -> None:
+def run_non_sdk_server(port: int) -> None:  # pragma: no cover
     """Run the non-SDK server in a separate process."""
     app = create_non_sdk_server_app()
     config = uvicorn.Config(
@@ -95,9 +95,9 @@ def non_sdk_server(non_sdk_server_port: int) -> Generator[None, None, None]:
     proc.start()
 
     # Wait for server to be ready
-    try:
+    try:  # pragma: no cover
         wait_for_server(non_sdk_server_port, timeout=10.0)
-    except TimeoutError:
+    except TimeoutError:  # pragma: no cover
         proc.kill()
         proc.join(timeout=2)
         pytest.fail("Server failed to start within 10 seconds")
@@ -120,7 +120,7 @@ async def test_non_compliant_notification_response(non_sdk_server: None, non_sdk
     server_url = f"http://127.0.0.1:{non_sdk_server_port}/mcp"
     returned_exception = None
 
-    async def message_handler(
+    async def message_handler(  # pragma: no cover
         message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
     ):
         nonlocal returned_exception
@@ -141,5 +141,5 @@ async def message_handler(
                 ClientNotification(RootsListChangedNotification(method="notifications/roots/list_changed"))
             )
 
-    if returned_exception:
+    if returned_exception:  # pragma: no cover
         pytest.fail(f"Server encountered an exception: {returned_exception}")

tests/client/test_resource_cleanup.py

@@ -19,7 +19,7 @@ async def test_send_request_stream_cleanup():
 
     # Create a mock session with the minimal required functionality
     class TestSession(BaseSession[ClientRequest, ClientNotification, ClientResult, Any, Any]):
-        async def _send_response(self, request_id: RequestId, response: SendResultT | ErrorData) -> None:
+        async def _send_response(self, request_id: RequestId, response: SendResultT | ErrorData) -> None:  # pragma: no cover
             pass
 
     # Create streams

tests/client/test_session.py

@@ -82,7 +82,7 @@ async def mock_server():
             )
 
     # Create a message handler to catch exceptions
-    async def message_handler(
+    async def message_handler(  # pragma: no cover
         message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
     ) -> None:
         if isinstance(message, Exception):
@@ -426,7 +426,7 @@ async def test_client_capabilities_with_custom_callbacks():
 
     received_capabilities = None
 
-    async def custom_sampling_callback(
+    async def custom_sampling_callback(  # pragma: no cover
         context: RequestContext["ClientSession", Any],
         params: types.CreateMessageRequestParams,
     ) -> types.CreateMessageResult | types.ErrorData:
@@ -436,7 +436,7 @@ async def custom_sampling_callback(
             model="test-model",
         )
 
-    async def custom_list_roots_callback(
+    async def custom_list_roots_callback(  # pragma: no cover
         context: RequestContext["ClientSession", Any],
     ) -> types.ListRootsResult | types.ErrorData:
         return types.ListRootsResult(roots=[])