Shorten stdio test comments (#3329)

This commit is contained in:
Marcelo Trylesinski 2026-08-18 10:50:47 +02:00 • committed by GitHub
commit 0d92192765
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 62 additions and 262 deletions

View file

@ -45,6 +45,8 @@ assigned to their author are closed automatically.
## Code Quality ## Code Quality
- Keep comments brief. Explain only non-obvious reasons or constraints; do not
narrate the code or restate names, types, or assertions.
- Type hints required for all code - Type hints required for all code
- Public APIs must have docstrings. When a public API raises exceptions a - Public APIs must have docstrings. When a public API raises exceptions a
caller would reasonably catch, document them in a `Raises:` section. Don't caller would reasonably catch, document them in a `Raises:` section. Don't

View file

@ -1,10 +1,4 @@
"""A real low-level Server over the stdio transport, for the suite's one subprocess test. """Low-level stdio server for the interaction suite's subprocess test."""
Runnable as `python -m tests.interaction.transports._stdio_server` from the repo root; the test
launches it that way via `stdio_client`. Kept separate from the test module so the server lives in
its own importable file (subprocess coverage applies) while the test file follows the suite's
test-only-functions convention.
"""
import sys import sys
import warnings import warnings
@ -63,26 +57,13 @@ with warnings.catch_warnings():
async def main() -> None: async def main() -> None:
async with stdio_server() as (read_stream, write_stream): async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options()) await server.run(read_stream, write_stream, server.create_initialization_options())
# Flush this process's coverage data before the clean-exit line below. Without this, the # Save subprocess coverage before the marker so forced teardown cannot lose it.
# data is only written by coverage's atexit hook during interpreter teardown -- and on a
# slow Windows runner that can overrun the transport's termination grace, so the kill
# silently destroys the data file and the 100% gate trips on this module's subprocess-only
# lines. Saving here puts the write before the line the test synchronizes on: once the
# parent has seen "clean exit", the data is durably on disk and the escalation is harmless.
# Nothing measured may execute after the save (it would be unrecordable by construction),
# hence the excluded lines below. The branch is pragma'd because under coverage the
# instance always exists, and without coverage nothing is measured anyway.
cov = getattr(coverage.process_startup, "coverage", None) cov = getattr(coverage.process_startup, "coverage", None)
if cov is not None: # pragma: no branch if cov is not None: # pragma: no branch
# stop() is load-bearing twice over: it ends tracing, making itself the last # Leave nothing for coverage's atexit hook to rewrite if teardown is interrupted.
# recordable line, and it leaves nothing new for coverage's atexit re-save to flush --
# so a kill landing during interpreter teardown cannot corrupt the file save() wrote
# (coverage opens it with sqlite journaling off; a torn rewrite would not roll back).
cov.stop() cov.stop()
cov.save() # pragma: lax no cover - untraced: stop() above already ended measurement cov.save() # pragma: lax no cover - untraced: stop() above already ended measurement
# Reached only when the run loop exits because stdin closed; if the process were terminated # The test uses this marker to distinguish clean exit from termination.
# the test's stderr capture would not see this line. lax no cover: runs after the coverage
# save by design, so it can never appear covered.
print("stdio-echo: clean exit", file=sys.stderr, flush=True) # pragma: lax no cover print("stdio-echo: clean exit", file=sys.stderr, flush=True) # pragma: lax no cover

View file

@ -1,13 +1,6 @@
"""The stdio transport: one subprocess end-to-end test and one in-process framing test. """Stdio subprocess round-trip and in-process framing tests.
The subprocess test proves the client-server round trip over the transport's real process Lifecycle edge cases remain in `tests/client/test_stdio.py`.
boundary; its server lives in `_stdio_server.py` and is launched via `python -m` so subprocess
coverage measurement applies. The framing test drives `stdio_server` over injected in-process
streams instead.
stdio is deliberately not a leg of the `connect`-fixture matrix: a subprocess per test would be
slow, and the matrix already proves transport-agnosticism in-process. Process-lifecycle edge
cases (terminate/kill escalation, parse errors) stay in `tests/client/test_stdio.py`.
""" """
import io import io
@ -50,20 +43,8 @@ _REPO_ROOT = Path(__file__).parents[3]
async def test_tool_call_and_notification_round_trip_over_a_stdio_subprocess( async def test_tool_call_and_notification_round_trip_over_a_stdio_subprocess(
monkeypatch: pytest.MonkeyPatch, monkeypatch: pytest.MonkeyPatch,
) -> None: ) -> None:
"""A stdio-subprocess Client round-trips a tool call, a notification, and a clean exit. """A stdio client round-trips a tool call and notification before clean exit."""
# Allow slow Windows runners to flush subprocess coverage before escalation.
The Client initializes, calls a tool with arguments, and receives the server's log
notification before the call returns; the server exits when the transport closes its
stdin.
"""
# After stdin closes, the child must unwind, flush its subprocess coverage data, and write
# the clean-exit line before escalation (the server saves coverage *before* printing, so a
# post-print kill can no longer silently lose the data file -- see _stdio_server.main). The
# production 2s default is too tight for the unwind+save tail on loaded Windows runners
# (measured in-situ p99 of the whole test is ~7s); a kill before the print fails the stderr
# assertion below loudly rather than tripping the coverage gate. The 20s grace covers even a
# badly starved runner (a >10s stall has been seen once in CI) and costs nothing when the
# child exits promptly. Not under test.
monkeypatch.setattr(stdio, "PROCESS_TERMINATION_TIMEOUT", 20.0) monkeypatch.setattr(stdio, "PROCESS_TERMINATION_TIMEOUT", 20.0)
received: list[LoggingMessageNotificationParams] = [] received: list[LoggingMessageNotificationParams] = []
@ -77,10 +58,7 @@ async def test_tool_call_and_notification_round_trip_over_a_stdio_subprocess(
command=sys.executable, command=sys.executable,
args=["-m", _stdio_server.__name__], args=["-m", _stdio_server.__name__],
cwd=str(_REPO_ROOT), cwd=str(_REPO_ROOT),
# stdio_client filters the inherited environment, dropping the variables # Preserve subprocess coverage and suppress anyio's `SyntaxWarning` on Python 3.14.
# coverage.py's subprocess support uses; pass them through so the server module is
# measured. PYTHONWARNINGS: the child recompiles anyio (pytest's pyc tag differs),
# and on 3.14 anyio's return-in-finally SyntaxWarning would land on the snapshot stderr.
env={key: value for key, value in os.environ.items() if key.startswith("COVERAGE_")} env={key: value for key, value in os.environ.items() if key.startswith("COVERAGE_")}
| {"PYTHONWARNINGS": "ignore::SyntaxWarning"}, | {"PYTHONWARNINGS": "ignore::SyntaxWarning"},
), ),
@ -98,26 +76,18 @@ async def test_tool_call_and_notification_round_trip_over_a_stdio_subprocess(
captured_stderr = errlog.read() captured_stderr = errlog.read()
assert result == snapshot(CallToolResult(content=[TextContent(text="across\nprocesses")])) assert result == snapshot(CallToolResult(content=[TextContent(text="across\nprocesses")]))
# stdio carries one ordered server-to-client stream, so the same notification-before-response # Stdio preserves notification-before-response ordering.
# guarantee holds here as for the in-memory transport.
assert received == snapshot( assert received == snapshot(
[LoggingMessageNotificationParams(level="info", logger="echo", data="echoing across\nprocesses")] [LoggingMessageNotificationParams(level="info", logger="echo", data="echoing across\nprocesses")]
) )
# The server writes this line only after its run loop returns on stdin close: seeing it proves # The marker distinguishes clean exit from termination.
# a self-exit, not the terminate escalation. The capture itself proves stderr passthrough.
assert captured_stderr == snapshot("stdio-echo: clean exit\n") assert captured_stderr == snapshot("stdio-echo: clean exit\n")
@requirement("transport:stdio:stream-purity") @requirement("transport:stdio:stream-purity")
@requirement("transport:stdio:no-embedded-newlines") @requirement("transport:stdio:no-embedded-newlines")
async def test_stdio_server_writes_one_jsonrpc_message_per_line() -> None: async def test_stdio_server_writes_one_jsonrpc_message_per_line() -> None:
"""Every `stdio_server` write is one valid JSON-RPC message on its own line. """Each `stdio_server` write is one newline-terminated JSON-RPC message."""
Each line is newline-terminated with payload newlines JSON-escaped. This proves the
transport's own framing over injected streams; the descriptor-level guard that keeps
handler code off the wire is pinned by tests/server/test_stdio.py (see the narrowed
divergence on `transport:stdio:stream-purity`).
"""
captured = io.StringIO() captured = io.StringIO()
sent_line = json.dumps(initialize_body(request_id=1)) + "\n" sent_line = json.dumps(initialize_body(request_id=1)) + "\n"
@ -148,7 +118,5 @@ async def test_stdio_server_writes_one_jsonrpc_message_per_line() -> None:
assert len(lines) == 2 assert len(lines) == 2
messages = [jsonrpc_message_adapter.validate_json(line) for line in lines] messages = [jsonrpc_message_adapter.validate_json(line) for line in lines]
assert [type(message).__name__ for message in messages] == snapshot(["JSONRPCResponse", "JSONRPCNotification"]) assert [type(message).__name__ for message in messages] == snapshot(["JSONRPCResponse", "JSONRPCNotification"])
# The newline inside the payload is JSON-escaped on the wire, not a literal newline that would
# break the one-message-per-line framing.
assert r"line\nbreak" in lines[0] assert r"line\nbreak" in lines[0]
assert r"two\nlines" in lines[1] assert r"two\nlines" in lines[1]

View file

@ -1,13 +1,4 @@
"""Real-subprocess stdio lifecycle tests that hold on both POSIX and Windows. """Cross-platform stdio lifecycle tests using real subprocesses."""
The `stdio_client` tests each launch a real server through the public API and pin
one lifecycle behaviour, with kernel-level liveness sockets as the only
synchronization; the `FallbackProcess` tests wrap a raw `subprocess.Popen`
directly. Platform-divergent shutdown policy lives in test_posix.py /
test_windows.py; the full protocol round trip is pinned by
tests/interaction/transports/test_stdio.py and in-process shutdown logic by
tests/client/test_stdio.py.
"""
import os import os
import subprocess import subprocess
@ -39,16 +30,11 @@ async def test_a_server_that_exits_on_stdin_close_is_reaped_and_never_terminated
spawned_processes: list[anyio.abc.Process | FallbackProcess], spawned_processes: list[anyio.abc.Process | FallbackProcess],
terminate_calls: list[anyio.abc.Process | FallbackProcess], terminate_calls: list[anyio.abc.Process | FallbackProcess],
) -> None: ) -> None:
"""The happy path: closing stdin alone shuts a well-behaved server down. """Closing stdin reaps a well-behaved server without escalation."""
The server exits with code 0 and the escalation seam is never invoked.
"""
async with AsyncExitStack() as stack: async with AsyncExitStack() as stack:
sock, port = await open_liveness_listener() sock, port = await open_liveness_listener()
stack.push_async_callback(sock.aclose) stack.push_async_callback(sock.aclose)
# The server exits on its own at stdin EOF -- the well-behaved response
# to shutdown's first step.
server = ( server = (
f"import socket, sys\n" f"import socket, sys\n"
f"s = socket.create_connection(('127.0.0.1', {port}))\n" f"s = socket.create_connection(('127.0.0.1', {port}))\n"
@ -57,8 +43,7 @@ async def test_a_server_that_exits_on_stdin_close_is_reaped_and_never_terminated
) )
params = StdioServerParameters(command=sys.executable, args=["-c", server]) params = StdioServerParameters(command=sys.executable, args=["-c", server])
# The bound covers one interpreter cold start on a loaded runner; a healthy # Allow one cold interpreter start on loaded CI.
# run takes well under a second.
with anyio.fail_after(10.0): with anyio.fail_after(10.0):
async with stdio_client(params): async with stdio_client(params):
stream = await accept_alive(sock) stream = await accept_alive(sock)
@ -76,11 +61,7 @@ async def test_cancelling_the_client_mid_session_terminates_the_whole_server_tre
spawned_processes: list[anyio.abc.Process | FallbackProcess], spawned_processes: list[anyio.abc.Process | FallbackProcess],
terminate_calls: list[anyio.abc.Process | FallbackProcess], terminate_calls: list[anyio.abc.Process | FallbackProcess],
) -> None: ) -> None:
"""Cancellation still runs the full shutdown against a real process tree. """Cancellation terminates a server tree that ignores stdin closure."""
Cancellation here stands in for a client timeout or app shutdown: a server that
ignores stdin closure is escalated against, and its child dies with it.
"""
monkeypatch.setattr(stdio, "PROCESS_TERMINATION_TIMEOUT", 0.2) monkeypatch.setattr(stdio, "PROCESS_TERMINATION_TIMEOUT", 0.2)
async with AsyncExitStack() as stack: async with AsyncExitStack() as stack:
@ -88,18 +69,13 @@ async def test_cancelling_the_client_mid_session_terminates_the_whole_server_tre
stack.push_async_callback(sock.aclose) stack.push_async_callback(sock.aclose)
child = connect_back_script(port) child = connect_back_script(port)
# The parent never reads stdin and blocks forever, so only the escalation
# can end it -- which cancellation must not skip.
parent = f"import subprocess, sys\nsubprocess.Popen([sys.executable, '-c', {child!r}])\n" + connect_back_script( parent = f"import subprocess, sys\nsubprocess.Popen([sys.executable, '-c', {child!r}])\n" + connect_back_script(
port port
) )
params = StdioServerParameters(command=sys.executable, args=["-c", parent]) params = StdioServerParameters(command=sys.executable, args=["-c", parent])
entered = anyio.Event() entered = anyio.Event()
# Cancel a scope owned by the client's task, not the test's task group: a # A child-task scope avoids a CPython 3.11 coverage tracing bug during host self-cancellation.
# host self-cancel is delivered by throwing through this test function's
# suspended frames, and Python 3.11's tracer loses coverage events after
# such a throw() traversal (python/cpython#106749).
cancel_scope = anyio.CancelScope() cancel_scope = anyio.CancelScope()
async def run_client_until_cancelled() -> None: async def run_client_until_cancelled() -> None:
@ -109,8 +85,7 @@ async def test_cancelling_the_client_mid_session_terminates_the_whole_server_tre
await anyio.sleep_forever() await anyio.sleep_forever()
streams: list[anyio.abc.SocketStream] = [] streams: list[anyio.abc.SocketStream] = []
# The bound covers two interpreter cold starts on a loaded runner plus the # Allow two cold interpreter starts and the shortened escalation wait.
# shortened escalation wait; a healthy run takes around a second.
with anyio.fail_after(10.0): with anyio.fail_after(10.0):
async with anyio.create_task_group() as tg: async with anyio.create_task_group() as tg:
tg.start_soon(run_client_until_cancelled) tg.start_soon(run_client_until_cancelled)
@ -132,11 +107,7 @@ async def test_a_server_that_exits_mid_session_keeps_its_own_exit_code(
spawned_processes: list[anyio.abc.Process | FallbackProcess], spawned_processes: list[anyio.abc.Process | FallbackProcess],
terminate_calls: list[anyio.abc.Process | FallbackProcess], terminate_calls: list[anyio.abc.Process | FallbackProcess],
) -> None: ) -> None:
"""A server that dies on its own mid-session is reaped with the exit code it chose. """A server that dies mid-session retains its exit code without escalation."""
The client surfaces the child's true status rather than synthesizing one, and
the escalation seam confirms nothing was terminated along the way.
"""
async with AsyncExitStack() as stack: async with AsyncExitStack() as stack:
sock, port = await open_liveness_listener() sock, port = await open_liveness_listener()
stack.push_async_callback(sock.aclose) stack.push_async_callback(sock.aclose)
@ -149,14 +120,12 @@ async def test_a_server_that_exits_mid_session_keeps_its_own_exit_code(
) )
params = StdioServerParameters(command=sys.executable, args=["-c", server]) params = StdioServerParameters(command=sys.executable, args=["-c", server])
# The bound covers one interpreter cold start on a loaded runner; a healthy # Allow one cold interpreter start on loaded CI.
# run takes well under a second.
with anyio.fail_after(10.0): with anyio.fail_after(10.0):
# no branch: coverage mis-traces the exit arcs of a nested `async with` on 3.11+. # Coverage mis-traces nested `async with` exit arcs on Python 3.11+.
async with stdio_client(params): # pragma: no branch async with stdio_client(params): # pragma: no branch
stream = await accept_alive(sock) stream = await accept_alive(sock)
stack.push_async_callback(stream.aclose) stack.push_async_callback(stream.aclose)
# The server is already gone before shutdown begins.
await assert_stream_closed(stream) await assert_stream_closed(stream)
assert spawned_processes[0].returncode == 7 assert spawned_processes[0].returncode == 7
@ -168,11 +137,7 @@ async def test_server_stderr_output_reaches_the_errlog_file(
tmp_path: Path, tmp_path: Path,
spawned_processes: list[anyio.abc.Process | FallbackProcess], spawned_processes: list[anyio.abc.Process | FallbackProcess],
) -> None: ) -> None:
"""What the server writes to stderr lands in the file passed as `errlog`. """Server stderr reaches the file passed as `errlog`."""
The spawn hands over errlog's file descriptor as the child's stderr, so it must
be a real file -- an in-memory StringIO has no fileno.
"""
marker = "stdio-lifecycle stderr marker 4242" marker = "stdio-lifecycle stderr marker 4242"
async with AsyncExitStack() as stack: async with AsyncExitStack() as stack:
@ -190,15 +155,12 @@ async def test_server_stderr_output_reaches_the_errlog_file(
params = StdioServerParameters(command=sys.executable, args=["-c", server]) params = StdioServerParameters(command=sys.executable, args=["-c", server])
with (tmp_path / "errlog.txt").open("w+", encoding="utf-8") as errlog: with (tmp_path / "errlog.txt").open("w+", encoding="utf-8") as errlog:
# The bound covers one interpreter cold start on a loaded runner; a # Allow one cold interpreter start on loaded CI.
# healthy run takes well under a second.
with anyio.fail_after(10.0): with anyio.fail_after(10.0):
async with stdio_client(params, errlog=errlog): async with stdio_client(params, errlog=errlog):
stream = await accept_alive(sock) stream = await accept_alive(sock)
stack.push_async_callback(stream.aclose) stack.push_async_callback(stream.aclose)
# The server exited on stdin EOF, so every stderr write it made has
# reached the file descriptor.
errlog.seek(0) errlog.seek(0)
content = errlog.read() content = errlog.read()
@ -212,15 +174,9 @@ async def test_server_stderr_output_reaches_the_errlog_file(
# lax no cover: Windows runners enforce 100% per job but lack os.waitid and skip this # lax no cover: Windows runners enforce 100% per job but lack os.waitid and skip this
# test; test_windows.py's SelectorEventLoop lifecycle test exercises the property there. # test; test_windows.py's SelectorEventLoop lifecycle test exercises the property there.
def test_fallback_process_reports_death_through_returncode_without_a_wait_call() -> None: # pragma: lax no cover def test_fallback_process_reports_death_through_returncode_without_a_wait_call() -> None: # pragma: lax no cover
"""`FallbackProcess.returncode` observes process death on its own. """`FallbackProcess.returncode` observes death without calling `wait()`.
Pre-fix it returned Popen's cached value, which stays None until someone calls wait()/poll(). `waitid(WNOWAIT)` avoids priming Popen's cached return code or reaping the child.
`os.waitid(WEXITED | WNOWAIT)` waits for the child to become reapable without
reaping it or priming Popen's cache (which would mask the regression); the
pre-fix cached read would still see None here. stdout EOF is NOT such a signal:
the kernel closes the pipes before the exit status is published, so an
EOF-then-assert version flakes.
""" """
popen = subprocess.Popen( popen = subprocess.Popen(
[sys.executable, "-c", "pass"], [sys.executable, "-c", "pass"],
@ -236,40 +192,31 @@ def test_fallback_process_reports_death_through_returncode_without_a_wait_call()
finally: finally:
popen.stdin.close() popen.stdin.close()
popen.stdout.close() popen.stdout.close()
# The WNOWAIT above left the child unreaped; reap it so no zombie (and no # Reap the child left by `WNOWAIT`.
# Popen ResourceWarning) outlives the test.
popen.wait() popen.wait()
@pytest.mark.anyio @pytest.mark.anyio
async def test_fallback_process_wait_is_cancellable_while_the_child_lives() -> None: async def test_fallback_process_wait_is_cancellable_while_the_child_lives() -> None:
"""`FallbackProcess.wait()` honours cancellation while the child is still running. """`FallbackProcess.wait()` remains cancellable while the child runs."""
Pre-fix it parked `Popen.wait()` in a worker thread anyio will not abandon,
which blocks every cancellation aimed at it. Runs everywhere: the wrapper holds
a plain Popen.
"""
popen = subprocess.Popen( popen = subprocess.Popen(
[sys.executable, "-c", "import sys; sys.stdin.read()"], [sys.executable, "-c", "import sys; sys.stdin.read()"],
stdin=subprocess.PIPE, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
) )
assert popen.stdin is not None and popen.stdout is not None assert popen.stdin is not None and popen.stdout is not None
# Pre-fix, no timeout below can fire while the worker thread is parked in # The watchdog converts a blocked worker thread into a failure.
# Popen.wait(); killing the child turns that regression's hang into a clean failure.
watchdog = threading.Timer(8.0, popen.kill) watchdog = threading.Timer(8.0, popen.kill)
watchdog.start() watchdog.start()
try: try:
process = FallbackProcess(popen) process = FallbackProcess(popen)
# move_on_after's short deadline is the time-based feature under test -- # The short deadline is the cancellability behavior under test.
# cancellability -- not a wait for an async condition.
with anyio.fail_after(5): with anyio.fail_after(5):
with anyio.move_on_after(0.1) as scope: with anyio.move_on_after(0.1) as scope:
await process.wait() await process.wait()
assert scope.cancelled_caught assert scope.cancelled_caught
# Only the wait was cancelled; the child itself is untouched.
assert popen.poll() is None assert popen.poll() is None
finally: finally:
watchdog.cancel() watchdog.cancel()
@ -281,11 +228,7 @@ async def test_fallback_process_wait_is_cancellable_while_the_child_lives() -> N
@pytest.mark.anyio @pytest.mark.anyio
async def test_a_tool_spawned_childs_stdout_writes_never_reach_the_wire(tmp_path: Path) -> None: async def test_a_tool_spawned_childs_stdout_writes_never_reach_the_wire(tmp_path: Path) -> None:
"""A child writing to its inherited stdout pollutes the server's stderr, never the protocol. """A child's inherited stdout reaches server stderr, not the protocol."""
Pre-isolation the junk line landed in the JSON-RPC stream (fails on base);
fd 1 has exactly one target, so stderr delivery proves the wire never saw it.
"""
server = dedent( server = dedent(
""" """
import subprocess, sys import subprocess, sys
@ -295,7 +238,6 @@ async def test_a_tool_spawned_childs_stdout_writes_never_reach_the_wire(tmp_path
@mcp.tool() @mcp.tool()
def run_noisy_child() -> str: def run_noisy_child() -> str:
# No redirection: the child inherits the server's stdout.
proc = subprocess.run([sys.executable, "-c", "print('this is not json')"], timeout=20) proc = subprocess.run([sys.executable, "-c", "print('this is not json')"], timeout=20)
return str(proc.returncode) return str(proc.returncode)
@ -305,7 +247,7 @@ async def test_a_tool_spawned_childs_stdout_writes_never_reach_the_wire(tmp_path
with (tmp_path / "server-stderr.txt").open("w+", encoding="utf-8") as errlog: with (tmp_path / "server-stderr.txt").open("w+", encoding="utf-8") as errlog:
transport = stdio_client(StdioServerParameters(command=sys.executable, args=["-c", server]), errlog=errlog) transport = stdio_client(StdioServerParameters(command=sys.executable, args=["-c", server]), errlog=errlog)
# Bound covers three interpreter cold starts; a regressed Windows leg hangs rather than corrupts. # Allow three cold interpreter starts.
with anyio.fail_after(40): with anyio.fail_after(40):
async with Client(transport) as client: async with Client(transport) as client:
result = await client.call_tool("run_noisy_child") result = await client.call_tool("run_noisy_child")

View file

@ -1,8 +1,6 @@
"""POSIX-only stdio lifecycle tests: a gracefully-exited server's children survive the client shutdown. """POSIX stdio tests for children of gracefully exited servers.
SDK-defined policy, not spec-mandated (docs/migration.md, "`stdio_client` no Unlike Windows, POSIX leaves these children running after client shutdown.
longer kills children of a gracefully-exited server on POSIX"). Windows has the
opposite documented outcome; see tests/transports/stdio/test_windows.py.
""" """
import errno import errno
@ -31,57 +29,40 @@ async def test_a_gracefully_exiting_servers_child_survives_the_client_shutdown(
spawned_processes: list[anyio.abc.Process | FallbackProcess], spawned_processes: list[anyio.abc.Process | FallbackProcess],
terminate_calls: list[anyio.abc.Process | FallbackProcess], terminate_calls: list[anyio.abc.Process | FallbackProcess],
) -> None: ) -> None:
"""A server that exits on stdin closure keeps its background child running after `stdio_client` returns. """A server that exits on stdin closure leaves its background child running.
The client never escalates against the gracefully-exited server. SDK-defined This SDK policy intentionally differs from Windows.
policy per docs/migration.md; regression for the pre-fix client that
tree-killed the child. The Windows twin in test_windows.py pins the opposite outcome.
""" """
sock, port = await open_liveness_listener() sock, port = await open_liveness_listener()
async with sock: async with sock:
child = connect_back_script(port, echo=True) child = connect_back_script(port, echo=True)
# The server hands its inherited pipes to a child, then exits as soon as
# its stdin closes: the well-behaved graceful path.
server = f"import subprocess, sys\nsubprocess.Popen([sys.executable, '-c', {child!r}])\nsys.stdin.read()\n" server = f"import subprocess, sys\nsubprocess.Popen([sys.executable, '-c', {child!r}])\nsys.stdin.read()\n"
params = StdioServerParameters(command=sys.executable, args=["-c", server]) params = StdioServerParameters(command=sys.executable, args=["-c", server])
# Two interpreter cold starts on a loaded runner; healthy runs take ~0.3s. # Allow two cold interpreter starts on loaded CI.
with anyio.fail_after(10.0): with anyio.fail_after(10.0):
async with stdio_client(params): async with stdio_client(params):
child_stream = await accept_alive(sock) child_stream = await accept_alive(sock)
async with child_stream: async with child_stream:
# Only a live process answers an echo: the child survived shutdown.
await assert_peer_echoes(child_stream) await assert_peer_echoes(child_stream)
# A FIN-shaped probe cannot tell graceful exit from a kill; the seam can:
# no escalation was invoked, and the leader exited 0 on stdin closure.
assert terminate_calls == [] assert terminate_calls == []
leader = spawned_processes[0] leader = spawned_processes[0]
assert leader.returncode == 0 assert leader.returncode == 0
# The child is deliberately left running; the spawned_processes teardown # The fixture reaps the intentionally surviving child.
# SIGKILLs the spawn-time process group to reap it.
@pytest.mark.anyio @pytest.mark.anyio
@pytest.mark.usefixtures("spawned_processes") # failure-path safety net for the parked child @pytest.mark.usefixtures("spawned_processes") # failure-path safety net for the parked child
# lax no cover: same Windows-runner coverage-gate reason as above. # lax no cover: same Windows-runner coverage-gate reason as above.
async def test_a_surviving_childs_write_to_the_inherited_stdout_fails_with_epipe() -> None: # pragma: lax no cover async def test_a_surviving_childs_write_to_the_inherited_stdout_fails_with_epipe() -> None: # pragma: lax no cover
"""A surviving child writing to the stdout pipe it inherited from the server gets EPIPE once the client is gone. """A surviving child's inherited stdout fails with `EPIPE` after client shutdown.
The pipe's only read end was the client's, and shutdown closed it The child waits for shutdown, writes to fd 1, then reports errno over its socket.
deterministically rather than at GC time. Pins the docs/migration.md claim
"a surviving child that keeps writing to an inherited stdout receives
EPIPE/SIGPIPE once the client is gone" (SDK-defined).
Steps: the server hands its stdio pipes to a child and exits on stdin closure;
the child parks on its socket until `stdio_client` has fully exited (so the
write cannot race transport teardown), then writes one byte to its inherited
fd 1 and reports the errno (0 on success) back over the socket.
""" """
sock, port = await open_liveness_listener() sock, port = await open_liveness_listener()
async with sock: async with sock:
# Pin SIGPIPE to SIG_IGN explicitly (CPython already starts that way) so # Ignore SIGPIPE so the write reports `EPIPE`.
# the write fails with EPIPE instead of relying on interpreter startup details.
child = ( child = (
f"import os, signal, socket\n" f"import os, signal, socket\n"
f"signal.signal(signal.SIGPIPE, signal.SIG_IGN)\n" f"signal.signal(signal.SIGPIPE, signal.SIG_IGN)\n"
@ -98,16 +79,13 @@ async def test_a_surviving_childs_write_to_the_inherited_stdout_fails_with_epipe
server = f"import subprocess, sys\nsubprocess.Popen([sys.executable, '-c', {child!r}])\nsys.stdin.read()\n" server = f"import subprocess, sys\nsubprocess.Popen([sys.executable, '-c', {child!r}])\nsys.stdin.read()\n"
params = StdioServerParameters(command=sys.executable, args=["-c", server]) params = StdioServerParameters(command=sys.executable, args=["-c", server])
# Two interpreter cold starts on a loaded runner; healthy runs take ~0.3s. # Allow two cold interpreter starts on loaded CI.
with anyio.fail_after(10.0): with anyio.fail_after(10.0):
async with stdio_client(params): async with stdio_client(params):
child_stream = await accept_alive(sock) child_stream = await accept_alive(sock)
async with child_stream: async with child_stream:
# The context has fully exited: the transport, and with it the
# pipe's only read end, is closed. Release the child's write.
await child_stream.send(b"go") await child_stream.send(b"go")
# The child sends its errno report and exits, so read to EOF: the # Read the complete errno report.
# complete reply is everything before the kernel's FIN.
reply = b"" reply = b""
with suppress(anyio.EndOfStream): with suppress(anyio.EndOfStream):
while True: while True:

View file

@ -1,14 +1,6 @@
"""Windows-only stdio lifecycle behaviors, against real subprocesses. """Windows stdio tests for Job Object cleanup, selector fallback, and CRLF framing.
Each test pins a contract that exists only on Windows: Job-Object reaping of a The test bodies are excluded because non-Windows CI also enforces coverage.
gracefully-exited server's children (the deliberate divergence from the POSIX
policy in test_posix.py), the SelectorEventLoop fallback wrapper, and the CRLF
line endings a native text-mode server emits. Synchronization is kernel-level
only (liveness sockets); see `_liveness`.
Per-test no-cover pragmas (as in tests/issues/test_552_windows_hang.py): bodies run
only on windows-latest CI legs, the per-job 100% gate would count them uncovered on
non-Windows runners, and strict-no-cover is skipped on Windows where they execute.
""" """
import asyncio import asyncio
@ -49,43 +41,17 @@ async def test_a_gracefully_exited_servers_child_is_reaped_when_the_job_handle_c
spawned_processes: list[anyio.abc.Process | FallbackProcess], spawned_processes: list[anyio.abc.Process | FallbackProcess],
terminate_calls: list[anyio.abc.Process | FallbackProcess], terminate_calls: list[anyio.abc.Process | FallbackProcess],
) -> None: ) -> None:
"""A gracefully-exited server's child is killed deterministically when shutdown closes the job handle. """Closing a gracefully exited server's Job Object reaps its child.
The server exits cleanly on stdin closure, leaving a child behind; shutdown's This differs from POSIX. Empty `terminate_calls` distinguishes cleanup from escalation.
close of the server's Job Object handle (`close_process_job` +
`JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE`) kills that child deterministically, not at
GC time. Documented divergence from POSIX (docs/migration.md; the POSIX twin is
test_posix.py::test_a_gracefully_exiting_servers_child_survives_the_client_shutdown).
`terminate_calls == []` is the load-bearing distinction: the child died through
the graceful path's job-handle close, not the escalation's `TerminateJobObject`;
the two kills are indistinguishable on the socket.
Both processes connect back and their stderr is captured via `errlog`, so a
timeout failure can report which process never showed and the child's fate
(xdist swallows subprocess stderr on CI).
""" """
async with AsyncExitStack() as stack: async with AsyncExitStack() as stack:
sock, port = await open_liveness_listener() sock, port = await open_liveness_listener()
stack.push_async_callback(sock.aclose) stack.push_async_callback(sock.aclose)
# The startup marker (and any child traceback, via stderr=sys.stderr below) # Capture startup failures that xdist would otherwise hide.
# lands in errlog, splitting "never started" from "started but never connected".
child = "import sys\nprint('child-started', file=sys.stderr, flush=True)\n" + connect_back_script(port) child = "import sys\nprint('child-started', file=sys.stderr, flush=True)\n" + connect_back_script(port)
# The server spawns a child, connects back itself, then exits as soon as # Job membership is inherited; DEVNULL avoids a Windows CPython startup deadlock.
# its stdin closes: the graceful path, so the escalation never runs.
# The child inherits Job membership: the SDK assigns the server to the Job
# synchronously after spawn, long before the cold-starting interpreter can
# Popen the child (membership is inherited at CreateProcess, never
# acquired retroactively).
#
# The child's stdin must be DEVNULL: CPython startup queries fd 0, and
# Windows serializes that query behind the server's pending blocking
# `sys.stdin.read()` on the inherited pipe, so the child would freeze at
# interpreter startup until the next inbound byte or EOF.
#
# After stdin EOF ends the server, it reports the child's `poll()` status:
# `None` means alive at server exit; an exit/NTSTATUS code names the killer.
server = ( server = (
f"import socket, subprocess, sys\n" f"import socket, subprocess, sys\n"
f"try:\n" f"try:\n"
@ -111,8 +77,7 @@ async def test_a_gracefully_exited_servers_child_is_reaped_when_the_job_handle_c
spawn_started = anyio.current_time() spawn_started = anyio.current_time()
entered_at: float | None = None entered_at: float | None = None
try: try:
# Two interpreter cold starts on a loaded runner; healthy runs # Allow two cold interpreter starts on loaded CI.
# take well under a second.
with anyio.fail_after(15.0): with anyio.fail_after(15.0):
async with stdio_client(server_params, errlog=errlog): async with stdio_client(server_params, errlog=errlog):
entered_at = anyio.current_time() entered_at = anyio.current_time()
@ -123,9 +88,6 @@ async def test_a_gracefully_exited_servers_child_is_reaped_when_the_job_handle_c
stack.push_async_callback(stream.aclose) stack.push_async_callback(stream.aclose)
streams.append(stream) streams.append(stream)
except TimeoutError: except TimeoutError:
# `stdio_client.__aexit__` has already completed its shielded shutdown,
# so the stderr read carries the server's final `child-rc` line, not a
# mid-flight snapshot.
missing_leg = "the server never ran its connect line" if not streams else "the child never connected" missing_leg = "the server never ran its connect line" if not streams else "the child never connected"
spawn_split = ( spawn_split = (
"the context never entered" "the context never entered"
@ -137,12 +99,7 @@ async def test_a_gracefully_exited_servers_child_is_reaped_when_the_job_handle_c
f"{spawn_split}; server stderr: {server_stderr()!r}" f"{spawn_split}; server stderr: {server_stderr()!r}"
) )
# Context exit closed the job handle: KILL_ON_JOB_CLOSE killed the # Keep references alive so GC cannot close the weak-keyed Job Object early.
# child and the server exited gracefully, so both sockets close.
# The `spawned_processes` strong reference is load-bearing: `_process_jobs`
# is weak-keyed, so without it a GC between context exit and this assert
# could close the job handle itself and mask a regression in the
# deterministic close.
try: try:
for stream in streams: for stream in streams:
await assert_stream_closed(stream) await assert_stream_closed(stream)
@ -150,38 +107,24 @@ async def test_a_gracefully_exited_servers_child_is_reaped_when_the_job_handle_c
pytest.fail(f"a socket stayed open after shutdown; server stderr: {server_stderr()!r}") pytest.fail(f"a socket stayed open after shutdown; server stderr: {server_stderr()!r}")
leader = spawned_processes[0] leader = spawned_processes[0]
# The graceful path: the server exited on stdin closure with code 0,
# and the tree-termination escalation was never invoked.
assert leader.returncode == 0, server_stderr() assert leader.returncode == 0, server_stderr()
assert terminate_calls == [], server_stderr() assert terminate_calls == [], server_stderr()
# Overrides the suite-wide anyio_backend fixture for this test only: a selector # A selector loop forces `stdio_client` to use `FallbackProcess`.
# event loop cannot run asyncio subprocesses, forcing stdio_client onto FallbackProcess.
@pytest.mark.parametrize("anyio_backend", [("asyncio", {"loop_factory": asyncio.SelectorEventLoop})]) @pytest.mark.parametrize("anyio_backend", [("asyncio", {"loop_factory": asyncio.SelectorEventLoop})])
async def test_a_selector_event_loop_session_uses_the_fallback_process_and_exits_cleanly( # pragma: no cover async def test_a_selector_event_loop_session_uses_the_fallback_process_and_exits_cleanly( # pragma: no cover
spawned_processes: list[anyio.abc.Process | FallbackProcess], spawned_processes: list[anyio.abc.Process | FallbackProcess],
terminate_calls: list[anyio.abc.Process | FallbackProcess], terminate_calls: list[anyio.abc.Process | FallbackProcess],
) -> None: ) -> None:
"""Under a `SelectorEventLoop`, `stdio_client` falls back to `FallbackProcess` and still exits cleanly. """`stdio_client` uses `FallbackProcess` under `SelectorEventLoop`.
A selector event loop has no asyncio subprocess support, so `stdio_client` This covers clean exit because forced shutdown can strand the fallback reader thread.
falls back to the Popen-based `FallbackProcess` wrapper; a well-behaved server
still completes the full clean lifecycle: spawn, liveness, exit on stdin
closure, reaped, never escalated against.
The `isinstance` check is the engagement proof: if a future anyio gains selector
subprocess support, the spawn would silently return a normal Process. A hang here
most likely means the known fallback hazard documented in `stdio_client`'s
shutdown comment (reader thread parked in a synchronous `ReadFile`), which is
why this test pins only the clean-exit path, never a kill path.
""" """
async with AsyncExitStack() as stack: async with AsyncExitStack() as stack:
sock, port = await open_liveness_listener() sock, port = await open_liveness_listener()
stack.push_async_callback(sock.aclose) stack.push_async_callback(sock.aclose)
# Connect back for liveness, then exit as soon as stdin closes: the
# well-behaved server, so shutdown's first step suffices.
server = ( server = (
f"import socket, sys\n" f"import socket, sys\n"
f"s = socket.create_connection(('127.0.0.1', {port}))\n" f"s = socket.create_connection(('127.0.0.1', {port}))\n"
@ -190,35 +133,24 @@ async def test_a_selector_event_loop_session_uses_the_fallback_process_and_exits
) )
server_params = StdioServerParameters(command=sys.executable, args=["-c", server]) server_params = StdioServerParameters(command=sys.executable, args=["-c", server])
# One interpreter cold start on a loaded runner; healthy runs take ~0.3s. # Allow one cold interpreter start on loaded CI.
with anyio.fail_after(10.0): with anyio.fail_after(10.0):
async with stdio_client(server_params): async with stdio_client(server_params):
stream = await accept_alive(sock) stream = await accept_alive(sock)
stack.push_async_callback(stream.aclose) stack.push_async_callback(stream.aclose)
# The engagement proof, asserted while the session is live.
assert isinstance(spawned_processes[0], FallbackProcess) assert isinstance(spawned_processes[0], FallbackProcess)
# The server exited on stdin closure: socket closed, exit code 0, and the
# escalation never fired.
await assert_stream_closed(stream) await assert_stream_closed(stream)
assert spawned_processes[0].returncode == 0 assert spawned_processes[0].returncode == 0
assert terminate_calls == [] assert terminate_calls == []
async def test_a_native_server_emitting_crlf_line_endings_round_trips_messages() -> None: # pragma: no cover async def test_a_native_server_emitting_crlf_line_endings_round_trips_messages() -> None: # pragma: no cover
"""The client round-trips messages from a text-mode Windows server that frames its output with \\r\\n. """The client accepts CRLF-framed messages from a native Windows server.
`TextIOWrapper`'s `newline=None` translates "\\n" to `os.linesep`, so such a Text-mode Windows servers write `\\r\\n`, while the client splits on `\\n`.
server emits \\r\\n; the client still parses each line because the reader
splits on "\\n" only and the JSON parser tolerates the trailing "\\r" as
whitespace. The SDK's own server writes through such a wrapper, so this
tolerance is load-bearing for Windows interop.
tests/issues/test_552_windows_hang.py exercises the same wire form implicitly
through `initialize()`; this test is the explicit owner of the framing claim.
""" """
# Read one request, answer it via print() (which emits \r\n on Windows), then # `print()` emits CRLF on Windows.
# exit when stdin closes. json.loads/dumps keep the script free of SDK imports.
server = ( server = (
"import json, sys\n" "import json, sys\n"
"line = sys.stdin.readline()\n" "line = sys.stdin.readline()\n"
@ -231,13 +163,11 @@ async def test_a_native_server_emitting_crlf_line_endings_round_trips_messages()
ping = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping") ping = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
# One interpreter cold start on a loaded runner; healthy runs take ~0.3s. # Allow one cold interpreter start on loaded CI.
with anyio.fail_after(10.0): with anyio.fail_after(10.0):
async with stdio_client(server_params) as (read_stream, write_stream): async with stdio_client(server_params) as (read_stream, write_stream):
await write_stream.send(SessionMessage(ping)) await write_stream.send(SessionMessage(ping))
received = await read_stream.receive() received = await read_stream.receive()
# A reader that choked on the trailing \r would deliver a ValueError
# here instead of a parsed message.
assert isinstance(received, SessionMessage) assert isinstance(received, SessionMessage)
assert received.message == JSONRPCResponse(jsonrpc="2.0", id=1, result={}) assert received.message == JSONRPCResponse(jsonrpc="2.0", id=1, result={})
@ -262,7 +192,6 @@ async def test_a_tool_spawned_python_child_with_default_stdin_completes_promptly
@mcp.tool() @mcp.tool()
def run_child_bare() -> str: def run_child_bare() -> str:
# Even without redirection the console subsystem hands the child the standard handles.
proc = subprocess.run([sys.executable, "-c", "pass"], timeout=20) proc = subprocess.run([sys.executable, "-c", "pass"], timeout=20)
return str(proc.returncode) return str(proc.returncode)
@ -271,7 +200,7 @@ async def test_a_tool_spawned_python_child_with_default_stdin_completes_promptly
) )
transport = stdio_client(StdioServerParameters(command=sys.executable, args=["-c", server])) transport = stdio_client(StdioServerParameters(command=sys.executable, args=["-c", server]))
# A regression hangs forever, so the bound only has to beat "never". # Guard against the original deadlock.
with anyio.fail_after(40.0): with anyio.fail_after(40.0):
async with Client(transport) as client: async with Client(transport) as client:
result = await client.call_tool("run_child") result = await client.call_tool("run_child")