Ignore coverage on test lines that are designed not to be executed

Author: LucaButBoring

tests/client/test_session_tasks.py

@@ -92,7 +92,7 @@ async def test_client_get_task_not_found():
                     # Try to get non-existent task
                     try:
                         await client_session.get_task("non-existent")
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         assert "Task not found" in str(e) or str(types.INVALID_PARAMS) in str(e)
             finally:
@@ -182,7 +182,7 @@ async def test_client_get_task_result_not_completed():
                     # Try to get result
                     try:
                         await client_session.get_task_result(task_id, types.ServerResult)
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         assert "not 'completed'" in str(e) or str(types.INVALID_PARAMS) in str(e)
             finally:
@@ -301,7 +301,7 @@ async def test_client_list_tasks_with_cursor():
                     # List tasks with invalid cursor should raise error
                     try:
                         await client_session.list_tasks(cursor="invalid-cursor")
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         assert "Invalid cursor" in str(e) or str(types.INVALID_PARAMS) in str(e)
             finally:
@@ -389,7 +389,7 @@ async def test_client_delete_task_not_found():
                     # Try to delete non-existent task
                     try:
                         await client_session.delete_task("non-existent")
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         assert "Failed to delete task" in str(e) or str(types.INVALID_PARAMS) in str(e)
             finally:

tests/server/test_session_tasks.py

@@ -102,7 +102,7 @@ async def test_get_task_not_found():
                             ),
                             types.GetTaskResult,
                         )
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         # Should get an error
                         assert "Task not found" in str(e) or str(types.INVALID_PARAMS) in str(e)
@@ -203,7 +203,7 @@ async def test_get_task_payload_not_completed():
                             ),
                             types.ServerResult,
                         )
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         # Should get an error about task not being completed
                         assert "not 'completed'" in str(e) or str(types.INVALID_PARAMS) in str(e)
@@ -247,7 +247,7 @@ async def test_get_task_payload_not_found():
                             ),
                             types.ServerResult,
                         )
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         # Should get an error about task not found
                         assert "Task not found" in str(e) or str(types.INVALID_PARAMS) in str(e)
@@ -431,7 +431,7 @@ async def test_delete_task_not_found():
                             ),
                             types.EmptyResult,
                         )
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         # Should get an error
                         assert "Failed to delete task" in str(e) or str(types.INVALID_PARAMS) in str(e)
@@ -478,7 +478,7 @@ async def test_get_task_without_capability():
                             types.ClientRequest(types.GetTaskRequest(params=types.GetTaskParams(taskId=task_id))),
                             types.GetTaskResult,
                         )
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         assert "not announced tasks capability" in str(e) or str(types.INVALID_REQUEST) in str(e)
             finally:
@@ -520,7 +520,7 @@ async def test_get_task_payload_without_capability():
                             ),
                             types.ServerResult,
                         )
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         assert "not announced tasks capability" in str(e) or str(types.INVALID_REQUEST) in str(e)
             finally:
@@ -560,7 +560,7 @@ async def test_list_tasks_without_capability():
                             types.ClientRequest(types.ListTasksRequest()),
                             types.ListTasksResult,
                         )
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         assert "not announced tasks capability" in str(e) or str(types.INVALID_REQUEST) in str(e)
             finally:
@@ -602,7 +602,7 @@ async def test_delete_task_without_capability():
                             ),
                             types.EmptyResult,
                         )
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         assert "not announced tasks capability" in str(e) or str(types.INVALID_REQUEST) in str(e)
             finally:
@@ -991,7 +991,7 @@ async def test_get_task_without_task_store():
                             types.ClientRequest(types.GetTaskRequest(params=types.GetTaskParams(taskId="test-task"))),
                             types.GetTaskResult,
                         )
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         assert "Task store not configured" in str(e) or str(types.INVALID_REQUEST) in str(e)
             finally:
@@ -1033,7 +1033,7 @@ async def test_get_task_payload_without_task_store():
                             ),
                             types.ServerResult,
                         )
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         assert "Task store not configured" in str(e) or str(types.INVALID_REQUEST) in str(e)
             finally:
@@ -1073,7 +1073,7 @@ async def test_list_tasks_without_task_store():
                             types.ClientRequest(types.ListTasksRequest()),
                             types.ListTasksResult,
                         )
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         assert "Task store not configured" in str(e) or str(types.INVALID_REQUEST) in str(e)
             finally:
@@ -1115,7 +1115,7 @@ async def test_delete_task_without_task_store():
                             ),
                             types.EmptyResult,
                         )
-                        assert False, "Should have raised McpError"
+                        assert False, "Should have raised McpError"  # pragma: no cover
                     except Exception as e:
                         assert "Task store not configured" in str(e) or str(types.INVALID_REQUEST) in str(e)
             finally:

tests/shared/test_pending_request.py

@@ -9,6 +9,11 @@
 from mcp.shared.request import DEFAULT_POLLING_INTERVAL, PendingRequest, TaskHandlerOptions
 from mcp.types import CallToolResult, GetTaskResult, TextContent
 
