1+ """Test Windows-specific FallbackProcess functionality."""
2+
3+ import os
4+ import signal
5+ import subprocess
6+ import sys
7+ import tempfile
8+ from pathlib import Path
9+
10+ import pytest
11+
12+ if sys .platform == "win32" :
13+ from mcp .client .stdio .win32 import FallbackProcess , create_windows_process
14+
15+
16+ @pytest .mark .skipif (sys .platform != "win32" , reason = "Windows-specific tests" )
17+ class TestFallbackProcess :
18+ """Test suite for Windows FallbackProcess graceful shutdown."""
19+
20+ @pytest .mark .anyio
21+ async def test_fallback_process_graceful_shutdown (self , tmp_path : Path ):
22+ """Test that FallbackProcess sends CTRL_C_EVENT for graceful shutdown."""
23+ # Create a test script that writes a marker on cleanup
24+ test_script = tmp_path / "test_cleanup.py"
25+ marker_file = tmp_path / "cleanup_marker.txt"
26+
27+ test_script .write_text (f'''
28+ import signal
29+ import time
30+ from pathlib import Path
31+
32+ marker = Path(r"{ marker_file } ")
33+ marker.write_text("STARTED")
34+
35+ def cleanup_handler(signum, frame):
36+ marker.write_text("CLEANED_UP")
37+ exit(0)
38+
39+ # Register CTRL_C_EVENT handler
40+ signal.signal(signal.SIGINT, cleanup_handler)
41+
42+ # Keep process alive
43+ while True:
44+ time.sleep(0.1)
45+ ''' )
46+
47+ # Create process using FallbackProcess
48+ process = await create_windows_process (
49+ sys .executable ,
50+ [str (test_script )],
51+ cwd = tmp_path
52+ )
53+
54+ # Wait for process to start
55+ import asyncio
56+ await asyncio .sleep (0.5 )
57+
58+ # Verify process started
59+ assert marker_file .exists ()
60+ assert marker_file .read_text () == "STARTED"
61+
62+ # Exit context manager - should trigger CTRL_C_EVENT
63+ await process .__aexit__ (None , None , None )
64+
65+ # Check if cleanup ran
66+ await asyncio .sleep (0.5 )
67+
68+ # This is the critical test: cleanup should have executed
69+ assert marker_file .read_text () == "CLEANED_UP" , \
70+ "CTRL_C_EVENT cleanup did not execute - issue #1027 not fixed"
71+
72+ @pytest .mark .anyio
73+ async def test_fallback_process_timeout_fallback (self , tmp_path : Path ):
74+ """Test that FallbackProcess falls back to terminate() if CTRL_C_EVENT times out."""
75+ # Create a test script that ignores CTRL_C_EVENT
76+ test_script = tmp_path / "test_ignore_signal.py"
77+ marker_file = tmp_path / "status_marker.txt"
78+
79+ test_script .write_text (f'''
80+ import signal
81+ import time
82+ from pathlib import Path
83+
84+ marker = Path(r"{ marker_file } ")
85+ marker.write_text("STARTED")
86+
87+ # Ignore CTRL_C_EVENT
88+ signal.signal(signal.SIGINT, signal.SIG_IGN)
89+
90+ # Keep process alive
91+ while True:
92+ time.sleep(0.1)
93+ ''' )
94+
95+ # Create process
96+ process = await create_windows_process (
97+ sys .executable ,
98+ [str (test_script )],
99+ cwd = tmp_path
100+ )
101+
102+ # Wait for process to start
103+ import asyncio
104+ await asyncio .sleep (0.5 )
105+
106+ assert marker_file .exists ()
107+ assert marker_file .read_text () == "STARTED"
108+
109+ # Exit context manager - should try CTRL_C_EVENT, timeout, then terminate
110+ await process .__aexit__ (None , None , None )
111+
112+ # Process should be terminated even though it ignored CTRL_C_EVENT
113+ # Check that process is no longer running
114+ try :
115+ # This should raise because process is terminated
116+ os .kill (process .popen .pid , 0 )
117+ pytest .fail ("Process should have been terminated" )
118+ except (ProcessLookupError , OSError ):
119+ # Expected - process is terminated
120+ pass
121+
122+ def test_ctrl_c_event_availability (self ):
123+ """Test that CTRL_C_EVENT is available on Windows."""
124+ if sys .platform == "win32" :
125+ assert hasattr (signal , "CTRL_C_EVENT" ), \
126+ "CTRL_C_EVENT not available on this Windows system"
127+
128+ # Verify it's the expected value (should be 0)
129+ assert signal .CTRL_C_EVENT == 0
130+
131+ @pytest .mark .anyio
132+ async def test_fallback_process_with_stdio (self , tmp_path : Path ):
133+ """Test that FallbackProcess properly wraps stdin/stdout streams."""
134+ # Create a simple echo script
135+ echo_script = tmp_path / "echo.py"
136+ echo_script .write_text ('''
137+ import sys
138+ while True:
139+ line = sys.stdin.readline()
140+ if not line:
141+ break
142+ sys.stdout.write(f"ECHO: {line}")
143+ sys.stdout.flush()
144+ ''' )
145+
146+ # Create process
147+ process = await create_windows_process (
148+ sys .executable ,
149+ [str (echo_script )],
150+ cwd = tmp_path
151+ )
152+
153+ # Test async I/O
154+ assert process .stdin is not None
155+ assert process .stdout is not None
156+
157+ # Write to stdin
158+ test_message = b"Hello Windows\\ n"
159+ await process .stdin .send (test_message )
160+
161+ # Read from stdout
162+ import asyncio
163+ response = await asyncio .wait_for (
164+ process .stdout .receive (1024 ),
165+ timeout = 2.0
166+ )
167+
168+ assert b"ECHO: Hello Windows" in response
169+
170+ # Cleanup
171+ await process .__aexit__ (None , None , None )
0 commit comments