Fix Windows compatibility for process group tests

The start_new_session=True parameter in anyio.open_process() is Unix-specific
and causes child processes not to start properly on Windows. This fix:

- Adds WindowsProcessWrapper class to provide anyio-compatible interface
  for subprocess.Popen on Windows
- Uses CREATE_NEW_PROCESS_GROUP flag on Windows instead of start_new_session
- Maintains Unix behavior with start_new_session=True for process group creation

This ensures the child process cleanup tests work correctly on both Windows
and Unix platforms.

Author: felixweinberger

tests/client/test_stdio.py

@@ -1,5 +1,6 @@
 import os
 import shutil
+import subprocess
 import sys
 import tempfile
 import textwrap
@@ -21,6 +22,25 @@
 python: str = shutil.which("python")  # type: ignore
 
 
+class WindowsProcessWrapper:
+    """Minimal wrapper for subprocess.Popen to work with anyio-style process interface."""
+
+    def __init__(self, popen):
+        self.pid = popen.pid
+        self._popen = popen
+
+    async def wait(self):
+        while self._popen.poll() is None:
+            await anyio.sleep(0.1)
+        return self._popen.returncode
+
+    def terminate(self):
+        self._popen.terminate()
+
+    def kill(self):
+        self._popen.kill()
+
+
 @pytest.mark.anyio
 @pytest.mark.skipif(tee is None, reason="could not find tee command")
 async def test_stdio_context_manager_exiting():
@@ -120,7 +140,7 @@ async def test_stdio_client_universal_cleanup():
             sys.stderr.flush()
         """
     )
-    
+
     server_params = StdioServerParameters(
         command=sys.executable,
         args=["-c", long_running_script],
@@ -386,7 +406,15 @@ async def test_stdio_client_child_process_cleanup():
         print("\nStarting child process termination test...")
 
         # Start the parent process directly with process group
-        proc = await anyio.open_process([sys.executable, "-c", parent_script], start_new_session=True)
+        if sys.platform == "win32":
+            # Windows: Use subprocess.Popen with CREATE_NEW_PROCESS_GROUP flag
+            popen = subprocess.Popen(
+                [sys.executable, "-c", parent_script], creationflags=subprocess.CREATE_NEW_PROCESS_GROUP
+            )
+            proc = WindowsProcessWrapper(popen)
+        else:
+            # Unix: Use start_new_session for process group creation
+            proc = await anyio.open_process([sys.executable, "-c", parent_script], start_new_session=True)
 
         # Wait for processes to start
         await anyio.sleep(0.5)
@@ -492,7 +520,15 @@ async def test_stdio_client_nested_process_tree():
         )
 
         # Start parent process directly
-        proc = await anyio.open_process([sys.executable, "-c", parent_script], start_new_session=True)
+        if sys.platform == "win32":
+            # Windows: Use subprocess.Popen with CREATE_NEW_PROCESS_GROUP flag
+            popen = subprocess.Popen(
+                [sys.executable, "-c", parent_script], creationflags=subprocess.CREATE_NEW_PROCESS_GROUP
+            )
+            proc = WindowsProcessWrapper(popen)
+        else:
+            # Unix: Use start_new_session for process group creation
+            proc = await anyio.open_process([sys.executable, "-c", parent_script], start_new_session=True)
 
         # Let all processes start
         await anyio.sleep(1.0)
@@ -575,7 +611,15 @@ def handle_term(sig, frame):
         )
 
         # Start the parent process
-        proc = await anyio.open_process([sys.executable, "-c", parent_script], start_new_session=True)
+        if sys.platform == "win32":
+            # Windows: Use subprocess.Popen with CREATE_NEW_PROCESS_GROUP flag
+            popen = subprocess.Popen(
+                [sys.executable, "-c", parent_script], creationflags=subprocess.CREATE_NEW_PROCESS_GROUP
+            )
+            proc = WindowsProcessWrapper(popen)
+        else:
+            # Unix: Use start_new_session for process group creation
+            proc = await anyio.open_process([sys.executable, "-c", parent_script], start_new_session=True)
 
         # Let child start writing
         await anyio.sleep(0.5)