wehub-resource-sync
49b9bb6724
Deploy Docs / deploy-docs (push) Failing after 1s
Conformance Tests / client-conformance (push) Failing after 3s
Conformance Tests / server-conformance (push) Failing after 1s
GitHub Actions Security Analysis / zizmor (push) Failing after 1s
CI / checks (push) Failing after 59m20s
CI / all-green (push) Waiting to run
78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
"""Fixtures for the stdio lifecycle suite.
|
|
|
|
Provides recording seams around `stdio_client`'s spawn and tree-termination
|
|
internals (the real implementations still run), plus a teardown that keeps a
|
|
crashed test from orphaning its sleep-forever subprocesses.
|
|
"""
|
|
|
|
import os
|
|
import signal
|
|
import sys
|
|
from collections.abc import Generator
|
|
from contextlib import suppress
|
|
from pathlib import Path
|
|
from typing import TextIO
|
|
|
|
import anyio.abc
|
|
import pytest
|
|
|
|
from mcp.client import stdio
|
|
from mcp.client.stdio import _create_platform_compatible_process, _terminate_process_tree
|
|
from mcp.os.win32.utilities import FallbackProcess
|
|
|
|
|
|
@pytest.fixture
|
|
def spawned_processes(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> Generator[list[anyio.abc.Process | FallbackProcess]]:
|
|
"""Record every process `stdio_client` spawns; the real spawn still runs.
|
|
|
|
Teardown SIGKILLs each spawn-time process group on POSIX: the safety net for a
|
|
test that dies mid-body and the reaper for deliberate survivors. On Windows
|
|
there is no group to signal (the Job Object covers strays).
|
|
"""
|
|
spawned: list[anyio.abc.Process | FallbackProcess] = []
|
|
|
|
async def recording_spawn(
|
|
command: str,
|
|
args: list[str],
|
|
env: dict[str, str] | None = None,
|
|
errlog: TextIO = sys.stderr,
|
|
cwd: Path | str | None = None,
|
|
) -> anyio.abc.Process | FallbackProcess:
|
|
process = await _create_platform_compatible_process(command, args, env, errlog, cwd)
|
|
spawned.append(process)
|
|
return process
|
|
|
|
monkeypatch.setattr(stdio, "_create_platform_compatible_process", recording_spawn)
|
|
yield spawned
|
|
_kill_spawn_groups(spawned)
|
|
|
|
|
|
@pytest.fixture
|
|
def terminate_calls(monkeypatch: pytest.MonkeyPatch) -> list[anyio.abc.Process | FallbackProcess]:
|
|
"""Record every invocation of `stdio_client`'s tree-termination seam; the real termination still runs.
|
|
|
|
An empty list after the context exits proves the graceful path: a FIN looks the
|
|
same whether the peer exited on stdin closure or was killed.
|
|
"""
|
|
terminated: list[anyio.abc.Process | FallbackProcess] = []
|
|
|
|
async def recording_terminate(process: anyio.abc.Process | FallbackProcess) -> None:
|
|
terminated.append(process)
|
|
await _terminate_process_tree(process)
|
|
|
|
monkeypatch.setattr(stdio, "_terminate_process_tree", recording_terminate)
|
|
return terminated
|
|
|
|
|
|
# lax no cover: registered on every platform but a no-op on Windows, whose runners enforce 100% per job.
|
|
def _kill_spawn_groups(spawned: list[anyio.abc.Process | FallbackProcess]) -> None: # pragma: lax no cover
|
|
"""SIGKILL each spawn-time process group; see `spawned_processes`."""
|
|
if sys.platform == "win32":
|
|
return
|
|
for process in spawned:
|
|
# macOS killpg raises EPERM for a group holding only unreaped zombies.
|
|
with suppress(ProcessLookupError, PermissionError):
|
|
os.killpg(process.pid, signal.SIGKILL)
|