wehub-resource-sync
120 lines
3.6 KiB
Python
120 lines
3.6 KiB
Python
"""Unit tests for ``run_with_group_timeout``.
|
|
|
|
Exercises the bug the helper exists to fix: a parent that exits
|
|
after spawning a long-running grandchild which holds stdout open.
|
|
Stock ``subprocess.run`` would block until the grandchild dies on
|
|
its own; the helper must fail at ~timeout instead.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import textwrap
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from tests.e2e._run_with_group_timeout import run_with_group_timeout
|
|
|
|
|
|
def test_happy_path_returns_completed_process() -> None:
|
|
"""Child completes within timeout → CompletedProcess returned."""
|
|
result = run_with_group_timeout(
|
|
[sys.executable, "-c", "print('ok')"],
|
|
timeout=10,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0
|
|
assert result.stdout.strip() == "ok"
|
|
|
|
|
|
def test_timeout_kills_grandchild_holding_pipe(tmp_path) -> None:
|
|
"""Helper fails at ~timeout, doesn't wait for the grandchild's 30s sleep."""
|
|
script = tmp_path / "spawn_orphan.py"
|
|
script.write_text(
|
|
textwrap.dedent(
|
|
"""
|
|
import subprocess, sys
|
|
# Grandchild inherits stdout via default fork semantics.
|
|
subprocess.Popen(['sleep', '30'])
|
|
print('parent exiting', flush=True)
|
|
sys.exit(0)
|
|
"""
|
|
)
|
|
)
|
|
|
|
start = time.monotonic()
|
|
with pytest.raises(subprocess.TimeoutExpired):
|
|
run_with_group_timeout(
|
|
[sys.executable, str(script)],
|
|
timeout=2,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
elapsed = time.monotonic() - start
|
|
|
|
# 2s timeout + ≤10s drain = 12s upper-bound on a working helper.
|
|
# 8s leaves plenty of margin and still proves we didn't block 30s.
|
|
assert elapsed < 8.0, f"helper took {elapsed:.1f}s; expected <8s. Group kill probably failed."
|
|
|
|
|
|
def test_timeout_killed_group_is_actually_dead(tmp_path) -> None:
|
|
"""Grandchild PID is unreachable within 2s of the helper raising."""
|
|
pid_file = tmp_path / "grandchild.pid"
|
|
script = tmp_path / "spawn_orphan.py"
|
|
script.write_text(
|
|
textwrap.dedent(
|
|
f"""
|
|
import subprocess, sys
|
|
proc = subprocess.Popen(['sleep', '30'])
|
|
open({str(pid_file)!r}, 'w').write(str(proc.pid))
|
|
sys.exit(0)
|
|
"""
|
|
)
|
|
)
|
|
|
|
with pytest.raises(subprocess.TimeoutExpired):
|
|
run_with_group_timeout(
|
|
[sys.executable, str(script)],
|
|
timeout=2,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
assert pid_file.exists(), "parent didn't reach the pid_file write"
|
|
gc_pid = int(pid_file.read_text())
|
|
|
|
# Poll briefly: kernel may take a moment after killpg to reap.
|
|
deadline = time.monotonic() + 2.0
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
os.kill(gc_pid, 0)
|
|
except ProcessLookupError:
|
|
break
|
|
time.sleep(0.05)
|
|
else:
|
|
pytest.fail(f"grandchild PID {gc_pid} still alive 2s after killpg")
|
|
|
|
|
|
def test_capture_output_is_honored() -> None:
|
|
result = run_with_group_timeout(
|
|
[sys.executable, "-c", "import sys; print('out'); print('err', file=sys.stderr)"],
|
|
timeout=10,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert "out" in result.stdout
|
|
assert "err" in result.stderr
|
|
|
|
|
|
def test_start_new_session_override_rejected() -> None:
|
|
with pytest.raises(ValueError, match="start_new_session must be True"):
|
|
run_with_group_timeout(
|
|
[sys.executable, "-c", "pass"],
|
|
timeout=1,
|
|
start_new_session=False,
|
|
)
|