refactor: Replace mock-based test with real transport cleanup test

The previous test was self-fulfilling as it only verified that a mocked
method was called. The new test uses real StreamableHTTPServerTransport
instances and verifies that:
- The transport's _terminated flag is set to True
- The _request_streams dictionary is empty
- The test fails when _terminate_session() is not called

This provides a more robust regression test for PR #1116's memory leak fix.

Github-Issue:#1116

Author: felixweinberger

tests/server/test_streamable_http_manager.py

@@ -1,10 +1,11 @@
 """Tests for StreamableHTTPSessionManager."""
 
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, patch
 
 import anyio
 import pytest
 
+from mcp.server import streamable_http_manager
 from mcp.server.lowlevel import Server
 from mcp.server.streamable_http import MCP_SESSION_ID_HEADER
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
@@ -200,49 +201,29 @@ async def mock_receive():
 
 
 @pytest.mark.anyio
-async def test_stateless_requests_cleanup_transport_resources():
-    """Test that stateless requests properly clean up transport resources."""
-    from contextlib import asynccontextmanager
-    from unittest.mock import patch
-
-    app = Server("test-stateless-cleanup-server")
+async def test_stateless_requests_memory_cleanup():
+    """Test that stateless requests actually clean up resources using real transports."""
+    app = Server("test-stateless-real-cleanup")
     manager = StreamableHTTPSessionManager(app=app, stateless=True)
 
-    # Track created transports and their termination
+    # Track created transport instances
     created_transports = []
 
-    # Mock the transport class to track termination
-    with patch("mcp.server.streamable_http_manager.StreamableHTTPServerTransport") as MockTransport:
-        # Create a mock transport instance
-        def create_transport(*args, **kwargs):
-            transport = AsyncMock()
-            transport._terminated = False
-
-            # Track when terminate is called
-            async def mock_terminate():
-                transport._terminated = True
-
-            transport._terminate_session = mock_terminate
-            transport.handle_request = AsyncMock()
-
-            # Mock the connect context manager
-            @asynccontextmanager
-            async def mock_connect():
-                yield (AsyncMock(), AsyncMock())
-
-            transport.connect = mock_connect
+    # Patch StreamableHTTPServerTransport constructor to track instances
 
-            created_transports.append(transport)
-            return transport
+    original_constructor = streamable_http_manager.StreamableHTTPServerTransport
 
-        MockTransport.side_effect = create_transport
+    def track_transport(*args, **kwargs):
+        transport = original_constructor(*args, **kwargs)
+        created_transports.append(transport)
+        return transport
 
+    with patch.object(streamable_http_manager, "StreamableHTTPServerTransport", side_effect=track_transport):
         async with manager.run():
-            # Mock app.run to return quickly
-            mock_mcp_run = AsyncMock(return_value=None)
-            app.run = mock_mcp_run
+            # Mock app.run to complete immediately
+            app.run = AsyncMock(return_value=None)
 
-            # Send a stateless request
+            # Send a simple request
             sent_messages = []
 
             async def mock_send(message):
@@ -258,25 +239,27 @@ async def mock_send(message):
                 ],
             }
 
+            # Empty body to trigger early return
             async def mock_receive():
                 return {
                     "type": "http.request",
-                    "body": b'{"jsonrpc": "2.0", "method": "test", "id": 1}',
+                    "body": b"",
                     "more_body": False,
                 }
 
-            # Send multiple requests
-            num_requests = 3
-            for _ in range(num_requests):
-                await manager.handle_request(scope, mock_receive, mock_send)
-                # Give async tasks time to complete
-                await anyio.sleep(0.1)
+            # Send a request
+            await manager.handle_request(scope, mock_receive, mock_send)
+
+            # Give async tasks time to complete
+            await anyio.sleep(0.1)
+
+            # Verify transport was created
+            assert len(created_transports) == 1, "Should have created one transport"
+
+            transport = created_transports[0]
 
-            # Verify each transport was created
-            assert len(created_transports) == num_requests, (
-                f"Expected {num_requests} transports, got {len(created_transports)}"
-            )
+            # The key assertion - transport should be terminated
+            assert transport._terminated, "Transport should be terminated after stateless request"
 
-            # This is the key assertion - each transport should have been terminated
-            for i, transport in enumerate(created_transports):
-                assert transport._terminated, f"Transport {i} was not terminated"
+            # Verify internal state is cleaned up
+            assert len(transport._request_streams) == 0, "Transport should have no active request streams"