Detect event-loop blocking in tests (#3510)

Author: Kludex

pyproject.toml

@@ -77,6 +77,7 @@ dev = [
     "strict-no-cover",
     "logfire>=3.0.0",
     "opentelemetry-sdk>=1.39.1",
+    "blockbuster>=1.5.27",
 ]
 docs = [
     # Zensical is the Material team's successor to MkDocs; it natively

src/mcp/client/stdio.py

@@ -18,6 +18,7 @@
 
 import anyio
 import anyio.lowlevel
+import anyio.to_thread
 import mcp_types as types
 from anyio.abc import AsyncResource, Process
 from anyio.streams.text import TextReceiveStream
@@ -120,7 +121,7 @@ async def stdio_client(
         OSError: If the server process cannot be spawned.
         ValueError: If the spawn parameters are invalid (embedded NUL bytes).
     """
-    command = _get_executable_command(server.command)
+    command = await _get_executable_command(server.command)
 
     process = await _create_platform_compatible_process(
         command=command,
@@ -317,10 +318,10 @@ def _close_subprocess_transport(process: ServerProcess) -> None:
             close()
 
 
-def _get_executable_command(command: str) -> str:
+async def _get_executable_command(command: str) -> str:
     """Normalizes the command for the current platform."""
-    if sys.platform == "win32":  # pragma: no cover
-        return get_windows_executable_command(command)
+    if sys.platform == "win32":
+        return await anyio.to_thread.run_sync(get_windows_executable_command, command, abandon_on_cancel=True)
     else:  # pragma: lax no cover
         return command
 

tests/client/test_stdio.py

@@ -14,14 +14,18 @@
 import os
 import signal
 import sys
+import threading
 from collections.abc import Callable
 from contextlib import AsyncExitStack, suppress
 from pathlib import Path
+from types import SimpleNamespace
 from typing import TextIO, cast
 
 import anyio
 import anyio.abc
+import anyio.from_thread
 import anyio.lowlevel
+import anyio.to_thread
 import pytest
 import trio
 import trio.testing
@@ -199,6 +203,9 @@ def install_fake_process(
     """
     terminated: list[FakeProcess] = []
 
+    async def fake_get_executable_command(command: str) -> str:
+        return command
+
     async def fake_spawn(
         command: str,
         args: list[str],
@@ -212,6 +219,7 @@ async def fake_terminate_tree(proc: FakeProcess) -> None:
         terminated.append(proc)
         proc.exit(-15)
 
+    monkeypatch.setattr(stdio, "_get_executable_command", fake_get_executable_command)
     monkeypatch.setattr(stdio, "_create_platform_compatible_process", fake_spawn)
     monkeypatch.setattr(stdio, "_terminate_process_tree", fake_terminate_tree)
     if grace_period is not None:
@@ -568,6 +576,45 @@ async def test_a_command_that_cannot_be_execed_raises_enoent() -> None:
     assert exc_info.value.errno == errno.ENOENT
 
 
+@pytest.mark.anyio
+async def test_cancellation_during_windows_command_resolution_returns_before_resolution_finishes(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Cancelling `stdio_client` does not wait for blocked Windows command resolution."""
+    resolution_started = anyio.Event()
+    resolution_release = threading.Event()
+    resolution_finished = threading.Event()
+
+    def blocking_resolver(command: str) -> str:
+        anyio.from_thread.run_sync(resolution_started.set)
+        resolution_release.wait()
+        resolution_finished.set()
+        return command
+
+    monkeypatch.setattr(stdio, "sys", SimpleNamespace(platform="win32"))
+    monkeypatch.setattr(stdio, "get_windows_executable_command", blocking_resolver)
+
+    cancel_scope = anyio.CancelScope()
+    client_stopped = anyio.Event()
+
+    async def run_client() -> None:
+        with cancel_scope:
+            async with AsyncExitStack() as stack:
+                await stack.enter_async_context(stdio_client(FAKE_PARAMS))
+        client_stopped.set()
+
+    with anyio.fail_after(5):
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(run_client)
+            await resolution_started.wait()
+            cancel_scope.cancel()
+            try:
+                await client_stopped.wait()
+            finally:
+                resolution_release.set()
+            await anyio.to_thread.run_sync(resolution_finished.wait)
+
+
 @pytest.mark.anyio
 async def test_cancellation_during_spawn_leaks_no_streams(monkeypatch: pytest.MonkeyPatch) -> None:
     """Cancellation while the spawn is still in flight must not leak the internal streams.

tests/conftest.py

@@ -1,7 +1,9 @@
 import os
 from collections.abc import AsyncIterator, Iterator
 
+import httpcore2 as _httpcore2
 import pytest
+from blockbuster import BlockBuster
 
 # OpenTelemetry's `set_tracer_provider` is set-once per process, so the suite
 # uses a single span-capture mechanism: logfire's `capfire` fixture (its
@@ -17,12 +19,36 @@
 
 import mcp.shared._otel  # noqa: E402
 
+# Load httpx2's lazy default transport before BlockBuster starts.
+del _httpcore2
+
 
 @pytest.fixture(scope="session")
 def anyio_backend() -> str:
     return "asyncio"
 
 
+@pytest.fixture(autouse=True)
+def blockbuster() -> Iterator[None]:
+    bb = BlockBuster(["mcp", "mcp_types"])
+    # Coverage reads source files while collecting data.
+    bb.functions["os.stat"].can_block_in("coverage/python.py", "get_python_source")
+    bb.functions["io.BufferedReader.read"].can_block_in("coverage/python.py", "read_python_source")
+    # jsonschema discovers its bundled schemas during its first import.
+    bb.functions["os.listdir"].can_block_in("/jsonschema_specifications/_core.py", "_schemas")
+    bb.functions["os.scandir"].can_block_in("/jsonschema_specifications/_core.py", "_schemas")
+    bb.functions["io.TextIOWrapper.read"].can_block_in("/jsonschema_specifications/_core.py", "_schemas")
+    # These public synchronous conversions read the media file by design.
+    bb.functions["io.BufferedReader.read"].can_block_in(
+        "mcp/server/mcpserver/utilities/types.py", ("to_image_content", "to_audio_content")
+    )
+    bb.activate()
+    try:
+        yield
+    finally:
+        bb.deactivate()
+
+
 @pytest.fixture(scope="module", autouse=True)
 async def _module_runner_lease(anyio_backend: str) -> AsyncIterator[None]:
     """Share one event loop across each module's tests instead of one per test.