22
33import os
44import signal
5- import subprocess
65import sys
7- import tempfile
86from pathlib import Path
7+ from typing import TYPE_CHECKING
98
109import pytest
1110
12- if sys .platform == "win32" :
13- from mcp .client .stdio .win32 import FallbackProcess , create_windows_process
11+ if TYPE_CHECKING or sys .platform == "win32" :
12+ from mcp .client .stdio .win32 import create_windows_process
1413
1514
1615@pytest .mark .skipif (sys .platform != "win32" , reason = "Windows-specific tests" )
@@ -23,8 +22,8 @@ async def test_fallback_process_graceful_shutdown(self, tmp_path: Path):
2322 # Create a test script that writes a marker on cleanup
2423 test_script = tmp_path / "test_cleanup.py"
2524 marker_file = tmp_path / "cleanup_marker.txt"
26-
27- test_script .write_text (f'''
25+
26+ test_script .write_text (f"""
2827import signal
2928import time
3029from pathlib import Path
@@ -42,41 +41,37 @@ def cleanup_handler(signum, frame):
4241# Keep process alive
4342while True:
4443 time.sleep(0.1)
45- ''' )
46-
44+ """ )
45+
4746 # Create process using FallbackProcess
48- process = await create_windows_process (
49- sys .executable ,
50- [str (test_script )],
51- cwd = tmp_path
52- )
53-
47+ process = await create_windows_process (sys .executable , [str (test_script )], cwd = tmp_path )
48+
5449 # Wait for process to start
5550 import asyncio
51+
5652 await asyncio .sleep (0.5 )
57-
53+
5854 # Verify process started
5955 assert marker_file .exists ()
6056 assert marker_file .read_text () == "STARTED"
61-
57+
6258 # Exit context manager - should trigger CTRL_C_EVENT
6359 await process .__aexit__ (None , None , None )
64-
60+
6561 # Check if cleanup ran
6662 await asyncio .sleep (0.5 )
67-
63+
6864 # 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"
65+ assert marker_file .read_text () == "CLEANED_UP" , "CTRL_C_EVENT cleanup did not execute - issue #1027 not fixed"
7166
7267 @pytest .mark .anyio
7368 async def test_fallback_process_timeout_fallback (self , tmp_path : Path ):
7469 """Test that FallbackProcess falls back to terminate() if CTRL_C_EVENT times out."""
7570 # Create a test script that ignores CTRL_C_EVENT
7671 test_script = tmp_path / "test_ignore_signal.py"
7772 marker_file = tmp_path / "status_marker.txt"
78-
79- test_script .write_text (f'''
73+
74+ test_script .write_text (f"""
8075import signal
8176import time
8277from pathlib import Path
@@ -90,25 +85,22 @@ async def test_fallback_process_timeout_fallback(self, tmp_path: Path):
9085# Keep process alive
9186while True:
9287 time.sleep(0.1)
93- ''' )
94-
88+ """ )
89+
9590 # Create process
96- process = await create_windows_process (
97- sys .executable ,
98- [str (test_script )],
99- cwd = tmp_path
100- )
101-
91+ process = await create_windows_process (sys .executable , [str (test_script )], cwd = tmp_path )
92+
10293 # Wait for process to start
10394 import asyncio
95+
10496 await asyncio .sleep (0.5 )
105-
97+
10698 assert marker_file .exists ()
10799 assert marker_file .read_text () == "STARTED"
108-
100+
109101 # Exit context manager - should try CTRL_C_EVENT, timeout, then terminate
110102 await process .__aexit__ (None , None , None )
111-
103+
112104 # Process should be terminated even though it ignored CTRL_C_EVENT
113105 # Check that process is no longer running
114106 try :
@@ -122,50 +114,43 @@ async def test_fallback_process_timeout_fallback(self, tmp_path: Path):
122114 def test_ctrl_c_event_availability (self ):
123115 """Test that CTRL_C_EVENT is available on Windows."""
124116 if sys .platform == "win32" :
125- assert hasattr (signal , "CTRL_C_EVENT" ), \
126- "CTRL_C_EVENT not available on this Windows system"
127-
117+ assert hasattr (signal , "CTRL_C_EVENT" ), "CTRL_C_EVENT not available on this Windows system"
118+
128119 # Verify it's the expected value (should be 0)
129120 assert signal .CTRL_C_EVENT == 0
130121
131- @pytest .mark .anyio
122+ @pytest .mark .anyio
132123 async def test_fallback_process_with_stdio (self , tmp_path : Path ):
133124 """Test that FallbackProcess properly wraps stdin/stdout streams."""
134125 # Create a simple echo script
135126 echo_script = tmp_path / "echo.py"
136- echo_script .write_text ('''
127+ echo_script .write_text ("""
137128import sys
138129while True:
139130 line = sys.stdin.readline()
140131 if not line:
141132 break
142133 sys.stdout.write(f"ECHO: {line}")
143134 sys.stdout.flush()
144- ''' )
145-
135+ """ )
136+
146137 # Create process
147- process = await create_windows_process (
148- sys .executable ,
149- [str (echo_script )],
150- cwd = tmp_path
151- )
152-
138+ process = await create_windows_process (sys .executable , [str (echo_script )], cwd = tmp_path )
139+
153140 # Test async I/O
154141 assert process .stdin is not None
155142 assert process .stdout is not None
156-
143+
157144 # Write to stdin
158145 test_message = b"Hello Windows\\ n"
159146 await process .stdin .send (test_message )
160-
147+
161148 # Read from stdout
162149 import asyncio
163- response = await asyncio .wait_for (
164- process .stdout .receive (1024 ),
165- timeout = 2.0
166- )
167-
150+
151+ response = await asyncio .wait_for (process .stdout .receive (1024 ), timeout = 2.0 )
152+
168153 assert b"ECHO: Hello Windows" in response
169-
154+
170155 # Cleanup
171- await process .__aexit__ (None , None , None )
156+ await process .__aexit__ (None , None , None )
0 commit comments