Skip to content

Commit bedd4cc

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 Github-Issue: #1027
1 parent 4b65963 commit bedd4cc

File tree

1 file changed

+156
-0
lines changed

1 file changed

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

0 commit comments

Comments
 (0)