zemion
2ffdb23f69
Verified with the coordinated workspace changes by devkit full run 2026-09-08T225814-186389-0000-3e3ed7cd (all seven phases passed). This shared UI pass does not mark the individual module reviews complete.
250 lines
8.5 KiB
Python
Executable File
250 lines
8.5 KiB
Python
Executable File
"""Bounded fixture processes only; no project servers or external transports."""
|
|
|
|
from pathlib import Path
|
|
from dataclasses import FrozenInstanceError
|
|
import sys
|
|
import threading
|
|
import time
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools/devkit"))
|
|
from govoplan_devkit.process import _OutputBuffer, require_capture, run_captured
|
|
|
|
|
|
def python(code, **kwargs):
|
|
return run_captured([sys.executable, "-c", code], **kwargs)
|
|
|
|
|
|
def test_large_stdin_and_separate_outputs_are_multiplexed():
|
|
result = python(
|
|
"import sys; sys.stderr.write('diagnostic'); print(len(sys.stdin.buffer.read()))",
|
|
input_bytes=b"x" * 400000,
|
|
)
|
|
assert result.status == "passed"
|
|
assert result.stdout == b"400000\n"
|
|
assert result.stderr == b"diagnostic"
|
|
|
|
|
|
def test_live_output_limit_terminates_unbounded_producer():
|
|
result = python(
|
|
"import os;\nwhile True: os.write(1, b'x'*65536)", max_stdout=2048, timeout=3
|
|
)
|
|
assert result.status == "output_limit"
|
|
assert result.truncated
|
|
assert len(result.stdout) == 2048
|
|
|
|
|
|
def test_drain_mode_caps_memory_without_turning_success_into_failure():
|
|
result = python("print('x'*100000)", max_stdout=1024, terminate_on_limit=False)
|
|
assert result.status == "passed"
|
|
assert result.truncated and len(result.stdout) == 1024
|
|
|
|
|
|
def test_deadline_applies_after_command_closes_both_output_streams():
|
|
started = time.monotonic()
|
|
result = python(
|
|
"import os,time; os.close(1); os.close(2); time.sleep(30)", timeout=0.15
|
|
)
|
|
assert result.status == "timed_out"
|
|
assert time.monotonic() - started < 5
|
|
|
|
|
|
def test_cancellation_stops_owned_process():
|
|
cancelled = threading.Event()
|
|
timer = threading.Timer(0.1, cancelled.set)
|
|
timer.start()
|
|
try:
|
|
result = python("import time; time.sleep(30)", cancelled=cancelled)
|
|
finally:
|
|
timer.cancel()
|
|
timer.join()
|
|
assert result.status == "interrupted"
|
|
|
|
|
|
@pytest.mark.parametrize("redirected", [False, True])
|
|
def test_outliving_child_is_not_success_and_cannot_keep_running(redirected):
|
|
code = (
|
|
"import subprocess,sys; child=subprocess.Popen([sys.executable,'-c','import time; time.sleep(30)']"
|
|
+ (",stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL" if redirected else "")
|
|
+ "); print(child.pid,flush=True)"
|
|
)
|
|
result = python(code, timeout=4)
|
|
assert result.status == "leaked_process"
|
|
child_pid = int(result.stdout.strip())
|
|
# A terminated adopted child may remain a zombie until the host's init reaps it.
|
|
proc = Path(f"/proc/{child_pid}/stat")
|
|
if proc.exists():
|
|
assert proc.read_text().split(")", 1)[1].strip().split()[0] == "Z"
|
|
|
|
|
|
def test_required_capture_rejects_nonordinary_completion():
|
|
with pytest.raises(ValueError, match="output_limit"):
|
|
require_capture([sys.executable, "-c", "print('x'*10000)"], max_stdout=32)
|
|
|
|
|
|
def test_ordinary_nonzero_exit_retains_diagnostics():
|
|
result = python("import sys; print('problem',file=sys.stderr); sys.exit(7)")
|
|
assert (result.status, result.returncode, result.stderr) == (
|
|
"failed",
|
|
7,
|
|
b"problem\n",
|
|
)
|
|
|
|
|
|
def test_selector_setup_failure_still_terminates_spawned_process():
|
|
from govoplan_devkit import process
|
|
|
|
original = process.subprocess.Popen
|
|
spawned = []
|
|
|
|
def capture(*args, **kwargs):
|
|
child = original(*args, **kwargs)
|
|
spawned.append(child)
|
|
return child
|
|
|
|
with (
|
|
patch.object(process.subprocess, "Popen", side_effect=capture),
|
|
patch.object(
|
|
process.selectors,
|
|
"DefaultSelector",
|
|
side_effect=OSError("fixture selector unavailable"),
|
|
),
|
|
):
|
|
with pytest.raises(OSError, match="selector unavailable"):
|
|
python("import time; time.sleep(30)")
|
|
assert len(spawned) == 1 and spawned[0].poll() is not None
|
|
|
|
|
|
def test_head_tail_keeps_the_actual_failure_tail_within_original_bound():
|
|
output = b"BEGIN" + b"middle" * 100 + b"FINAL ERROR"
|
|
result = python(
|
|
f"import os; os.write(1, {output!r}); raise SystemExit(7)",
|
|
max_stdout=32,
|
|
capture_mode="head_tail",
|
|
terminate_on_limit=False,
|
|
)
|
|
assert (result.status, result.returncode) == ("failed", 7)
|
|
assert result.stdout == output[:16] + output[-16:]
|
|
assert result.stdout_head_bytes == 16
|
|
assert result.omitted_stdout_bytes == len(output) - 32
|
|
assert result.snapshot().final
|
|
assert "FINAL ERROR" in result.snapshot().text()
|
|
assert str(len(output) - 32) + " output bytes omitted" in result.snapshot().text()
|
|
|
|
|
|
@pytest.mark.parametrize("limit", [0, 1, 2, 3, 31, 1024])
|
|
def test_rolling_buffers_remain_bounded_for_every_chunk(limit):
|
|
buffer = _OutputBuffer(limit, "head_tail")
|
|
original = b""
|
|
for data in (b"a", b"bcdef", b"x" * 65536, b"last error"):
|
|
original += data
|
|
buffer.append(data)
|
|
assert len(buffer.head) + len(buffer.tail) <= limit
|
|
assert buffer.omitted == max(0, len(original) - limit)
|
|
if len(original) > limit:
|
|
head = limit // 2
|
|
tail = limit - head
|
|
expected = original[:head] + (original[-tail:] if tail else b"")
|
|
assert buffer.value() == expected
|
|
|
|
|
|
def test_prefix_capture_remains_exact_and_counts_omitted_bytes():
|
|
result = python(
|
|
"import os; os.write(1,b'0123456789')", max_stdout=4, terminate_on_limit=False
|
|
)
|
|
assert result.stdout == b"0123"
|
|
assert result.stdout_head_bytes == 4
|
|
assert result.omitted_stdout_bytes == 6
|
|
|
|
|
|
def test_callback_exposes_bounded_immutable_initial_and_final_snapshots():
|
|
snapshots = []
|
|
result = python(
|
|
"import os,time; os.write(1,b'first\\n'); time.sleep(.15); "
|
|
"os.write(1,b'x'*10000+b'FINAL\\n'); time.sleep(.1)",
|
|
max_stdout=32,
|
|
capture_mode="head_tail",
|
|
terminate_on_limit=False,
|
|
on_output=snapshots.append,
|
|
)
|
|
assert not snapshots[0].final
|
|
assert snapshots[-1] == result.snapshot()
|
|
assert snapshots[0].stdout == b"first\n"
|
|
assert snapshots[-1].stdout.endswith(b"FINAL\n")
|
|
assert all(len(item.stdout) <= 32 for item in snapshots)
|
|
with pytest.raises(FrozenInstanceError):
|
|
snapshots[0].final = True
|
|
|
|
|
|
def test_callback_updates_dirty_output_while_child_becomes_quiet():
|
|
snapshots = []
|
|
python(
|
|
"import os,time; os.write(1,b'first\\n'); time.sleep(.15); "
|
|
"os.write(1,b'second\\n'); time.sleep(1.15)",
|
|
on_output=lambda item: snapshots.append((time.monotonic(), item)),
|
|
)
|
|
assert len(snapshots) >= 3
|
|
assert snapshots[0][1].stdout == b"first\n"
|
|
assert not snapshots[1][1].final
|
|
assert snapshots[1][1].stdout.endswith(b"second\n")
|
|
assert snapshots[1][0] - snapshots[0][0] >= 0.95
|
|
assert snapshots[-1][1].final
|
|
|
|
|
|
def test_quiet_process_has_one_final_empty_snapshot():
|
|
snapshots = []
|
|
python("pass", on_output=snapshots.append)
|
|
assert len(snapshots) == 1
|
|
assert snapshots[0].final and snapshots[0].stdout == b""
|
|
|
|
|
|
def test_snapshot_text_handles_cut_and_invalid_utf8_without_malformed_strings():
|
|
result = python(
|
|
"import os; os.write(1, 'Ä😊Z'.encode()*20+b'\\xff\\xfe')",
|
|
max_stdout=9,
|
|
capture_mode="head_tail",
|
|
terminate_on_limit=False,
|
|
)
|
|
text = result.snapshot().text()
|
|
text.encode("utf-8", errors="strict")
|
|
assert "output bytes omitted" in text
|
|
assert len(result.stdout) == 9
|
|
with pytest.raises(ValueError, match="stream"):
|
|
result.snapshot().text("other")
|
|
|
|
|
|
def test_callback_failure_terminates_owned_process_and_propagates():
|
|
from govoplan_devkit import process
|
|
|
|
original = process.subprocess.Popen
|
|
spawned = []
|
|
|
|
def capture(*args, **kwargs):
|
|
child = original(*args, **kwargs)
|
|
spawned.append(child)
|
|
return child
|
|
|
|
def fail(_snapshot):
|
|
raise ValueError("fixture callback failed")
|
|
|
|
with patch.object(process.subprocess, "Popen", side_effect=capture):
|
|
with pytest.raises(ValueError, match="fixture callback failed"):
|
|
python(
|
|
"import time; print('ready',flush=True); time.sleep(30)", on_output=fail
|
|
)
|
|
assert len(spawned) == 1 and spawned[0].poll() is not None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"kwargs", [{"capture_mode": "other"}, {"max_stdout": -1}, {"max_stderr": True}]
|
|
)
|
|
def test_invalid_capture_configuration_fails_before_starting_a_process(kwargs):
|
|
from govoplan_devkit import process
|
|
|
|
with patch.object(process.subprocess, "Popen") as popen:
|
|
with pytest.raises(ValueError):
|
|
python("pass", **kwargs)
|
|
popen.assert_not_called()
|