Skip to content

Commit e17b13a

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. Includes detailed documentation explaining why the metaprogramming approach is necessary for testing OS-level signal handling between processes. 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 e17b13a

File tree

1 file changed

+200
-0
lines changed

1 file changed

+200
-0
lines changed

tests/client/test_windows_fallback.py

Lines changed: 200 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,200 @@
1+
"""Test Windows-specific FallbackProcess functionality.
2+
3+
Why this test approach is necessary:
4+
------------------------------------
5+
Testing Windows process signal handling requires actual subprocess creation because:
6+
7+
1. SIGNAL HANDLING: We need to verify that CTRL_C_EVENT signals are properly sent and
8+
received. This cannot be mocked as it involves OS-level signal propagation between
9+
parent and child processes.
10+
11+
2. CLEANUP VERIFICATION: The core issue (#1027) is that cleanup code in lifespan context
12+
managers wasn't executing on Windows. We must verify that signal handlers actually run
13+
and that cleanup code executes before process termination.
14+
15+
3. WINDOWS-SPECIFIC BEHAVIOR: The FallbackProcess class exists specifically to work around
16+
Windows asyncio limitations. Testing it requires actual Windows subprocess creation to
17+
ensure the workarounds function correctly.
18+
19+
4. INTEGRATION TESTING: These tests verify the integration between:
20+
- FallbackProcess wrapper
21+
- Windows signal handling (CTRL_C_EVENT)
22+
- Asyncio file streams
23+
- Process cleanup behavior
24+
25+
Test Implementation:
26+
-------------------
27+
The tests create temporary Python scripts that:
28+
1. Set up signal handlers for CTRL_C_EVENT
29+
2. Write marker files to indicate execution state
30+
3. Allow verification that cleanup ran before termination
31+
32+
This metaprogramming approach is used because:
33+
- The codebase doesn't have a test fixtures directory pattern
34+
- Inline `python -c` would be even less readable for complex scripts
35+
- We need actual subprocess execution to test OS-level behavior
36+
"""
37+
38+
import os
39+
import signal
40+
import sys
41+
import textwrap
42+
from pathlib import Path
43+
from typing import TYPE_CHECKING
44+
45+
import pytest
46+
47+
if TYPE_CHECKING or sys.platform == "win32":
48+
from mcp.client.stdio.win32 import create_windows_process
49+
50+
51+
@pytest.mark.skipif(os.name != "nt", reason="Windows-specific functionality")
52+
class TestFallbackProcess:
53+
"""Test suite for Windows FallbackProcess graceful shutdown."""
54+
55+
@pytest.mark.anyio
56+
async def test_fallback_process_graceful_shutdown(self, tmp_path: Path):
57+
"""Test that FallbackProcess sends CTRL_C_EVENT for graceful shutdown."""
58+
# Create a test script that writes a marker on cleanup
59+
test_script = tmp_path / "test_cleanup.py"
60+
marker_file = tmp_path / "cleanup_marker.txt"
61+
62+
# Create a test script that handles CTRL_C_EVENT and writes a marker on cleanup
63+
test_script.write_text(
64+
textwrap.dedent(f"""
65+
import signal
66+
import time
67+
from pathlib import Path
68+
69+
marker = Path(r"{marker_file}")
70+
marker.write_text("STARTED")
71+
72+
def cleanup_handler(signum, frame):
73+
# This handler should be called when CTRL_C_EVENT is received
74+
marker.write_text("CLEANED_UP")
75+
exit(0)
76+
77+
# Register CTRL_C_EVENT handler (SIGINT on Windows)
78+
signal.signal(signal.SIGINT, cleanup_handler)
79+
80+
# Keep process alive waiting for signal
81+
while True:
82+
time.sleep(0.1)
83+
""").strip()
84+
)
85+
86+
# Create process using FallbackProcess
87+
process = await create_windows_process(sys.executable, [str(test_script)], cwd=tmp_path)
88+
89+
# Wait for process to start
90+
import asyncio
91+
92+
await asyncio.sleep(0.5)
93+
94+
# Verify process started
95+
assert marker_file.exists()
96+
assert marker_file.read_text() == "STARTED"
97+
98+
# Exit context manager - should trigger CTRL_C_EVENT
99+
await process.__aexit__(None, None, None)
100+
101+
# Check if cleanup ran
102+
await asyncio.sleep(0.5)
103+
104+
# This is the critical test: cleanup should have executed
105+
assert marker_file.read_text() == "CLEANED_UP", "CTRL_C_EVENT cleanup did not execute - issue #1027 not fixed"
106+
107+
@pytest.mark.anyio
108+
async def test_fallback_process_timeout_fallback(self, tmp_path: Path):
109+
"""Test that FallbackProcess falls back to terminate() if CTRL_C_EVENT times out."""
110+
# Create a test script that ignores CTRL_C_EVENT
111+
test_script = tmp_path / "test_ignore_signal.py"
112+
marker_file = tmp_path / "status_marker.txt"
113+
114+
# Create a test script that ignores CTRL_C_EVENT to test fallback behavior
115+
test_script.write_text(
116+
textwrap.dedent(f"""
117+
import signal
118+
import time
119+
from pathlib import Path
120+
121+
marker = Path(r"{marker_file}")
122+
marker.write_text("STARTED")
123+
124+
# Explicitly ignore CTRL_C_EVENT to test fallback to terminate()
125+
signal.signal(signal.SIGINT, signal.SIG_IGN)
126+
127+
# Keep process alive - should be forcefully terminated
128+
while True:
129+
time.sleep(0.1)
130+
""").strip()
131+
)
132+
133+
# Create process
134+
process = await create_windows_process(sys.executable, [str(test_script)], cwd=tmp_path)
135+
136+
# Wait for process to start
137+
import asyncio
138+
139+
await asyncio.sleep(0.5)
140+
141+
assert marker_file.exists()
142+
assert marker_file.read_text() == "STARTED"
143+
144+
# Exit context manager - should try CTRL_C_EVENT, timeout, then terminate
145+
await process.__aexit__(None, None, None)
146+
147+
# Process should be terminated even though it ignored CTRL_C_EVENT
148+
# Check that process is no longer running
149+
try:
150+
# This should raise because process is terminated
151+
os.kill(process.popen.pid, 0)
152+
pytest.fail("Process should have been terminated")
153+
except (ProcessLookupError, OSError):
154+
# Expected - process is terminated
155+
pass
156+
157+
def test_ctrl_c_event_availability(self):
158+
"""Test that CTRL_C_EVENT is available on Windows."""
159+
assert hasattr(signal, "CTRL_C_EVENT"), "CTRL_C_EVENT not available on this Windows system"
160+
161+
# Verify it's the expected value (should be 0)
162+
assert signal.CTRL_C_EVENT == 0
163+
164+
@pytest.mark.anyio
165+
async def test_fallback_process_with_stdio(self, tmp_path: Path):
166+
"""Test that FallbackProcess properly wraps stdin/stdout streams."""
167+
# Create a simple echo script to test stdio stream wrapping
168+
echo_script = tmp_path / "echo.py"
169+
echo_script.write_text(
170+
textwrap.dedent("""
171+
import sys
172+
while True:
173+
line = sys.stdin.readline()
174+
if not line:
175+
break
176+
sys.stdout.write(f"ECHO: {line}")
177+
sys.stdout.flush()
178+
""").strip()
179+
)
180+
181+
# Create process
182+
process = await create_windows_process(sys.executable, [str(echo_script)], cwd=tmp_path)
183+
184+
# Test async I/O
185+
assert process.stdin is not None
186+
assert process.stdout is not None
187+
188+
# Write to stdin
189+
test_message = b"Hello Windows\\n"
190+
await process.stdin.send(test_message)
191+
192+
# Read from stdout
193+
import asyncio
194+
195+
response = await asyncio.wait_for(process.stdout.receive(1024), timeout=2.0)
196+
197+
assert b"ECHO: Hello Windows" in response
198+
199+
# Cleanup
200+
await process.__aexit__(None, None, None)

0 commit comments

Comments
 (0)