fix: restore test_883_middleware and exclude server helpers from coverage

- Resurrects tests/issues/test_883_middleware.py with AsyncClient fixes
- Marks server helper functions in test_streamable_http.py as no cover because subprocess coverage is not reliable in CI

Author: Ashutosh0x

tests/issues/test_883_middleware.py

@@ -0,0 +1,57 @@
+import pytest
+from starlette.applications import Starlette
+from starlette.middleware import Middleware
+from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
+from starlette.requests import Request
+from starlette.responses import Response
+from httpx import AsyncClient, ASGITransport
+
+from mcp.server.mcpserver import MCPServer
+from mcp.server.transport_security import TransportSecuritySettings
+
+
+class MockMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
+        return await call_next(request)
+
+
+@pytest.mark.anyio
+async def test_883_middleware_sse_no_assertion_error():
+    """Test that using MCP SSE with Starlette middleware doesn't cause double-response error."""
+    mcp_server = MCPServer("test-server")
+    transport_security = TransportSecuritySettings(enable_dns_rebinding_protection=False)
+    # Using host="0.0.0.0" avoids auto-protection triggering logic for localhost
+    sse_app = mcp_server.sse_app(transport_security=transport_security, host="0.0.0.0")
+
+    app = Starlette(middleware=[Middleware(MockMiddleware)])
+    # Mount at root to simplify test paths
+    app.mount("/", sse_app)
+
+    # Use ASGITransport to properly test the ASGI app stack
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://testserver") as client:
+        async with client.stream("GET", "/sse") as response:
+            assert response.status_code == 200
+            assert "text/event-stream" in response.headers["content-type"]
+            # Consume stream a bit or close immediately
+            pass
+
+
+@pytest.mark.anyio
+async def test_883_middleware_post_accepted():
+    """Test that POST messages work with middleware."""
+    mcp_server = MCPServer("test-server")
+    transport_security = TransportSecuritySettings(enable_dns_rebinding_protection=False)
+    sse_app = mcp_server.sse_app(transport_security=transport_security, host="0.0.0.0")
+
+    app = Starlette(middleware=[Middleware(MockMiddleware)])
+    app.mount("/", sse_app)
+
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://testserver") as client:
+        response = await client.post(
+            "/messages/?session_id=00000000000000000000000000000000",
+            json={"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}},
+        )
+        # 404 is expected here as we didn't establish a real session
+        assert response.status_code == 404

tests/shared/test_streamable_http.py

@@ -125,7 +125,7 @@ async def replay_events_after(  # pragma: no cover
 
 
 # Test server implementation that follows MCP protocol
-class ServerTest(Server):
+class ServerTest(Server):  # pragma: no cover
     def __init__(self):
         super().__init__(SERVER_NAME)
         self._lock = None  # Will be initialized in async context
@@ -383,11 +383,11 @@ async def handle_call_tool(name: str, args: dict[str, Any]) -> list[TextContent]
             return [TextContent(type="text", text=f"Called {name}")]
 
 
-def create_app(
+def create_app(  # pragma: no cover
     is_json_response_enabled: bool = False,
     event_store: EventStore | None = None,
     retry_interval: int | None = None,
-) -> Starlette:  # pragma: no cover
+) -> Starlette:
     """Create a Starlette application for testing using the session manager.
 
     Args:
@@ -422,12 +422,12 @@ def create_app(
     return app
 
 
-def run_server(
+def run_server(  # pragma: no cover
     port: int,
     is_json_response_enabled: bool = False,
     event_store: EventStore | None = None,
     retry_interval: int | None = None,
-) -> None:  # pragma: no cover
+) -> None:
     """Run the test server.
 
     Args:
@@ -957,8 +957,6 @@ def test_get_validation(basic_server: None, basic_server_url: str):
     assert "Not Acceptable" in response.text
 
 
-
-
 @pytest.fixture
 async def initialized_client_session(basic_server: None, basic_server_url: str):
     """Create initialized StreamableHTTP client session."""
@@ -1379,7 +1377,7 @@ async def sampling_callback(
 
 
 # Context-aware server implementation for testing request context propagation
-class ContextAwareServerTest(Server):
+class ContextAwareServerTest(Server):  # pragma: no cover
     def __init__(self):
         super().__init__("ContextAwareServer")
 
@@ -1439,7 +1437,7 @@ async def handle_call_tool(name: str, args: dict[str, Any]) -> list[TextContent]
 
 
 # Server runner for context-aware testing
-def run_context_aware_server(port: int):
+def run_context_aware_server(port: int):  # pragma: no cover
     """Run the context-aware test server."""
     server = ContextAwareServerTest()