Apply review feedback

- Replace asyncio.timeout with anyio.fail_after for consistency
- Simplify exception handling
- Uncomment Windows-specific test skip condition
- Rewrite tests to focus on verifying no hang occurs (issue #552)
- Use simple Python subprocesses that exit quickly
- Remove complex assertions - test passes if it completes without timeout

Author: felixweinberger

tests/issues/test_552_windows_hang.py

@@ -1,50 +1,67 @@
-import asyncio
+"""Test for issue #552: stdio_client hangs on Windows."""
+
 import sys
 
+import anyio
 import pytest
 
-from mcp import ClientSession, StdioServerParameters
+from mcp import StdioServerParameters
 from mcp.client.stdio import stdio_client
 
-# @pytest.mark.skipif(sys.platform != "win32", reason="Windows-specific test")
-@pytest.mark.skipif(sys.version_info < (3, 11), reason="asyncio.timeout in 3.11+")
-@pytest.mark.parametrize(
-    "args",
-    [
-        ["/C", "echo", '{"jsonrpc": "2.0", "id": 1, "result": null}'],
-        ["dfghfgh"],
-        ["/C", "echo"],
-    ],
-)
+
+@pytest.mark.skipif(sys.platform != "win32", reason="Windows-specific test")
+@pytest.mark.anyio
+async def test_windows_stdio_client_no_hang_on_exit():
+    """
+    Test for issue #552: stdio_client hangs on Windows 11.
+
+    This test verifies that the stdio_client properly handles subprocess
+    termination on Windows without hanging. The key is that the test
+    completes within the timeout period.
+    """
+    # Create a subprocess that exits after a short delay
+    # This gives stdio_client time to set up pipes before the process exits
+    exit_script = "import time; time.sleep(0.1); import sys; sys.exit(0)"
+
+    params = StdioServerParameters(
+        command=sys.executable,
+        args=["-c", exit_script],
+    )
+
+    # The test passes if this completes without timing out
+    # We don't care about exceptions - just that it doesn't hang
+    with anyio.fail_after(10):  # Generous timeout - should complete in < 1s
+        try:
+            async with stdio_client(params) as (read, write):
+                # If we get here, the connection was established
+                # The subprocess will exit soon, causing cleanup
+                await anyio.sleep(0.5)  # Wait for subprocess to exit
+        except Exception:
+            # Any exception is fine - we're just testing for hangs
+            pass
+
+
+@pytest.mark.skipif(sys.platform != "win32", reason="Windows-specific test")
 @pytest.mark.anyio
-async def test_windows_process_creation(args):
+async def test_windows_stdio_client_immediate_exit():
     """
-    Test that directly tests the process creation function that was fixed in issue #552.
-    This simpler test verifies that Windows process creation works without hanging.
+    Test that stdio_client handles immediate subprocess exit on Windows.
+
+    This tests the edge case where the subprocess exits before
+    stdio_client can fully initialize.
     """
-    # Use a simple command that should complete quickly on Windows
+    # Subprocess exits immediately
     params = StdioServerParameters(
-        command="cmd",
-        # Echo a valid JSON-RPC response message that will be parsed correctly
-        args=args,
+        command=sys.executable,
+        args=["-c", "import sys; sys.exit(1)"],
     )
 
-    # Directly test the fixed function that was causing the hanging issue
-    try:
-        # Set a timeout to prevent hanging
-        async with asyncio.timeout(5):
-            # Test the actual process creation function that was fixed
+    # The test passes if this completes without timing out
+    with anyio.fail_after(10):  # Generous timeout
+        try:
             async with stdio_client(params) as (read, write):
-                print("inside client")
-                async with ClientSession(read, write) as c:
-                    print("inside ClientSession")
-                    await c.initialize()
-
-    except asyncio.TimeoutError:
-        pytest.xfail("Process creation timed out, indicating a hang issue")
-    except ProcessLookupError:
-        pytest.xfail("Process creation failed with ProcessLookupError")
-    except Exception as e:
-        assert "ExceptionGroup" in repr(e), f"Unexpected error: {e}"
-        assert "ProcessLookupError" in repr(e), f"Unexpected error: {e}"
-        pytest.xfail(f"Expected error: {e}")
+                # Subprocess should have exited already
+                pass
+        except Exception:
+            # Expected - subprocess exited
+            pass