#!/usr/bin/env python3 """Bounded stdlib PTY proof for live boot presentation preferences.""" import errno import fcntl import json import os import pathlib import re import selectors import struct import subprocess import sys import termios import time DEADLINE_SECONDS = 240.0 CAPTURE_LIMIT = 4 * 1024 * 1024 ANSI = re.compile(rb"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07]*(?:\x07|\x1b\\)|[@-_])") PLAY_CUE = b"Esc quit" def stripped(data): return ANSI.sub(b"", data).replace(b"\r", b"") def spawn(binary, data_dir, content, extra): master, slave = os.openpty() fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 100, 0, 0)) child = subprocess.Popen( [binary, "--data-dir", str(data_dir), "--content-root", str(content)] + list(extra), stdin=slave, stdout=slave, stderr=slave, close_fds=True, start_new_session=True, ) os.close(slave) os.set_blocking(master, False) return child, master def terminate(child): if child.poll() is None: child.kill() try: child.wait(timeout=5) except subprocess.TimeoutExpired: child.kill() child.wait(timeout=5) def drive(binary, data_dir, content, extra, cue, on_cue, keys, expected_exit=0): child, master = spawn(binary, data_dir, content, extra) selector = selectors.DefaultSelector() selector.register(master, selectors.EVENT_READ) registered = True captured = bytearray() acted = False deadline = time.monotonic() + DEADLINE_SECONDS try: while child.poll() is None or registered: remaining = deadline - time.monotonic() if remaining <= 0: raise RuntimeError("PTY child or output drain exceeded 240 second deadline") if registered: for _, _ in selector.select(timeout=min(0.25, remaining)): try: chunk = os.read(master, 65536) except OSError as error: if error.errno == errno.EIO: chunk = b"" elif error.errno in (errno.EAGAIN, errno.EWOULDBLOCK): continue else: raise if not chunk: selector.unregister(master) registered = False break if len(captured) + len(chunk) > CAPTURE_LIMIT: raise RuntimeError("PTY capture exceeded 4 MiB") captured.extend(chunk) else: try: child.wait(timeout=min(0.25, remaining)) except subprocess.TimeoutExpired: pass if ( child.poll() is None and registered and not acted and cue is not None and stripped(bytes(captured)).find(cue) >= 0 ): on_cue(bytes(captured)) if keys: os.write(master, keys) acted = True status = child.wait(timeout=5) if cue is not None and not acted: raise RuntimeError("child exited before render cue; tail=%r" % stripped(bytes(captured))[-2000:]) if status != expected_exit: raise RuntimeError("child exit %d expected %d; tail=%r" % (status, expected_exit, stripped(bytes(captured))[-2000:])) return bytes(captured), status finally: selector.close() os.close(master) terminate(child) def only_run_id(data_dir): values = sorted(path.name for path in (data_dir / "runs").iterdir() if path.is_dir()) if len(values) != 1: raise RuntimeError("expected exactly one run, got %r" % values) return values[0] def explicit_args(): return ["--ancestry", "1", "--calling", "1", "--gamma", "1.5", "--fov", "60", "--palette", "protan"] def assert_complete_preferences(path): value = json.loads(path.read_text(encoding="utf-8")) expected = {"schema_version": 1, "gamma": 1.5, "fov_deg": 60.0, "palette": "protan"} if value != expected: raise RuntimeError("unexpected preferences: %r" % value) def main(): if len(sys.argv) != 5: raise RuntimeError("usage: presentation_scenario.py BINARY DATA CONTENT MODE") binary = sys.argv[1] data_dir = pathlib.Path(sys.argv[2]) content = pathlib.Path(sys.argv[3]) mode = sys.argv[4] data_dir.mkdir(parents=True, exist_ok=True) path = data_dir / "presentation.json" if mode == "contended": captured, status = drive(binary, data_dir, content, explicit_args(), None, lambda _: None, b"", 6) summary = {"exit": status, "contention": b"lock" in stripped(captured).lower()} elif mode == "publish": before_quit = [] def published(_): assert_complete_preferences(path) before_quit.append(True) drive(binary, data_dir, content, explicit_args(), PLAY_CUE, published, b"\x1b") summary = {"published_before_quit": bool(before_quit)} elif mode == "persist-reload": before_quit = [] def published(_): assert_complete_preferences(path) before_quit.append(True) first, _ = drive(binary, data_dir, content, explicit_args(), PLAY_CUE, published, b"\x1b") run_id = only_run_id(data_dir) run_save_path = data_dir / "runs" / run_id / "run.save" capture_root = data_dir.parent first_save_capture = capture_root / "presentation-first-actual.run.save" second_save_capture = capture_root / "presentation-second-actual.run.save" first_save_capture.write_bytes(run_save_path.read_bytes()) before_bytes = path.read_bytes() before_mtime = path.stat().st_mtime_ns second, _ = drive(binary, data_dir, content, ["--load", run_id], PLAY_CUE, lambda _: None, b"\x1b") second_save_capture.write_bytes(run_save_path.read_bytes()) plain_second = stripped(second) summary = { "published_before_quit": bool(before_quit), "reload_completed": b"HEXWORK session over" in plain_second, "bytes_unchanged": path.read_bytes() == before_bytes, "mtime_unchanged": path.stat().st_mtime_ns == before_mtime, "run_id_unchanged": only_run_id(data_dir) == run_id, "reload_status_has_values": b"source=file gamma=1.5 fov=60 palette=protan commit=unchanged" in plain_second, "first_rendered": PLAY_CUE in stripped(first), } elif mode == "defaults": rendered = [] drive(binary, data_dir, content, ["--ancestry", "1", "--calling", "1"], PLAY_CUE, lambda _: rendered.append(True), b"\x1b") summary = {"rendered": bool(rendered), "presentation_absent": not path.exists()} else: raise RuntimeError("unknown scenario mode: " + mode) sys.stdout.write(json.dumps(summary, separators=(",", ":")) + "\n") if __name__ == "__main__": try: main() except Exception as error: sys.stderr.write(("presentation PTY scenario failed: %s\n" % error)[-8192:]) raise