|
| 1 | +"""Regression test for issue #42: stdout suppression race condition. |
| 2 | +
|
| 3 | +The global FastMCP middleware held os.dup2() on fd 1 (stdout→stderr) for |
| 4 | +the entire tool call. FastMCP dispatches sync tools via |
| 5 | +anyio.to_thread.run_sync, so two tools CAN run concurrently. When Thread A |
| 6 | +held the redirect, Thread B's JSON-RPC response goes to stderr and the |
| 7 | +client receives nothing → MCP error -32001 timeout. |
| 8 | +
|
| 9 | +This test fires a slow tool (create_baseline_osm, several seconds) and a |
| 10 | +fast tool (get_server_status, near-instant) concurrently. On buggy code, |
| 11 | +get_server_status's response is lost → timeout. After the fix, both return. |
| 12 | +""" |
| 13 | +import asyncio |
| 14 | +import pytest |
| 15 | + |
| 16 | +from conftest import integration_enabled, server_params, unwrap |
| 17 | +from mcp import ClientSession |
| 18 | +from mcp.client.stdio import stdio_client |
| 19 | + |
| 20 | + |
| 21 | +@pytest.mark.integration |
| 22 | +def test_concurrent_tool_calls_both_respond(): |
| 23 | + # Regression: issue #42 — concurrent tool calls lost responses due to |
| 24 | + # global stdout suppression middleware redirecting fd 1 for entire tool duration. |
| 25 | + if not integration_enabled(): |
| 26 | + pytest.skip("Set RUN_OPENSTUDIO_INTEGRATION=1 to enable MCP integration tests.") |
| 27 | + |
| 28 | + async def _run(): |
| 29 | + async with stdio_client(server_params()) as (read, write): |
| 30 | + async with ClientSession(read, write) as session: |
| 31 | + await session.initialize() |
| 32 | + |
| 33 | + # --- Arrange --- |
| 34 | + # Fire slow tool first |
| 35 | + baseline_task = asyncio.create_task( |
| 36 | + session.call_tool("create_baseline_osm", { |
| 37 | + "name": "concurrent_race_test", "num_floors": 1, |
| 38 | + }) |
| 39 | + ) |
| 40 | + # Small delay so baseline_osm enters its execution window |
| 41 | + await asyncio.sleep(0.5) |
| 42 | + |
| 43 | + # --- Act --- |
| 44 | + # Fire fast tool while slow tool holds middleware fd redirect |
| 45 | + status_task = asyncio.create_task( |
| 46 | + session.call_tool("get_server_status", {}) |
| 47 | + ) |
| 48 | + |
| 49 | + # --- Assert --- |
| 50 | + # 30s timeout: get_server_status should return in <1s. |
| 51 | + # If it times out, the race condition is present — the response |
| 52 | + # went to stderr and the client never received it. |
| 53 | + try: |
| 54 | + baseline_res, status_res = await asyncio.wait_for( |
| 55 | + asyncio.gather(baseline_task, status_task), |
| 56 | + timeout=30, |
| 57 | + ) |
| 58 | + except asyncio.TimeoutError: |
| 59 | + pytest.fail( |
| 60 | + "Concurrent tool call timed out — stdout suppression race " |
| 61 | + "condition is present (issue #42). get_server_status response " |
| 62 | + "was likely written to stderr while create_baseline_osm held " |
| 63 | + "the fd 1 redirect." |
| 64 | + ) |
| 65 | + |
| 66 | + baseline = unwrap(baseline_res) |
| 67 | + status = unwrap(status_res) |
| 68 | + |
| 69 | + assert baseline.get("ok") is True, f"create_baseline_osm failed: {baseline}" |
| 70 | + assert status.get("ok") is True, f"get_server_status failed: {status}" |
| 71 | + assert "run_root" in status, f"status missing expected keys: {status}" |
| 72 | + |
| 73 | + asyncio.run(_run()) |
| 74 | + |
| 75 | + |
| 76 | +@pytest.mark.integration |
| 77 | +def test_concurrent_fast_tools_both_respond(): |
| 78 | + # Regression: issue #42 — even two fast tools can race if both enter |
| 79 | + # the middleware's fd redirect window simultaneously. |
| 80 | + if not integration_enabled(): |
| 81 | + pytest.skip("Set RUN_OPENSTUDIO_INTEGRATION=1 to enable MCP integration tests.") |
| 82 | + |
| 83 | + async def _run(): |
| 84 | + async with stdio_client(server_params()) as (read, write): |
| 85 | + async with ClientSession(read, write) as session: |
| 86 | + await session.initialize() |
| 87 | + |
| 88 | + # Fire two fast tools concurrently |
| 89 | + task_a = asyncio.create_task( |
| 90 | + session.call_tool("get_server_status", {}) |
| 91 | + ) |
| 92 | + task_b = asyncio.create_task( |
| 93 | + session.call_tool("get_server_status", {}) |
| 94 | + ) |
| 95 | + |
| 96 | + try: |
| 97 | + res_a, res_b = await asyncio.wait_for( |
| 98 | + asyncio.gather(task_a, task_b), |
| 99 | + timeout=15, |
| 100 | + ) |
| 101 | + except asyncio.TimeoutError: |
| 102 | + pytest.fail( |
| 103 | + "Concurrent fast tool calls timed out — stdout suppression " |
| 104 | + "race condition (issue #42)." |
| 105 | + ) |
| 106 | + |
| 107 | + a = unwrap(res_a) |
| 108 | + b = unwrap(res_b) |
| 109 | + |
| 110 | + assert a.get("ok") is True, f"First get_server_status failed: {a}" |
| 111 | + assert b.get("ok") is True, f"Second get_server_status failed: {b}" |
| 112 | + |
| 113 | + asyncio.run(_run()) |
0 commit comments