/heid-code-review panel 01M35CK8YKEKMV7T15JXEF6A8N, verdict NOT drift-zero. Three findings arrived from all four arms independently, and they share a shape: a contract clause written as prose and never converted into an assertion. That is the lens working. - The panel dropped the added date the contract promised to show. - `bench ls` printed no ids, and the URL it printed was truncated to 52 columns so the line was not pasteable into `bench state|rm`. The test's docstring claimed it printed ids and asserted nothing of the kind. - `bench import` printed the description instead of the raw URL beside each normalized id, hiding the collapse the clause exists to expose. - An IPv6 literal lost its brackets: http://[::1]:8080/a normalized to http://::1:8080/a, a broken identity that no re-post can match. Bracketed literals are re-wrapped; an unbracketed one is refused rather than guessed. - A deeply-nested JSON RecursionError escaped read_benches' except pair. The byte cap does not help -- 200k open brackets is 200 KB. - An empty board hid the whole benches panel, registration form included. - The link refusal classified by captured-text emptiness, which bash can erase; it now answers with a B:/N sentinel so no name reads as "not a booth". INV-4's tie-break falsifier could not fail: _write_all serializes with sort_keys=True, so both insertion orders came back already id-sorted and removing the tie-break left the test green. It now calls order_benches directly. Same class as the five vacuous U4 falsifiers, found by a cold reader rather than by us. Also from the arms' per-invariant vacuity pass: INV-6 had no vector pinning a non-default port as part of the identity; INV-3 asserted only that links/ was absent; INV-8's hashed sequence omitted a read verb; INV-9's AST walk is defeated by a string import. All closed. Contract amended where the code was right: `updated` means last mutation, the id cap is write-only because the id is the locator controls post back, INV-8's file list includes the lock sidecar it always mandated. Every line number is out of the prose -- the panel found two already stale. 565 -> 593 tests. Nothing declined.
556 lines
25 KiB
Python
556 lines
25 KiB
Python
"""`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 <name>` 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."""
|
|
data, _ = booth
|
|
run(data, "bench", "add", "http://x.test/p/", "ex")
|
|
assert run(data, "bench", "state", "http://x.test/p/", "retired").returncode == OK
|
|
assert run(data, "bench", "rm", "http://x.test/p/").returncode == OK
|
|
|
|
|
|
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/) <sub>· x · 2026-09-01 00:00</sub>\n"
|
|
"- [talk](https://talk.nh3.phasefinal.com:8092/) <sub>· x · 2026-09-01 00:00</sub>\n"
|
|
"- [talk again](https://talk.nh3.phasefinal.com:8092/) <sub>· x · 2026-09-02 00:00</sub>\n"
|
|
"- [a repo](https://gitea.phasefinal.com/vh/peedlar) <sub>· x · 2026-09-03 00:00</sub>\n"
|
|
"- [bad](ftp://x.test/f) <sub>· x · 2026-09-04 00:00</sub>\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_import_apply_collapses_the_repost(booth):
|
|
data, _ = booth
|
|
_seed_board(data)
|
|
assert run(data, "bench", "import", "--apply").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")
|
|
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"
|