"""`scripts/booth` — the surface every fleet session actually calls. It had no tests at all, which the 2026-09-22 bug-hunt panel found the hard way: its guard-strength table returned UNVERIFIED for every CLI claim because nothing in the suite executes the script. Two of that round's findings live in here. These run the real script under the real system `python3` with no venv, which also makes them a live check on INV-1 (stdlib-only): a third-party import in `marks.py` fails here the same way it fails on a fleet host. """ import json import os import pathlib import subprocess import pytest SCRIPT = pathlib.Path(__file__).parent.parent / "scripts" / "booth" # Exit codes the verbs promise. 0 is a successful read; a reader that CRASHED # must never be one of the meaningful codes, or a caller cannot tell "no" from # "broken" — which is the whole finding. OK, UNANSWERED, NO_SUCH_PICK, READER_FAILED = 0, 1, 2, 3 def run(data, *args, **kw): env = {**os.environ, "BOOTH_DATA_DIR": str(data), "BOOTH_URL": "http://booth.invalid"} return subprocess.run([str(SCRIPT), *args], capture_output=True, text=True, env=env, timeout=30, **kw) @pytest.fixture def booth(tmp_path): b = tmp_path / "b" b.mkdir() return tmp_path, b def _declare(booth_dir, mark_id="winner"): import sys sys.path.insert(0, str(pathlib.Path(__file__).parent.parent)) from booth.marks import declare_pick declare_pick(booth_dir, mark_id, {"prompt": "Which one?", "options": ["A", "B"]}) def test_marks_prints_one_json_document(booth): """`booth marks ` is a read. Its stdout is parsed by the session that called it, so it has to be ONE document — and exit 0, because the read succeeded. Whether a pick is open is in the payload's `open` list, which is where a caller should read it from.""" data, b = booth _declare(b) r = run(data, "marks", "b") assert r.returncode == OK, r.stderr doc = json.loads(r.stdout) assert doc["open"] == ["winner"] def test_marks_wait_prints_once_not_once_per_poll(booth): """`--wait` polls every 2 s and printed the whole document on every pass, so a capture held several concatenated JSON values and `jq` could not read any of them. The wait is a wait; the print is the result.""" data, b = booth _declare(b) import sys sys.path.insert(0, str(pathlib.Path(__file__).parent.parent)) from booth.marks import answer_pick # Answer it after the first poll so --wait genuinely loops at least once. r = subprocess.Popen([str(SCRIPT), "marks", "b", "--wait", "20"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env={**os.environ, "BOOTH_DATA_DIR": str(data), "BOOTH_URL": "http://booth.invalid"}) import time time.sleep(3) answer_pick(b, "winner", "A") out, err = r.communicate(timeout=30) assert r.returncode == OK, err json.loads(out) # ONE document, or this raises def test_marks_reports_a_reader_failure_instead_of_printing_garbage(booth): """A traceback on stdout with exit 0 is the worst of both: the caller's `jq` sees success and gets nothing. A read that could not happen is its own answer and gets its own code.""" data, b = booth (b / ".marks.json").write_bytes(b"\xff\xfe not utf-8 at all") r = run(data, "marks", "b") assert r.returncode == READER_FAILED, f"rc={r.returncode} out={r.stdout!r}" def test_answer_distinguishes_a_crash_from_an_unanswered_pick(booth): """`answer` funnelled a reader crash and "not yet answered" through the same exit 1, so `--wait` spun for the full hour on a broken file and then blamed the operator for not answering.""" data, b = booth _declare(b) r = run(data, "answer", "b", "winner") assert r.returncode == UNANSWERED (b / ".marks.json").write_bytes(b"\xff\xfe not utf-8 at all") r = run(data, "answer", "b", "winner", "--wait", "6") assert r.returncode == READER_FAILED, ( "a crash was read as 'unanswered' and waited out the timeout" ) def test_answer_on_a_note_id_says_no_such_pick(booth): """`answer` matched on id alone while the web route filters on shape, so a note id was reported 'unanswered' and polled forever — a question that could never be answered because it was never a question.""" data, b = booth import sys sys.path.insert(0, str(pathlib.Path(__file__).parent.parent)) from booth.marks import write_note write_note(b, "a.png", "just a note") r = run(data, "answer", "b", "note-1") assert r.returncode == NO_SUCH_PICK assert "no such pick" in r.stderr # ---- U5: self-announcing booths --------------------------------------------- def _manifest(booth_dir): import sys sys.path.insert(0, str(pathlib.Path(__file__).parent.parent)) from booth.manifest import read_manifest return read_manifest(booth_dir) def test_new_announces_the_booth(tmp_path): """`$ALTHING_HANDLE` is the whole provenance story: the session already has it, so the booth can say who made it without anybody typing a name.""" env = {**os.environ, "ALTHING_HANDLE": "shutter-dev"} r = subprocess.run([str(SCRIPT), "new", "r18-ab", "--why", "pick the winner"], capture_output=True, text=True, timeout=30, env={**env, "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"}) assert r.returncode == 0, r.stderr m = _manifest(tmp_path / "r18-ab") assert m.handle == "shutter-dev" assert m.why == "pick the winner" def test_new_without_a_why_is_still_legal(tmp_path): """The flags are optional and existing call sites keep working. A booth that says only who made it is still a booth that said something.""" r = subprocess.run([str(SCRIPT), "new", "scratch"], capture_output=True, text=True, timeout=30, env={**os.environ, "ALTHING_HANDLE": "booth-dev", "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"}) assert r.returncode == 0, r.stderr m = _manifest(tmp_path / "scratch") assert m.handle == "booth-dev" and m.why == "" def test_add_announces_and_still_copies_the_files(tmp_path): """`add` is the verb most sessions actually use — it creates the booth AND fills it — so the why has to ride on it or it rides nowhere.""" src = tmp_path / "src" src.mkdir() (src / "a.txt").write_text("content") r = subprocess.run([str(SCRIPT), "add", "r18-ab", str(src / "a.txt"), "--why", "second pass", "--title", "R18 A/B"], capture_output=True, text=True, timeout=30, env={**os.environ, "ALTHING_HANDLE": "booth-dev", "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"}) assert r.returncode == 0, r.stderr assert (tmp_path / "r18-ab" / "a.txt").read_text() == "content" m = _manifest(tmp_path / "r18-ab") assert m.why == "second pass" and m.title == "R18 A/B" def test_add_re_announcing_keeps_the_original_created(tmp_path): """The common shape: `new` opens the booth, `add` drops the second batch and sharpens the why. The booth appeared once.""" env = {**os.environ, "ALTHING_HANDLE": "booth-dev", "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"} src = tmp_path / "a.txt" src.write_text("x") subprocess.run([str(SCRIPT), "new", "b", "--why", "first"], check=True, capture_output=True, timeout=30, env=env) first = _manifest(tmp_path / "b").created subprocess.run([str(SCRIPT), "add", "b", str(src), "--why", "sharper"], check=True, capture_output=True, timeout=30, env=env) after = _manifest(tmp_path / "b") assert after.created == first assert after.why == "sharper" def test_the_link_board_announces_itself_as_the_booths_own(tmp_path): """No exemption list. The standing board is made by the service and posted to by seventeen handles, so no single agent owns it — `booth` is the truthful answer, and it keeps the rule to one line.""" r = subprocess.run([str(SCRIPT), "link", "http://example.invalid", "a thing"], capture_output=True, text=True, timeout=30, env={**os.environ, "ALTHING_HANDLE": "booth-dev", "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"}) assert r.returncode == 0, r.stderr m = _manifest(tmp_path / "links") assert m is not None and m.handle == "booth" assert m.why