modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Detect event-loop blocking in tests (#3510)
Some checks failed
Conformance Tests / server-conformance (push) Has been cancelled
Details
Conformance Tests / client-conformance (push) Has been cancelled
Details
Deploy Docs / deploy-docs (push) Has been cancelled
Details
CI / checks (push) Has been cancelled
Details
GitHub Actions Security Analysis / zizmor (push) Has been cancelled
Details
CI / all-green (push) Has been cancelled
Details

Browse source
This commit is contained in:
Marcelo Trylesinski 2026-09-16 12:12:27 +02:00 • committed by GitHub
commit 6affe5c0d3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 99 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

26
tests/conftest.py
View file

@ -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 @@ from logfire.testing import CaptureLogfire # noqa: E402
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.