coverage: ignore leftover test case issues

Author: maxisbey

src/mcp/client/streamable_http.py

@@ -242,7 +242,7 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None:
             event_source.response.raise_for_status()
             logger.debug("Resumption GET SSE connection established")
 
-            async for sse in event_source.aiter_sse():
+            async for sse in event_source.aiter_sse():  # pragma: no branch
                 is_complete = await self._handle_sse_event(
                     sse,
                     ctx.read_stream_writer,

src/mcp/shared/session.py

@@ -425,7 +425,7 @@ async def _receive_loop(self) -> None:
                 # This is expected when the client disconnects abruptly.
                 # Without this handler, the exception would propagate up and
                 # crash the server's task group.
-                logging.debug("Read stream closed by client")
+                logging.debug("Read stream closed by client")  # pragma: no cover
             except Exception as e:  # pragma: no cover
                 # Other exceptions are not expected and should be logged. We purposefully
                 # catch all exceptions here to avoid crashing the server.

tests/client/test_stdio.py

@@ -303,7 +303,7 @@ async def test_basic_child_process_cleanup(self):
             assert os.path.exists(parent_marker), "Parent process didn't start"
 
             # Verify child is writing
-            if os.path.exists(marker_file):
+            if os.path.exists(marker_file):  # pragma: no branch
                 initial_size = os.path.getsize(marker_file)
                 await anyio.sleep(0.3)
                 size_after_wait = os.path.getsize(marker_file)
@@ -318,7 +318,7 @@ async def test_basic_child_process_cleanup(self):
 
             # Verify processes stopped
             await anyio.sleep(0.5)
-            if os.path.exists(marker_file):
+            if os.path.exists(marker_file):  # pragma: no branch
                 size_after_cleanup = os.path.getsize(marker_file)
                 await anyio.sleep(0.5)
                 final_size = os.path.getsize(marker_file)
@@ -335,7 +335,7 @@ async def test_basic_child_process_cleanup(self):
             for f in [marker_file, parent_marker]:
                 try:
                     os.unlink(f)
-                except OSError:
+                except OSError:  # pragma: no cover
                     pass
 
     @pytest.mark.anyio
@@ -406,7 +406,7 @@ async def test_nested_process_tree(self):
 
             # Verify all are writing
             for file_path, name in [(parent_file, "parent"), (child_file, "child"), (grandchild_file, "grandchild")]:
-                if os.path.exists(file_path):
+                if os.path.exists(file_path):  # pragma: no branch
                     initial_size = os.path.getsize(file_path)
                     await anyio.sleep(0.3)
                     new_size = os.path.getsize(file_path)
@@ -420,7 +420,7 @@ async def test_nested_process_tree(self):
             # Verify all stopped
             await anyio.sleep(0.5)
             for file_path, name in [(parent_file, "parent"), (child_file, "child"), (grandchild_file, "grandchild")]:
-                if os.path.exists(file_path):
+                if os.path.exists(file_path):  # pragma: no branch
                     size1 = os.path.getsize(file_path)
                     await anyio.sleep(0.3)
                     size2 = os.path.getsize(file_path)
@@ -433,7 +433,7 @@ async def test_nested_process_tree(self):
             for f in [parent_file, child_file, grandchild_file]:
                 try:
                     os.unlink(f)
-                except OSError:
+                except OSError:  # pragma: no cover
                     pass
 
     @pytest.mark.anyio
@@ -487,7 +487,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 cover
                 size1 = os.path.getsize(marker_file)
                 await anyio.sleep(0.3)
                 size2 = os.path.getsize(marker_file)
@@ -500,7 +500,7 @@ def handle_term(sig, frame):
 
             # Verify child stopped
             await anyio.sleep(0.5)
-            if os.path.exists(marker_file):
+            if os.path.exists(marker_file):  # pragma: no branch
                 size3 = os.path.getsize(marker_file)
                 await anyio.sleep(0.3)
                 size4 = os.path.getsize(marker_file)
@@ -512,7 +512,7 @@ def handle_term(sig, frame):
             # Clean up marker file
             try:
                 os.unlink(marker_file)
-            except OSError:
+            except OSError:  # pragma: no cover
                 pass
 
 
@@ -560,7 +560,7 @@ async def test_stdio_client_graceful_stdin_exit():
         pytest.fail(
             "stdio_client cleanup timed out after 5.0 seconds. "
             "Process should have exited gracefully when stdin was closed."
-        )
+        )  # pragma: no cover
 
     end_time = time.time()
     elapsed = end_time - start_time
@@ -619,7 +619,7 @@ def sigterm_handler(signum, frame):
         pytest.fail(
             "stdio_client cleanup timed out after 7.0 seconds. "
             "Process should have been terminated via SIGTERM escalation."
-        )
+        )  # pragma: no cover
 
     end_time = time.time()
     elapsed = end_time - start_time

tests/server/fastmcp/resources/test_file_resources.py

@@ -102,7 +102,7 @@ async def test_missing_file_error(self, temp_file: Path):
 
     @pytest.mark.skipif(os.name == "nt", reason="File permissions behave differently on Windows")
     @pytest.mark.anyio
-    async def test_permission_error(self, temp_file: Path):
+    async def test_permission_error(self, temp_file: Path):  # pragma: no cover
         """Test reading a file without permissions."""
         temp_file.chmod(0o000)  # Remove all permissions
         try: