fix: Add integration markers to all multiprocessing tests

- Mark tests/shared/test_sse.py as integration (spawns subprocesses)
- Mark tests/shared/test_streamable_http.py as integration (spawns subprocesses)
- Mark tests/shared/test_ws.py as integration (spawns subprocesses)
- Mark tests/server/test_sse_security.py as integration (spawns subprocesses)
- Mark tests/server/test_streamable_http_security.py as integration (spawns subprocesses)
- Mark tests/client/test_stdio.py as integration (spawns subprocesses)

This ensures all subprocess-spawning tests run sequentially on Windows
to prevent parallelization conflicts that cause test hangs.

Author: spacelord16

tests/client/test_stdio.py

@@ -19,6 +19,9 @@
 from mcp.types import CONNECTION_CLOSED, JSONRPCMessage, JSONRPCRequest, JSONRPCResponse
 from tests.shared.test_win32_utils import escape_path_for_python
 
+# Mark all tests in this file as integration tests (spawn subprocesses)
+pytestmark = [pytest.mark.integration]
+
 # Timeout for cleanup of processes that ignore SIGTERM
 # This timeout ensures the test fails quickly if the cleanup logic doesn't have
 # proper fallback mechanisms (SIGINT/SIGKILL) for processes that ignore SIGTERM
@@ -63,14 +66,20 @@ async def test_stdio_client():
                     break
 
         assert len(read_messages) == 2
-        assert read_messages[0] == JSONRPCMessage(root=JSONRPCRequest(jsonrpc="2.0", id=1, method="ping"))
-        assert read_messages[1] == JSONRPCMessage(root=JSONRPCResponse(jsonrpc="2.0", id=2, result={}))
+        assert read_messages[0] == JSONRPCMessage(
+            root=JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
+        )
+        assert read_messages[1] == JSONRPCMessage(
+            root=JSONRPCResponse(jsonrpc="2.0", id=2, result={})
+        )
 
 
 @pytest.mark.anyio
 async def test_stdio_client_bad_path():
     """Check that the connection doesn't hang if process errors."""
-    server_params = StdioServerParameters(command="python", args=["-c", "non-existent-file.py"])
+    server_params = StdioServerParameters(
+        command="python", args=["-c", "non-existent-file.py"]
+    )
     async with stdio_client(server_params) as (read_stream, write_stream):
         async with ClientSession(read_stream, write_stream) as session:
             # The session should raise an error when the connection closes
@@ -158,7 +167,9 @@ async def test_stdio_client_universal_cleanup():
 
 
 @pytest.mark.anyio
-@pytest.mark.skipif(sys.platform == "win32", reason="Windows signal handling is different")
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="Windows signal handling is different"
+)
 async def test_stdio_client_sigint_only_process():
     """
     Test cleanup with a process that ignores SIGTERM but responds to SIGINT.
@@ -251,7 +262,9 @@ class TestChildProcessCleanup:
     """
 
     @pytest.mark.anyio
-    @pytest.mark.filterwarnings("ignore::ResourceWarning" if sys.platform == "win32" else "default")
+    @pytest.mark.filterwarnings(
+        "ignore::ResourceWarning" if sys.platform == "win32" else "default"
+    )
     async def test_basic_child_process_cleanup(self):
         """
         Test basic parent-child process cleanup.
@@ -300,7 +313,9 @@ async def test_basic_child_process_cleanup(self):
             print("\nStarting child process termination test...")
 
             # Start the parent process
-            proc = await _create_platform_compatible_process(sys.executable, ["-c", parent_script])
+            proc = await _create_platform_compatible_process(
+                sys.executable, ["-c", parent_script]
+            )
 
             # Wait for processes to start
             await anyio.sleep(0.5)
@@ -314,7 +329,9 @@ async def test_basic_child_process_cleanup(self):
                 await anyio.sleep(0.3)
                 size_after_wait = os.path.getsize(marker_file)
                 assert size_after_wait > initial_size, "Child process should be writing"
-                print(f"Child is writing (file grew from {initial_size} to {size_after_wait} bytes)")
+                print(
+                    f"Child is writing (file grew from {initial_size} to {size_after_wait} bytes)"
+                )
 
             # Terminate using our function
             print("Terminating process and children...")
@@ -330,9 +347,9 @@ async def test_basic_child_process_cleanup(self):
                 final_size = os.path.getsize(marker_file)
 
                 print(f"After cleanup: file size {size_after_cleanup} -> {final_size}")
-                assert final_size == size_after_cleanup, (
-                    f"Child process still running! File grew by {final_size - size_after_cleanup} bytes"
-                )
+                assert (
+                    final_size == size_after_cleanup
+                ), f"Child process still running! File grew by {final_size - size_after_cleanup} bytes"
 
             print("SUCCESS: Child process was properly terminated")
 
@@ -345,7 +362,9 @@ async def test_basic_child_process_cleanup(self):
                     pass
 
     @pytest.mark.anyio
-    @pytest.mark.filterwarnings("ignore::ResourceWarning" if sys.platform == "win32" else "default")
+    @pytest.mark.filterwarnings(
+        "ignore::ResourceWarning" if sys.platform == "win32" else "default"
+    )
     async def test_nested_process_tree(self):
         """
         Test nested process tree cleanup (parent → child → grandchild).
@@ -405,13 +424,19 @@ async def test_nested_process_tree(self):
             )
 
             # Start the parent process
-            proc = await _create_platform_compatible_process(sys.executable, ["-c", parent_script])
+            proc = await _create_platform_compatible_process(
+                sys.executable, ["-c", parent_script]
+            )
 
             # Let all processes start
             await anyio.sleep(1.0)
 
             # Verify all are writing
-            for file_path, name in [(parent_file, "parent"), (child_file, "child"), (grandchild_file, "grandchild")]:
+            for file_path, name in [
+                (parent_file, "parent"),
+                (child_file, "child"),
+                (grandchild_file, "grandchild"),
+            ]:
                 if os.path.exists(file_path):
                     initial_size = os.path.getsize(file_path)
                     await anyio.sleep(0.3)
@@ -425,7 +450,11 @@ 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")]:
+            for file_path, name in [
+                (parent_file, "parent"),
+                (child_file, "child"),
+                (grandchild_file, "grandchild"),
+            ]:
                 if os.path.exists(file_path):
                     size1 = os.path.getsize(file_path)
                     await anyio.sleep(0.3)
@@ -443,7 +472,9 @@ async def test_nested_process_tree(self):
                     pass
 
     @pytest.mark.anyio
-    @pytest.mark.filterwarnings("ignore::ResourceWarning" if sys.platform == "win32" else "default")
+    @pytest.mark.filterwarnings(
+        "ignore::ResourceWarning" if sys.platform == "win32" else "default"
+    )
     async def test_early_parent_exit(self):
         """
         Test cleanup when parent exits during termination sequence.
@@ -487,7 +518,9 @@ def handle_term(sig, frame):
             )
 
             # Start the parent process
-            proc = await _create_platform_compatible_process(sys.executable, ["-c", parent_script])
+            proc = await _create_platform_compatible_process(
+                sys.executable, ["-c", parent_script]
+            )
 
             # Let child start writing
             await anyio.sleep(0.5)

tests/server/test_sse_security.py

@@ -18,6 +18,10 @@
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.types import Tool
 
+# Mark all tests in this file as integration tests (spawn subprocesses)
+pytestmark = [pytest.mark.integration]
+
+
 logger = logging.getLogger(__name__)
 SERVER_NAME = "test_sse_security_server"
 
@@ -42,16 +46,22 @@ async def on_list_tools(self) -> list[Tool]:
         return []
 
 
-def run_server_with_settings(port: int, security_settings: TransportSecuritySettings | None = None):
+def run_server_with_settings(
+    port: int, security_settings: TransportSecuritySettings | None = None
+):
     """Run the SSE server with specified security settings."""
     app = SecurityTestServer()
     sse_transport = SseServerTransport("/messages/", security_settings)
 
     async def handle_sse(request: Request):
         try:
-            async with sse_transport.connect_sse(request.scope, request.receive, request._send) as streams:
+            async with sse_transport.connect_sse(
+                request.scope, request.receive, request._send
+            ) as streams:
                 if streams:
-                    await app.run(streams[0], streams[1], app.create_initialization_options())
+                    await app.run(
+                        streams[0], streams[1], app.create_initialization_options()
+                    )
         except ValueError as e:
             # Validation error was already handled inside connect_sse
             logger.debug(f"SSE connection failed validation: {e}")
@@ -66,9 +76,13 @@ async def handle_sse(request: Request):
     uvicorn.run(starlette_app, host="127.0.0.1", port=port, log_level="error")
 
 
-def start_server_process(port: int, security_settings: TransportSecuritySettings | None = None):
+def start_server_process(
+    port: int, security_settings: TransportSecuritySettings | None = None
+):
     """Start server in a separate process."""
-    process = multiprocessing.Process(target=run_server_with_settings, args=(port, security_settings))
+    process = multiprocessing.Process(
+        target=run_server_with_settings, args=(port, security_settings)
+    )
     process.start()
     # Give server time to start
     time.sleep(1)
@@ -84,7 +98,9 @@ async def test_sse_security_default_settings(server_port: int):
         headers = {"Host": "evil.com", "Origin": "http://evil.com"}
 
         async with httpx.AsyncClient(timeout=5.0) as client:
-            async with client.stream("GET", f"http://127.0.0.1:{server_port}/sse", headers=headers) as response:
+            async with client.stream(
+                "GET", f"http://127.0.0.1:{server_port}/sse", headers=headers
+            ) as response:
                 assert response.status_code == 200
     finally:
         process.terminate()
@@ -95,15 +111,19 @@ async def test_sse_security_default_settings(server_port: int):
 async def test_sse_security_invalid_host_header(server_port: int):
     """Test SSE with invalid Host header."""
     # Enable security by providing settings with an empty allowed_hosts list
-    security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["example.com"])
+    security_settings = TransportSecuritySettings(
+        enable_dns_rebinding_protection=True, allowed_hosts=["example.com"]
+    )
     process = start_server_process(server_port, security_settings)
 
     try:
         # Test with invalid host header
         headers = {"Host": "evil.com"}
 
         async with httpx.AsyncClient() as client:
-            response = await client.get(f"http://127.0.0.1:{server_port}/sse", headers=headers)
+            response = await client.get(
+                f"http://127.0.0.1:{server_port}/sse", headers=headers
+            )
             assert response.status_code == 421
             assert response.text == "Invalid Host header"
 
@@ -117,7 +137,9 @@ async def test_sse_security_invalid_origin_header(server_port: int):
     """Test SSE with invalid Origin header."""
     # Configure security to allow the host but restrict origins
     security_settings = TransportSecuritySettings(
-        enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"], allowed_origins=["http://localhost:*"]
+        enable_dns_rebinding_protection=True,
+        allowed_hosts=["127.0.0.1:*"],
+        allowed_origins=["http://localhost:*"],
     )
     process = start_server_process(server_port, security_settings)
 
@@ -126,7 +148,9 @@ async def test_sse_security_invalid_origin_header(server_port: int):
         headers = {"Origin": "http://evil.com"}
 
         async with httpx.AsyncClient() as client:
-            response = await client.get(f"http://127.0.0.1:{server_port}/sse", headers=headers)
+            response = await client.get(
+                f"http://127.0.0.1:{server_port}/sse", headers=headers
+            )
             assert response.status_code == 400
             assert response.text == "Invalid Origin header"
 
@@ -140,7 +164,9 @@ async def test_sse_security_post_invalid_content_type(server_port: int):
     """Test POST endpoint with invalid Content-Type header."""
     # Configure security to allow the host
     security_settings = TransportSecuritySettings(
-        enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"], allowed_origins=["http://127.0.0.1:*"]
+        enable_dns_rebinding_protection=True,
+        allowed_hosts=["127.0.0.1:*"],
+        allowed_origins=["http://127.0.0.1:*"],
     )
     process = start_server_process(server_port, security_settings)
 
@@ -158,7 +184,8 @@ async def test_sse_security_post_invalid_content_type(server_port: int):
 
             # Test POST with missing content type
             response = await client.post(
-                f"http://127.0.0.1:{server_port}/messages/?session_id={fake_session_id}", content="test"
+                f"http://127.0.0.1:{server_port}/messages/?session_id={fake_session_id}",
+                content="test",
             )
             assert response.status_code == 400
             assert response.text == "Invalid Content-Type header"
@@ -180,7 +207,9 @@ async def test_sse_security_disabled(server_port: int):
 
         async with httpx.AsyncClient(timeout=5.0) as client:
             # For SSE endpoints, we need to use stream to avoid timeout
-            async with client.stream("GET", f"http://127.0.0.1:{server_port}/sse", headers=headers) as response:
+            async with client.stream(
+                "GET", f"http://127.0.0.1:{server_port}/sse", headers=headers
+            ) as response:
                 # Should connect successfully even with invalid host
                 assert response.status_code == 200
 
@@ -205,15 +234,19 @@ async def test_sse_security_custom_allowed_hosts(server_port: int):
 
         async with httpx.AsyncClient(timeout=5.0) as client:
             # For SSE endpoints, we need to use stream to avoid timeout
-            async with client.stream("GET", f"http://127.0.0.1:{server_port}/sse", headers=headers) as response:
+            async with client.stream(
+                "GET", f"http://127.0.0.1:{server_port}/sse", headers=headers
+            ) as response:
                 # Should connect successfully with custom host
                 assert response.status_code == 200
 
         # Test with non-allowed host
         headers = {"Host": "evil.com"}
 
         async with httpx.AsyncClient() as client:
-            response = await client.get(f"http://127.0.0.1:{server_port}/sse", headers=headers)
+            response = await client.get(
+                f"http://127.0.0.1:{server_port}/sse", headers=headers
+            )
             assert response.status_code == 421
             assert response.text == "Invalid Host header"
 
@@ -239,15 +272,19 @@ async def test_sse_security_wildcard_ports(server_port: int):
 
             async with httpx.AsyncClient(timeout=5.0) as client:
                 # For SSE endpoints, we need to use stream to avoid timeout
-                async with client.stream("GET", f"http://127.0.0.1:{server_port}/sse", headers=headers) as response:
+                async with client.stream(
+                    "GET", f"http://127.0.0.1:{server_port}/sse", headers=headers
+                ) as response:
                     # Should connect successfully with any port
                     assert response.status_code == 200
 
             headers = {"Origin": f"http://localhost:{test_port}"}
 
             async with httpx.AsyncClient(timeout=5.0) as client:
                 # For SSE endpoints, we need to use stream to avoid timeout
-                async with client.stream("GET", f"http://127.0.0.1:{server_port}/sse", headers=headers) as response:
+                async with client.stream(
+                    "GET", f"http://127.0.0.1:{server_port}/sse", headers=headers
+                ) as response:
                     # Should connect successfully with any port
                     assert response.status_code == 200
 
@@ -261,7 +298,9 @@ async def test_sse_security_post_valid_content_type(server_port: int):
     """Test POST endpoint with valid Content-Type headers."""
     # Configure security to allow the host
     security_settings = TransportSecuritySettings(
-        enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"], allowed_origins=["http://127.0.0.1:*"]
+        enable_dns_rebinding_protection=True,
+        allowed_hosts=["127.0.0.1:*"],
+        allowed_origins=["http://127.0.0.1:*"],
     )
     process = start_server_process(server_port, security_settings)
 

tests/server/test_streamable_http_security.py

@@ -17,6 +17,9 @@
 from mcp.server import Server
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
+
+# Mark all tests in this file as integration tests (spawn subprocesses)
+pytestmark = [pytest.mark.integration]
 from mcp.types import Tool
 
 logger = logging.getLogger(__name__)

tests/shared/test_sse.py

@@ -32,6 +32,9 @@
     Tool,
 )
 
+# Mark all tests in this file as integration tests (spawn subprocesses)
+pytestmark = [pytest.mark.integration]
+
 SERVER_NAME = "test_server_for_SSE"