+# Mark all tests in this module to ignore memory stream cleanup warnings
+pytestmark = pytest.mark.filterwarnings(
+    "ignore:Exception ignored.*MemoryObject.*Stream:pytest.PytestUnraisableExceptionWarning"
+)
+
 
 @pytest.fixture
 def mock_session() -> MagicMock:
@@ -95,7 +100,7 @@ async def task_created():
 
         async def never_completes():
             await asyncio.Future()  # Never completes
-            return sample_result
+            return sample_result  # pragma: no cover
 
         # Set up task status progression
         mock_session.get_task.side_effect = [
@@ -135,7 +140,7 @@ async def task_created():
 
         async def never_completes():
             await asyncio.Future()  # Never completes
-            return sample_result
+            return sample_result  # pragma: no cover
 
         on_task_created = AsyncMock()
         on_task_status = AsyncMock()
@@ -185,7 +190,7 @@ async def task_created():
 
         async def never_completes():
             await asyncio.Future()  # Never completes
-            return sample_result
+            return sample_result  # pragma: no cover
 
         # Track polling timestamps
         poll_times: list[float] = []
@@ -229,7 +234,7 @@ async def task_created():
 
         async def never_completes():
             await asyncio.Future()  # Never completes
-            return sample_result
+            return sample_result  # pragma: no cover
 
         poll_times: list[float] = []
 
@@ -279,7 +284,7 @@ async def get_result():
         # Set up task polling to be slow
         async def slow_get_task(task_id: str):
             await asyncio.sleep(0.2)
-            return GetTaskResult(taskId=task_id, status="submitted", pollInterval=100)
+            return GetTaskResult(taskId=task_id, status="submitted", pollInterval=100)  # pragma: no cover
 
         mock_session.get_task.side_effect = slow_get_task
 
@@ -316,7 +321,7 @@ async def task_created():
 
         async def never_completes():
             await asyncio.Future()  # Never completes
-            return sample_result
+            return sample_result  # pragma: no cover
 
         # Set up task polling to complete quickly
         mock_session.get_task.return_value = GetTaskResult(taskId="task-6", status="completed", pollInterval=100)
@@ -449,12 +454,12 @@ async def task_created():
 
         async def never_completes():
             await asyncio.Future()  # Never completes
-            return sample_result
+            return sample_result  # pragma: no cover
 
         # Set up task polling to never complete
         async def never_complete_get_task(task_id: str):
             await asyncio.sleep(10)
-            return GetTaskResult(taskId=task_id, status="submitted", pollInterval=100)
+            return GetTaskResult(taskId=task_id, status="submitted", pollInterval=100)  # pragma: no cover
 
         mock_session.get_task.side_effect = never_complete_get_task
 
@@ -540,7 +545,7 @@ async def task_created():
 
         async def never_completes():
             await asyncio.Future()  # Never completes
-            return sample_result
+            return sample_result  # pragma: no cover
 
         mock_session.get_task.return_value = GetTaskResult(taskId="task-12", status="completed", pollInterval=100)
         mock_session.get_task_result.return_value = sample_result
@@ -569,7 +574,7 @@ async def task_created():
 
         async def never_completes():
             await asyncio.Future()  # Never completes
-            return sample_result
+            return sample_result  # pragma: no cover
 
         mock_session.get_task.side_effect = [
             GetTaskResult(taskId="task-13", status="submitted", pollInterval=50),
@@ -601,7 +606,7 @@ async def task_created():
 
         async def never_completes():
             await asyncio.Future()  # Never completes
-            return sample_result
+            return sample_result  # pragma: no cover
 
         mock_session.get_task.side_effect = [
             GetTaskResult(taskId="task-14", status="working", pollInterval=50),

tests/test_task_capabilities.py

@@ -136,7 +136,7 @@ def test_tasks_capability_includes_tools_when_available(self):
         # Register tool handler
         @server.call_tool()
         async def my_tool(arguments: dict[str, Any]) -> list[types.TextContent]:
-            return [types.TextContent(type="text", text="test")]
+            return [types.TextContent(type="text", text="test")]  # pragma: no cover
 
         caps = server.get_capabilities(NotificationOptions(), {})
         assert caps.tasks is not None
@@ -152,7 +152,7 @@ def test_tasks_capability_includes_resources_when_available(self):
         # Register resource handler
         @server.read_resource()
         async def read_resource(uri: AnyUrl) -> str:
-            return "test"
+            return "test"  # pragma: no cover
 
         caps = server.get_capabilities(NotificationOptions(), {})
         assert caps.tasks is not None
@@ -168,7 +168,7 @@ def test_tasks_capability_includes_prompts_when_available(self):
         # Register prompt handler
         @server.get_prompt()
         async def get_prompt(name: str, arguments: dict[str, str] | None = None) -> types.GetPromptResult:
-            return types.GetPromptResult(
+            return types.GetPromptResult(  # pragma: no cover
                 messages=[types.PromptMessage(role="user", content=types.TextContent(type="text", text="test"))]
             )