Skip to content

Commit 6a33357

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 Uses @pytest.mark.skipif pattern consistent with codebase conventions. Github-Issue: #1027
1 parent dced223 commit 6a33357

File tree

1 file changed

+155
-0
lines changed

1 file changed

+155
-0
lines changed

tests/client/test_windows_fallback.py

Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
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(os.name != "nt", reason="Windows-specific functionality")
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+
assert hasattr(signal, "CTRL_C_EVENT"), "CTRL_C_EVENT not available on this Windows system"
117+
118+
# Verify it's the expected value (should be 0)
119+
assert signal.CTRL_C_EVENT == 0
120+
121+
@pytest.mark.anyio
122+
async def test_fallback_process_with_stdio(self, tmp_path: Path):
123+
"""Test that FallbackProcess properly wraps stdin/stdout streams."""
124+
# Create a simple echo script
125+
echo_script = tmp_path / "echo.py"
126+
echo_script.write_text("""
127+
import sys
128+
while True:
129+
line = sys.stdin.readline()
130+
if not line:
131+
break
132+
sys.stdout.write(f"ECHO: {line}")
133+
sys.stdout.flush()
134+
""")
135+
136+
# Create process
137+
process = await create_windows_process(sys.executable, [str(echo_script)], cwd=tmp_path)
138+
139+
# Test async I/O
140+
assert process.stdin is not None
141+
assert process.stdout is not None
142+
143+
# Write to stdin
144+
test_message = b"Hello Windows\\n"
145+
await process.stdin.send(test_message)
146+
147+
# Read from stdout
148+
import asyncio
149+
150+
response = await asyncio.wait_for(process.stdout.receive(1024), timeout=2.0)
151+
152+
assert b"ECHO: Hello Windows" in response
153+
154+
# Cleanup
155+
await process.__aexit__(None, None, None)

0 commit comments

Comments
 (0)