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
81 lines
3.5 KiB
Python
81 lines
3.5 KiB
Python
"""Kernel-synchronized liveness probes for the real-subprocess stdio lifecycle suite.
|
|
|
|
A spawned (grand)child connects back to a test-owned TCP listener and sends
|
|
`b'alive'`; the kernel then provides every signal a test needs, with no sleeps or
|
|
polling. The kernel closes all of a process's file descriptors on exit, so EOF
|
|
(clean close / FIN) or `BrokenResourceError` (abrupt close / RST, typical of
|
|
SIGKILL and Windows job termination) proves death; only a running process can
|
|
answer an echo, so a reply proves liveness without racing a kill.
|
|
|
|
Extracted from the real-process section of tests/client/test_stdio.py; the two
|
|
copies on this branch are deliberate -- consolidating them is follow-up work.
|
|
"""
|
|
|
|
import anyio
|
|
import anyio.abc
|
|
import pytest
|
|
|
|
|
|
def connect_back_script(port: int, *, echo: bool = False) -> str:
|
|
"""Return a `python -c` script body that connects to 127.0.0.1:`port` and sends `b'alive'`.
|
|
|
|
After the banner the script blocks forever -- or, with `echo=True`, echoes every
|
|
received chunk back so `assert_peer_echoes` can prove the process still runs.
|
|
"""
|
|
# lax no cover: echo mode is used only by POSIX-gated tests; Windows runners enforce 100% per job.
|
|
if echo: # pragma: lax no cover
|
|
tail = "while True:\n data = s.recv(65536)\n if not data:\n break\n s.sendall(data)\n"
|
|
else:
|
|
tail = "time.sleep(3600)\n"
|
|
return f"import socket, time\ns = socket.create_connection(('127.0.0.1', {port}))\ns.sendall(b'alive')\n" + tail
|
|
|
|
|
|
async def open_liveness_listener() -> tuple[anyio.abc.SocketListener, int]:
|
|
"""Open a TCP listener on localhost and return it along with its port."""
|
|
multi = await anyio.create_tcp_listener(local_host="127.0.0.1")
|
|
sock = multi.listeners[0]
|
|
assert isinstance(sock, anyio.abc.SocketListener)
|
|
addr = sock.extra(anyio.abc.SocketAttribute.local_address)
|
|
# IPv4 local_address is (host: str, port: int)
|
|
assert isinstance(addr, tuple) and len(addr) >= 2 and isinstance(addr[1], int)
|
|
return sock, addr[1]
|
|
|
|
|
|
async def accept_alive(sock: anyio.abc.SocketListener) -> anyio.abc.SocketStream:
|
|
"""Accept one connection and assert the peer sent `b'alive'`.
|
|
|
|
Reads until the full 5-byte banner arrives (TCP may legally split even a tiny
|
|
send). Callers bound this with `anyio.fail_after` to catch a subprocess that
|
|
never started.
|
|
"""
|
|
stream = await sock.accept()
|
|
msg = b""
|
|
while len(msg) < 5:
|
|
msg += await stream.receive(5 - len(msg))
|
|
assert msg == b"alive", f"expected b'alive', got {msg!r}"
|
|
return stream
|
|
|
|
|
|
async def assert_stream_closed(stream: anyio.abc.SocketStream) -> None:
|
|
"""Assert the peer holding the other end of `stream` has terminated."""
|
|
with anyio.fail_after(5.0), pytest.raises((anyio.EndOfStream, anyio.BrokenResourceError)):
|
|
await stream.receive(1)
|
|
|
|
|
|
async def assert_peer_echoes(stream: anyio.abc.SocketStream) -> None: # pragma: lax no cover
|
|
"""Assert the peer holding the other end of `stream` is still running.
|
|
|
|
Round-trips one echo through the stream (the peer must use `echo=True`); a dead
|
|
process can never answer, so this cannot pass spuriously.
|
|
|
|
lax no cover: only POSIX-gated survival tests call this; Windows runners
|
|
enforce 100% coverage per job.
|
|
"""
|
|
with anyio.fail_after(5.0):
|
|
await stream.send(b"ping")
|
|
# Read until the full echo has arrived: TCP may legally split even a tiny send.
|
|
echoed = b""
|
|
while len(echoed) < 4:
|
|
echoed += await stream.receive(4 - len(echoed))
|
|
assert echoed == b"ping", f"expected b'ping', got {echoed!r}"
|