"""`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 def test_the_flags_can_sit_on_either_side_of_the_files(tmp_path): """`booth add b *.png --why "..."` and `booth add b --why "..." *.png` both work. A glob is usually last and a flag usually after it, but nothing enforces that and a session should not have to remember which.""" src = tmp_path / "a.png" src.write_bytes(b"x") env = {**os.environ, "ALTHING_HANDLE": "booth-dev", "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"} for name, args in (("after", ["add", "after", str(src), "--why", "w"]), ("before", ["add", "before", "--why", "w", str(src)])): r = subprocess.run([str(SCRIPT), *args], capture_output=True, text=True, timeout=30, env=env) assert r.returncode == 0, r.stderr assert _manifest(tmp_path / name).why == "w" assert (tmp_path / name / "a.png").exists(), "the files stopped being copied" def test_a_why_survives_quotes_and_non_ascii_and_is_flattened(tmp_path): """The reason this goes through manifest.py instead of printf-ing JSON from the shell: a why containing a quote, a backslash or a newline is not an edge case, it is a sentence somebody wrote. Newlines flatten because the field renders inside a card's sub-line.""" r = subprocess.run( [str(SCRIPT), "new", "b", "--why", 'he said "pick v3" — line1\nline2 · ünï'], 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 why = _manifest(tmp_path / "b").why assert why == 'he said "pick v3" — line1 line2 · ünï' def test_a_flag_with_no_value_does_not_eat_the_booth_name(tmp_path): """`booth new b --why` with nothing after it must not consume `b` as the value and then create a booth called nothing. Usage, and no directory.""" r = subprocess.run([str(SCRIPT), "new", "b", "--why"], capture_output=True, text=True, timeout=30, env={**os.environ, "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"}) assert r.returncode == 2 assert "usage:" in r.stderr assert not (tmp_path / "b").exists() def test_a_bare_add_does_not_wipe_the_why_the_new_set(tmp_path): """`booth new x --why "..."` then `booth add x out/*.png` is THE sequence, and the second call must not erase the first one's sentence. The module distinguishes omitted from empty; the shell has to carry that distinction across, which means an UNSET variable, not an empty one.""" env = {**os.environ, "ALTHING_HANDLE": "booth-dev", "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"} src = tmp_path / "a.png" src.write_bytes(b"x") subprocess.run([str(SCRIPT), "new", "b", "--why", "pick the denoiser", "--title", "R18 A/B"], check=True, capture_output=True, timeout=30, env=env) subprocess.run([str(SCRIPT), "add", "b", str(src)], check=True, capture_output=True, timeout=30, env=env) m = _manifest(tmp_path / "b") assert m.why == "pick the denoiser", "a bare `booth add` wiped the why" assert m.title == "R18 A/B" def test_an_explicitly_empty_why_still_clears_it(tmp_path): """Omitted means unchanged; supplied-and-empty means the poster meant to take it back. Both have to be reachable from the shell.""" env = {**os.environ, "ALTHING_HANDLE": "booth-dev", "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"} subprocess.run([str(SCRIPT), "new", "b", "--why", "wrong"], check=True, capture_output=True, timeout=30, env=env) subprocess.run([str(SCRIPT), "new", "b", "--why", ""], check=True, capture_output=True, timeout=30, env=env) assert _manifest(tmp_path / "b").why == "" def test_answer_and_marks_agree_about_what_open_means(tmp_path): """U2 made `_is_open` THE openness predicate — "nothing else may spell this out" — and `booth answer`'s reader spelled it out anyway, as `if m.answer is None`. So a PARTIALLY answered pick read as done to `answer` and still-open to `marks --wait`: one verb returns the half-filled form and the other blocks on the same booth at the same instant. Found 2/4. The two verbs are the session's whole view of the loop, and a session that asks both gets two answers. """ import sys sys.path.insert(0, str(pathlib.Path(__file__).parent.parent)) from booth.marks import answer_pick, declare_pick b = tmp_path / "b" b.mkdir() declare_pick(b, "batch", { "title": "R18", "questions": [ {"key": "q1", "prompt": "One?", "options": ["keep", "drop"]}, {"key": "q2", "prompt": "Two?", "options": ["keep", "drop"]}, ], }) answer_pick(b, "batch", {"q1": "keep", "q2": None}) # partial env = {**os.environ, "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"} marks = subprocess.run([str(SCRIPT), "marks", "b"], capture_output=True, text=True, timeout=30, env=env) answer = subprocess.run([str(SCRIPT), "answer", "b", "batch"], capture_output=True, text=True, timeout=30, env=env) still_open = "batch" in json.loads(marks.stdout)["open"] assert still_open, "a partial answer stopped counting as open" assert answer.returncode == UNANSWERED, ( "`answer` called a partially-answered pick done while `marks` called it open" ) def test_answer_does_not_poll_forever_on_a_pick_that_cannot_be_answered(tmp_path): """The mirror failure. A pick whose declaration went bad hydrates with `error` set, which makes it NOT open — so `marks --wait` returns at once while `answer --wait` polled the full hour against a form the web route refuses with a 400. Nothing was ever going to land.""" b = tmp_path / "b" b.mkdir() (b / ".marks.json").write_text(json.dumps({ "version": 1, "marks": [{"id": "broken", "shape": "pick", "declaration": {}, "error": "pick has no declaration", "created": "2026-09-21T00:00:00.000000+00:00"}], })) r = subprocess.run([str(SCRIPT), "answer", "b", "broken", "--wait", "8"], capture_output=True, text=True, timeout=40, env={**os.environ, "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"}) assert r.returncode != 0 assert "broken" in r.stderr.lower() or "cannot" in r.stderr.lower() # ---- U6: benches ------------------------------------------------------------ # # The CLI half of the unit. `docs/contracts/u6_benches.contract.md`. REFUSED = 2 # Shared with tests/test_benches.py::BOOTH_URL_TABLE — INV-2 says ONE predicate # decides what a booth URL is, and these are the rows the CLI must agree on. # A second `/b/` check inlined in the shell for speed goes red HERE. from test_benches import BOOTH_URL_TABLE # noqa: E402 from booth.benches import normalize_bench_url as normalize_bench_url_cli # noqa: E402 @pytest.mark.parametrize("url,is_booth", [(u, e is not None) for u, e in BOOTH_URL_TABLE]) def test_link_refuses_exactly_what_booth_target_matches(booth, url, is_booth): """INV-2. Defeating change: a `case "$url" in *':8090/b/'*)` in the shell, which would classify the host-agnostic and percent-encoded rows differently from the Python predicate the board's dead marker uses.""" data, _ = booth r = run(data, "link", url, "a description") assert (r.returncode == REFUSED) is is_booth, (url, r.returncode, r.stderr) def test_a_refused_link_writes_nothing_at_all(booth): """INV-3. Defeating change: putting the refusal AFTER the `mkdir -p` / announce block, which is where it would naturally land if written without thinking. Asserting only that links.md lacks the row would PASS under that change — so this asserts the board directory does not exist.""" data, _ = booth board = data / "links" assert not board.exists() before = sorted(p.name for p in data.iterdir()) r = run(data, "link", "http://10.100.10.50:8090/b/some-booth/", "nope") assert r.returncode == REFUSED assert not board.exists(), "a refused link created the board directory" # NOTHING AT ALL, not just no board. Asserting only `links/`'s absence let # a refusal that touched `.benches.lock` (or any other sidecar) on its way # out stay green — the cold panel's vacuity pass named exactly that. assert sorted(p.name for p in data.iterdir()) == before, "a refused link wrote something" def test_the_refusal_names_the_alternative(booth): """The teaching moment belongs at the point of use: 17 handles have the muscle memory, and a bare 'refused' sends them to a human.""" data, _ = booth r = run(data, "link", "http://10.100.10.50:8090/b/some-booth/", "nope") out = r.stderr + r.stdout assert "--why" in out and "some-booth" in out def test_a_reference_bookmark_is_still_a_link(booth): """The board keeps its residual job. Measured: ~14 of the 35 distinct non-booth targets are repos, model cards and docs, for which the board is the right and only home. A second refusal would break that.""" data, _ = booth r = run(data, "link", "https://gitea.phasefinal.com/vh/peedlar", "the repo") assert r.returncode == OK, r.stderr assert "the repo" in (data / "links" / "links.md").read_text() def test_bench_add_is_an_upsert(booth): data, _ = booth for i in range(3): r = run(data, "bench", "add", "https://talk.nh3.phasefinal.com:8092/", f"talk v{i}") assert r.returncode == OK, r.stderr r = run(data, "bench", "ls") assert r.returncode == OK, r.stderr assert r.stdout.count("talk v") == 1 and "talk v2" in r.stdout # THE ID, WHOLE AND UNTRUNCATED, because it is the locator `bench state` # and `bench rm` take. An earlier version of this test had a docstring # claiming `bench ls` prints ids and asserted nothing of the kind, while # the code printed a url truncated to 52 columns — a claim standing in for # evidence, which is how the drift would have survived CI. Found by all # four cold arms independently. bid = normalize_bench_url_cli("https://talk.nh3.phasefinal.com:8092/") assert bid in r.stdout, r.stdout # and what ls prints is addressable, end to end line = [l for l in r.stdout.splitlines() if "talk v2" in l][0] printed_id = line.split()[-1] assert run(data, "bench", "state", printed_id, "promoted").returncode == OK def test_bench_verbs_round_trip(booth): data, _ = booth assert run(data, "bench", "add", "http://x.test/", "ex").returncode == OK assert run(data, "bench", "state", "http://x.test/", "promoted").returncode == OK assert "promoted" in run(data, "bench", "ls").stdout assert run(data, "bench", "rm", "http://x.test/").returncode == OK assert "ex" not in run(data, "bench", "ls").stdout def test_bench_state_and_rm_take_an_id_or_a_url(booth): """`bench ls` prints ids; the operator has the URL. BOTH must address. This used to invoke both verbs with the URL only, twice, while its docstring claimed it covered the id — the same claim-not-evidence shape as the `ls` docstring. A raw URL whose normalization DIFFERS from it is used, so the two columns are genuinely distinct inputs. Cold panel, regin F8. """ data, _ = booth raw = "HTTP://X.Test:80/p/?b=2&a=1#frag" bid = normalize_bench_url_cli(raw) assert bid != raw, "pick a URL whose normalization actually differs" run(data, "bench", "add", raw, "ex") # by the ID the registry stores assert run(data, "bench", "state", bid, "retired").returncode == OK assert "retired" in run(data, "bench", "ls").stdout # and by the RAW URL the operator has in their scrollback assert run(data, "bench", "state", raw, "live").returncode == OK assert "live" in run(data, "bench", "ls").stdout assert run(data, "bench", "rm", raw).returncode == OK run(data, "bench", "add", raw, "ex again") assert run(data, "bench", "rm", bid).returncode == OK assert "ex" not in run(data, "bench", "ls").stdout def test_bench_add_refuses_a_bad_url_with_the_reason(booth): data, _ = booth r = run(data, "bench", "add", "ftp://x.test/f", "ex") assert r.returncode != OK assert "http" in (r.stderr + r.stdout).lower() def test_bare_bench_names_the_bench_verbs(booth): """Seam review SR-6: `bench` is the first two-word verb in this script, and falling through to the generic usage hides which word was wrong.""" data, _ = booth r = run(data, "bench") assert r.returncode != OK assert "add" in r.stderr and "import" in r.stderr def _seed_board(data): board = data / "links" board.mkdir(parents=True, exist_ok=True) (board / "links.md").write_text( "- [a booth](http://10.100.10.50:8090/b/gone/) · x · 2026-09-01 00:00\n" "- [talk](https://talk.nh3.phasefinal.com:8092/) · x · 2026-09-01 00:00\n" "- [talk again](https://talk.nh3.phasefinal.com:8092/) · x · 2026-09-02 00:00\n" "- [a repo](https://gitea.phasefinal.com/vh/peedlar) · x · 2026-09-03 00:00\n" "- [bad](ftp://x.test/f) · x · 2026-09-04 00:00\n" ) return board def test_import_writes_nothing_without_apply(booth): """INV-8. A proposal that writes is not a proposal.""" data, _ = booth board = _seed_board(data) before = (board / "links.md").read_text() r = run(data, "bench", "import") assert r.returncode == OK, r.stderr assert not (data / ".benches.json").exists() assert (board / "links.md").read_text() == before def test_import_classifies_into_three_groups(booth): data, _ = booth _seed_board(data) out = run(data, "bench", "import").stdout assert "gone" in out # the booth row, skipped assert "talk" in out # a candidate assert "ftp://x.test/f" in out # refused, with its reason # THE RAW URL BESIDE THE NORMALIZED ID, which is the entire point of the # proposal: five rows of `talk` collapsing to one is only checkable if you # can see which raw URLs produced the one id. This printed the description # instead, so the collapse was invisible in the one place it had to be # visible. All four cold arms found it. assert out.count("https://talk.nh3.phasefinal.com:8092/") >= 2, out def test_bare_apply_refuses_and_writes_nothing(booth): """THE SELECTION GAP — all four cold contract-review arms, independently. `--apply` used to register every candidate, while the same contract says roughly 14 of 35 are reference bookmarks that must STAY on the board. That made the write path do the exact thing the unit's own rationale calls impossible — tell a bench from a bookmark by its URL — silently, to rows that belong where they are. The dry-run prints ids; `--apply` takes the ones the operator names, and refuses without them. Defeating change: restoring the register-everything branch.""" data, _ = booth _seed_board(data) r = run(data, "bench", "import", "--apply") assert r.returncode == REFUSED assert "needs the ids" in r.stderr assert not (data / ".benches.json").exists(), "a bare --apply wrote the registry" def test_apply_refuses_an_id_that_is_not_a_candidate(booth): data, _ = booth _seed_board(data) r = run(data, "bench", "import", "--apply", "http://not-on-the-board/") assert r.returncode == REFUSED assert "not a candidate id" in r.stderr assert not (data / ".benches.json").exists() def test_apply_registers_ONLY_the_named_ids(booth): """The bookmark stays a bookmark unless the operator says otherwise.""" data, _ = booth _seed_board(data) talk = normalize_bench_url_cli("https://talk.nh3.phasefinal.com:8092/") assert run(data, "bench", "import", "--apply", talk).returncode == OK ls = run(data, "bench", "ls").stdout assert "peedlar" not in ls, "an unnamed candidate was registered anyway" assert len([l for l in ls.splitlines() if "talk" in l]) == 1 def test_import_apply_collapses_the_repost(booth): data, _ = booth _seed_board(data) talk = normalize_bench_url_cli("https://talk.nh3.phasefinal.com:8092/") repo = normalize_bench_url_cli("https://gitea.phasefinal.com/vh/peedlar") assert run(data, "bench", "import", "--apply", talk, repo).returncode == OK ls = run(data, "bench", "ls").stdout # ONE ROW, counted by line: "talk" appears in both the name and the # hostname, so a substring count would read 2 for a correctly collapsed row. assert len([l for l in ls.splitlines() if "talk" in l]) == 1, ls assert "peedlar" in ls assert "gone" not in ls, "a booth row was imported as a bench" def test_nothing_in_the_unit_touches_links_md(booth): """INV-8. Defeating change: `import --apply` tidying up the rows it consumed. The whole CLI surface runs against one board and the file must come out byte-identical.""" import hashlib data, _ = booth board = _seed_board(data) before = hashlib.sha256((board / "links.md").read_bytes()).hexdigest() run(data, "link", "http://10.100.10.50:8090/b/x/", "refused") run(data, "bench", "import") run(data, "bench", "import", "--apply") # refused, writes nothing run(data, "bench", "import", "--apply", normalize_bench_url_cli("https://talk.nh3.phasefinal.com:8092/")) run(data, "bench", "add", "http://new.test/", "new") run(data, "bench", "state", "http://new.test/", "retired") run(data, "bench", "ls") # a read verb can truncate too run(data, "bench", "rm", "http://new.test/") after = hashlib.sha256((board / "links.md").read_bytes()).hexdigest() assert before == after def test_link_fails_CLOSED_when_the_booth_check_cannot_run(booth, tmp_path): """A guard that fails open is not a guard. If `booth.links` cannot be imported, `booth link` must post NOTHING and say why — not append the row it could not classify, and not abort with a bare traceback. Defeating change: dropping the `|| pred_rc=$?` handling, which under `set -e` aborts with a Python traceback (safe, but unactionable), or treating a failed check as "not a booth" (unsafe — fails open).""" data, _ = booth lone = tmp_path / "lone" / "scripts" lone.mkdir(parents=True) (lone / "booth").write_text(SCRIPT.read_text()) (lone / "booth").chmod(0o755) r = subprocess.run([str(lone / "booth"), "link", "https://ok.test/x", "a bookmark"], capture_output=True, text=True, cwd="/tmp", timeout=30, env={**os.environ, "BOOTH_DATA_DIR": str(data), "BOOTH_URL": "http://booth.invalid"}) assert r.returncode != OK assert "could not check" in r.stderr, r.stderr assert not (data / "links" / "links.md").exists(), "a row landed despite an unusable check" def test_a_credential_never_reaches_the_board(booth): """`normalize_bench_url` refuses userinfo for a bench; `booth link` was the door this unit did not touch, and the board renders on an unauthenticated LAN surface. Cold contract panel, groa solo. A deliberate small widening of the unit, named rather than smuggled.""" data, _ = booth r = run(data, "link", "https://user:hunter2@x.test/p", "leaky") assert r.returncode != OK assert "credentials" in r.stderr assert not (data / "links").exists(), "a credentialed URL created the board" def test_the_append_happens_INSIDE_the_lock(booth): """Cold bug-hunt panel, hulda solo. `flock LOCK printf ... >> board` reads as locked and is not: the SHELL opens the append fd while parsing, before flock acquires. A concurrent `unlink` rewriting the board in that window replaces the inode, the old fd keeps pointing at the unlinked one, and the append succeeds, reports success, and vanishes. Proved by holding the lock: if the open were outside it, `booth link` would write and exit while blocked. Defeating change: reverting to the bare `flock LOCK printf ... >>` form, under which this test writes the row. """ import fcntl data, _ = booth board = data / "links" board.mkdir(parents=True) (board / "links.md").write_text("") lock = board / ".links.lock" lock.touch() with lock.open("r+") as lf: fcntl.flock(lf, fcntl.LOCK_EX) try: # subprocess.run directly: `run()` pins timeout=30 itself. with pytest.raises(subprocess.TimeoutExpired): subprocess.run( [str(SCRIPT), "link", "https://ok.test/x", "blocked"], capture_output=True, text=True, timeout=5, env={**os.environ, "BOOTH_DATA_DIR": str(data), "BOOTH_URL": "http://booth.invalid"}) finally: fcntl.flock(lf, fcntl.LOCK_UN) assert (board / "links.md").read_text() == "", \ "the row was appended while another writer held the lock" def test_blur_with_no_files_fogs_the_whole_booth(booth): """The Desk shows up to four images from EVERY booth on the page the operator opens first, so a booth that should not be glanced at has to say so as a booth — and the session that posts it is the one that knows. Seventeen handles call this script; a verb here is how they self-blur at post time without waiting for anyone to click anything.""" data, b = booth (b / "a.png").write_bytes(b"x") out = run(data, "blur", "b") assert out.returncode == 0, out.stderr assert (b / ".blurbooth").exists() assert "whole booth blurred" in out.stdout out = run(data, "unblur", "b") assert out.returncode == 0, out.stderr assert not (b / ".blurbooth").exists() def test_unblurring_the_booth_keeps_per_item_choices(booth): """COMPOSES, never overrides — the same promise the resolver makes. An agent's per-item blur must survive the booth flag being cleared. Defeating change: `unblur ` also clearing `.blurred`.""" data, b = booth for n in ("a.png", "b.png"): (b / n).write_bytes(b"x") run(data, "blur", "b", "a.png") run(data, "blur", "b") run(data, "unblur", "b") assert not (b / ".blurbooth").exists() # Read through the reader, not the bytes: `.blurred` became a JSON array # (the round-trip fix, operator-ruled 2026-09-23), and this test is about # the per-item choice surviving, not about the file's format. import sys sys.path.insert(0, str(pathlib.Path(__file__).parent.parent)) from booth.blur import read_blurred assert read_blurred(b) == {"a.png"}