Skip to content

Commit 1ef622c

Browse files
test: add Windows FallbackProcess graceful shutdown tests
Add comprehensive test suite for Windows-specific FallbackProcess to verify that CTRL_C_EVENT signal properly triggers cleanup code in lifespan context managers. These tests will fail until issue #1027 is fixed. Tests include: - Graceful shutdown with CTRL_C_EVENT signal - Timeout fallback to terminate() when signal is ignored - CTRL_C_EVENT availability verification - Async stdio stream functionality Tests pass silently on non-Windows platforms since FallbackProcess is Windows-specific functionality. Github-Issue: #1027
1 parent 4b65963 commit 1ef622c

File tree

1 file changed

+170
-0
lines changed

1 file changed

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

0 commit comments

Comments
 (0)