17
17
from mcp .shared .exceptions import McpError
18
18
from mcp .shared .message import SessionMessage
19
19
from mcp .types import CONNECTION_CLOSED , JSONRPCMessage , JSONRPCRequest , JSONRPCResponse
20
+ from tests .shared .test_win32_utils import escape_path_for_python
20
21
21
22
# Timeout for cleanup of processes that ignore SIGTERM
22
23
# This timeout ensures the test fails quickly if the cleanup logic doesn't have
@@ -249,12 +250,6 @@ class TestChildProcessCleanup:
249
250
This is a fundamental difference between Windows and Unix process termination.
250
251
"""
251
252
252
- @staticmethod
253
- def _escape_path_for_python (path : str ) -> str :
254
- """Escape a file path for use in Python code strings."""
255
- # Use forward slashes which work on all platforms and don't need escaping
256
- return repr (path .replace ("\\ " , "/" ))
257
-
258
253
@pytest .mark .anyio
259
254
@pytest .mark .filterwarnings ("ignore::ResourceWarning" if sys .platform == "win32" else "default" )
260
255
async def test_basic_child_process_cleanup (self ):
@@ -280,13 +275,13 @@ async def test_basic_child_process_cleanup(self):
280
275
import os
281
276
282
277
# Mark that parent started
283
- with open({ self . _escape_path_for_python (parent_marker )} , 'w') as f:
278
+ with open({ escape_path_for_python (parent_marker )} , 'w') as f:
284
279
f.write('parent started\\ n')
285
280
286
281
# Child script that writes continuously
287
282
child_script = f'''
288
283
import time
289
- with open({ self . _escape_path_for_python (marker_file )} , 'a') as f:
284
+ with open({ escape_path_for_python (marker_file )} , 'a') as f:
290
285
while True:
291
286
f.write(f"{ time .time ()} ")
292
287
f.flush()
@@ -335,9 +330,9 @@ async def test_basic_child_process_cleanup(self):
335
330
final_size = os .path .getsize (marker_file )
336
331
337
332
print (f"After cleanup: file size { size_after_cleanup } -> { final_size } " )
338
- assert final_size == size_after_cleanup , (
339
- f"Child process still running! File grew by { final_size - size_after_cleanup } bytes"
340
- )
333
+ assert (
334
+ final_size == size_after_cleanup
335
+ ), f"Child process still running! File grew by { final_size - size_after_cleanup } bytes"
341
336
342
337
print ("SUCCESS: Child process was properly terminated" )
343
338
@@ -381,7 +376,7 @@ async def test_nested_process_tree(self):
381
376
382
377
# Grandchild just writes to file
383
378
grandchild_script = \" \" \" import time
384
- with open({ self . _escape_path_for_python (grandchild_file )} , 'a') as f:
379
+ with open({ escape_path_for_python (grandchild_file )} , 'a') as f:
385
380
while True:
386
381
f.write(f"gc {{time.time()}}")
387
382
f.flush()
@@ -391,7 +386,7 @@ async def test_nested_process_tree(self):
391
386
subprocess.Popen([sys.executable, '-c', grandchild_script])
392
387
393
388
# Child writes to its file
394
- with open({ self . _escape_path_for_python (child_file )} , 'a') as f:
389
+ with open({ escape_path_for_python (child_file )} , 'a') as f:
395
390
while True:
396
391
f.write(f"c { time .time ()} ")
397
392
f.flush()
@@ -401,7 +396,7 @@ async def test_nested_process_tree(self):
401
396
subprocess.Popen([sys.executable, '-c', child_script])
402
397
403
398
# Parent writes to its file
404
- with open({ self . _escape_path_for_python (parent_file )} , 'a') as f:
399
+ with open({ escape_path_for_python (parent_file )} , 'a') as f:
405
400
while True:
406
401
f.write(f"p { time .time ()} ")
407
402
f.flush()
@@ -470,7 +465,7 @@ async def test_early_parent_exit(self):
470
465
471
466
# Child that continues running
472
467
child_script = f'''import time
473
- with open({ self . _escape_path_for_python (marker_file )} , 'a') as f:
468
+ with open({ escape_path_for_python (marker_file )} , 'a') as f:
474
469
while True:
475
470
f.write(f"child { time .time ()} ")
476
471
f.flush()
0 commit comments