Add coverage pragmas for SSE polling reconnection code

The SSE polling reconnection code paths run in a subprocess during testing,
making them difficult to cover with the main test process's coverage
instrumentation. Add pragma comments to exclude these from coverage
requirements:

- Client _attempt_sse_reconnection method and call site
- Server StreamableHTTPSessionManager.close_sse_stream method
- Test callback branch for empty data handling

Github-Issue:#1654

Author: felixweinberger

src/mcp/client/streamable_http.py

@@ -415,12 +415,12 @@ async def _handle_sse_response(
                 await ctx.read_stream_writer.send(e)
 
         # Auto-reconnect if stream ended without completion and we have priming event
-        if not is_complete and has_priming_event and last_event_id:
+        if not is_complete and has_priming_event and last_event_id:  # pragma: no cover
             await self._attempt_sse_reconnection(ctx, last_event_id, attempt)
 
         return has_priming_event, last_event_id
 
-    async def _attempt_sse_reconnection(
+    async def _attempt_sse_reconnection(  # pragma: no cover
         self,
         ctx: RequestContext,
         last_event_id: str,

src/mcp/server/streamable_http_manager.py

@@ -281,7 +281,9 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
             )
             await response(scope, receive, send)
 
-    async def close_sse_stream(self, session_id: str, request_id: str | int) -> bool:
+    async def close_sse_stream(  # pragma: no cover
+        self, session_id: str, request_id: str | int
+    ) -> bool:
         """Close an SSE stream for a specific request, triggering client reconnection.
 
         Use this to implement polling behavior during long-running operations.

tests/shared/test_streamable_http.py

@@ -1907,9 +1907,8 @@ async def test_streamablehttp_client_auto_reconnection(event_server: tuple[Simpl
 
     async def logging_callback(params: types.LoggingMessageNotificationParams) -> None:
         """Called when a log message notification is received from the server."""
-        data = params.data
-        if data:
-            notifications_received.append(str(data))
+        if params.data:  # pragma: no branch
+            notifications_received.append(str(params.data))
 
     # Configure client with reconnection options (fast delays for testing)
     reconnection_options = StreamableHTTPReconnectionOptions(