Vuong Hoang
a48ef83ef5
The index card showed a name, an item count and a countdown, and nothing
the poster chose. An agent with something to show therefore had no way to
make the booth say "look at this" and posted a URL to the link board
instead — which is why 145 of that board's 210 rows (69%) ended up
pointing at booths that had already been swept. The board was absorbing a
job it was never shaped for. This is the shape.
Each booth carries `.booth.json` — {handle, title, why, created} — written
by the CLI from $ALTHING_HANDLE, and the provenance line renders on both
index lanes and on the booth page header.
WHAT IS WHERE
- booth/manifest.py, stdlib-only and importing nothing from booth.* either:
scripts/booth imports it under the system python3 with no venv, and a
cross-import between two stdlib-only modules is a second way for that
invariant to break. It joins the shared test_stdlib_only list and keeps
a stricter copy of its own.
- The read is lenient and cannot raise. list_booths touches every booth on
every index load, so a manifest that cannot be parsed costs that booth's
provenance and nothing else. That is the v0.2.2 lesson applied before the
same mistake rather than after it.
- Absent and damaged render differently — `unannounced` and `unreadable`.
Folding "cannot be read" into "never said" would hide the one case
somebody has to go and fix.
- Re-announcing preserves `created`. A second `booth add` sharpening the
why is not a second appearance of the booth.
- The write is atomic (invariant 5); the temp file is itself a dotfile, so
no listing can see it mid-write.
THREE OPERATOR CALLS, 2026-09-22
Flags on the existing new/add verbs rather than a separate `announce` verb
(a second step is the step that gets forgotten, which is the rot's own
mechanism). Unannounced booths get a quiet marker rather than nothing — the
convention is only adoptable if the gap is visible. U5 adds provenance only
and does NOT add a second index ordering keyed on announcement time; that
is a different surface needing its own stated rule, parked for v1.1.
NO EXEMPTION LIST
A pickup booth and the standing link board are created by the service, so
they announce themselves with handle `booth`, which is true rather than
manufactured. One rule — a booth with no manifest is unannounced — instead
of a growing set of special cases.
ALSO
tests/test_booth.py's keep/release assertion was slicing the page on the
bare word `boothhead`, which has lived in the stylesheet far longer than
the assertion has; it was reading CSS and passing on luck, and went red the
first time a new rule landed above the old one. Same assertion, aimed at
the markup. A U5 test had the mirror-image bug: pytest derives tmp_path
from the test name and the index renders data_dir, so a test named
`test_an_unannounced_booth_says_so` put the needle in the haystack itself
and passed against a template that did not yet exist.
310 tests (304 before this unit's CLI half). Live service restarted, 26/26
booth pages verified 200, end-to-end smoke through the real CLI.
NOT TAGGED. The cold contract-review panel is still in flight and the
code-review and bug-hunt gates have not run. Tagging with a gate
outstanding is what made v0.2.0 premature.
1197 lines
44 KiB
Python
1197 lines
44 KiB
Python
"""Marks: one primitive for operator judgment attached to an artifact.
|
|
|
|
`pick` (the session asks), `note` (the operator tells), `flag` (the operator
|
|
points at the good ones) — three shapes, one storage model, one read path.
|
|
|
|
See docs/contracts/u2_marks.contract.md.
|
|
"""
|
|
import ast
|
|
import json
|
|
import pathlib
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from booth.asks import ASK_SUFFIX, AskError, build_answer, normalize_ask
|
|
from booth.marks import (
|
|
MARKS_FILE,
|
|
Mark,
|
|
answer_pick,
|
|
as_dict,
|
|
declare_pick,
|
|
marks_for,
|
|
open_marks,
|
|
)
|
|
|
|
|
|
def _single():
|
|
return {"prompt": "Which render wins?", "options": ["A — baseline", "B — async"]}
|
|
|
|
|
|
def _multi():
|
|
return {
|
|
"title": "R18 batch review",
|
|
"questions": [
|
|
{"key": "q1", "prompt": "Render 1?", "options": ["keep", "drop"]},
|
|
{"key": "q2", "prompt": "Render 2?", "options": ["keep", "drop"]},
|
|
],
|
|
}
|
|
|
|
|
|
# ---- slice 1: storage, the record, the read path ----------------------------
|
|
|
|
|
|
def test_pick_round_trips_through_marks_json(tmp_path):
|
|
"""A declared pick lands in ONE file and reads back normalized."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "winner", _single())
|
|
|
|
assert (booth / MARKS_FILE).is_file()
|
|
marks = marks_for(booth)
|
|
assert len(marks) == 1
|
|
m = marks[0]
|
|
assert isinstance(m, Mark)
|
|
assert (m.id, m.shape, m.target) == ("winner", "pick", None)
|
|
assert m.answer is None
|
|
assert m.prompt == "Which render wins?"
|
|
assert m.multi is False
|
|
# normalize_ask emits `notes` (bool); the record carries it as notes_enabled (SR-2)
|
|
assert m.notes_enabled is True
|
|
assert m.notes_label == "notes"
|
|
assert [o["id"] for o in m.options] == ["A — baseline", "B — async"]
|
|
assert len(m.questions) == 1 and m.questions[0]["key"] is None
|
|
|
|
|
|
def test_marks_json_is_a_dotfile_so_it_is_not_an_item(tmp_path):
|
|
"""The dotfile skip in booth_items already covers it. Asserted, not assumed."""
|
|
from booth.items import booth_items
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
(booth / "a.png").write_bytes(b"\x89PNG")
|
|
declare_pick(booth, "winner", _single())
|
|
|
|
assert MARKS_FILE.startswith(".")
|
|
assert [i.rel for i in booth_items(booth)] == ["a.png"]
|
|
|
|
|
|
def test_corrupt_marks_json_renders_as_empty(tmp_path):
|
|
"""A booth with no marks and a booth with a broken mark file both render as
|
|
'no marks'. Neither is a 500."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
(booth / MARKS_FILE).write_text("{not json at all")
|
|
assert marks_for(booth) == []
|
|
|
|
|
|
def test_missing_booth_and_missing_file_are_empty(tmp_path):
|
|
assert marks_for(tmp_path / "nope") == []
|
|
(tmp_path / "b").mkdir()
|
|
assert marks_for(tmp_path / "b") == []
|
|
|
|
|
|
def test_broken_pick_declaration_surfaces_error(tmp_path):
|
|
"""A declaration that cannot be rendered comes back with `error` set, so the
|
|
page can SAY so rather than silently hiding a question the session believes
|
|
it posted. Preserved from list_asks."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
(booth / MARKS_FILE).write_text(json.dumps({
|
|
"version": 1,
|
|
"marks": [{"id": "bad", "shape": "pick", "target": None,
|
|
"created": "2026-09-21T10:00:00-07:00",
|
|
"declaration": {"prompt": "no options"}, "answer": None}],
|
|
}))
|
|
m = marks_for(booth)[0]
|
|
assert m.error is not None
|
|
assert m.id == "bad"
|
|
|
|
|
|
def test_declare_pick_validates_before_writing(tmp_path):
|
|
"""A session cannot land a question the renderer would refuse."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
with pytest.raises(AskError):
|
|
declare_pick(booth, "bad", {"prompt": "only one", "options": ["just me"]})
|
|
assert not (booth / MARKS_FILE).exists()
|
|
|
|
|
|
def test_declare_pick_rejects_a_bad_id(tmp_path):
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
with pytest.raises(AskError):
|
|
declare_pick(booth, "../escape", _single())
|
|
|
|
|
|
def test_redeclaring_a_pick_clears_its_answer(tmp_path):
|
|
"""The question changed, so the old judgment is not an answer to it."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "winner", _single())
|
|
answer_pick(booth, "winner", "A — baseline")
|
|
assert marks_for(booth)[0].answer is not None
|
|
|
|
declare_pick(booth, "winner", {"prompt": "Which one now?", "options": ["X", "Y"]})
|
|
marks = marks_for(booth)
|
|
assert len(marks) == 1
|
|
assert marks[0].answer is None
|
|
assert marks[0].prompt == "Which one now?"
|
|
|
|
|
|
def test_marks_are_oldest_first_by_created(tmp_path):
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "first", _single())
|
|
declare_pick(booth, "second", _single())
|
|
assert [m.id for m in marks_for(booth)] == ["first", "second"]
|
|
|
|
|
|
# ---- openness: THE one predicate (INV-2) ------------------------------------
|
|
|
|
|
|
def test_open_marks_counts_an_unanswered_pick(tmp_path):
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "winner", _single())
|
|
assert [m.id for m in open_marks(marks_for(booth))] == ["winner"]
|
|
|
|
|
|
def test_open_marks_drops_a_complete_answer(tmp_path):
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "winner", _single())
|
|
answer_pick(booth, "winner", "A — baseline")
|
|
assert open_marks(marks_for(booth)) == []
|
|
|
|
|
|
def test_a_partial_answer_counts_as_open(tmp_path):
|
|
"""SR-7, the declared behaviour change. Today the index badge tests
|
|
`answer is None` and so reports a half-answered four-question pick as
|
|
closed, while the panel renders it `◐ partial`. A booth that still owes an
|
|
answer is open — which is also what makes U4's pin rule correct."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "batch", _multi())
|
|
answer_pick(booth, "batch", {"q1": "keep"}) # q2 left alone
|
|
|
|
m = marks_for(booth)[0]
|
|
assert m.answer is not None
|
|
assert m.answer["complete"] is False
|
|
assert m.answer["unanswered"] == ["q2"]
|
|
assert [x.id for x in open_marks(marks_for(booth))] == ["batch"]
|
|
|
|
|
|
# ---- the pick semantics, preserved by NOT rewriting them (INV-4) ------------
|
|
|
|
|
|
def test_partial_answer_is_recorded_not_refused(tmp_path):
|
|
"""Operator ruling 2026-09-09, as a regression test."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "batch", _multi())
|
|
m = answer_pick(booth, "batch", {"q1": "keep", "q2": ""})
|
|
assert m.answer["answers"]["q1"]["label"] == "keep"
|
|
assert "q2" not in m.answer["answers"]
|
|
assert m.answer["unanswered"] == ["q2"]
|
|
assert m.answer["complete"] is False
|
|
|
|
|
|
def test_nothing_to_record_is_still_refused(tmp_path):
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "batch", _multi())
|
|
with pytest.raises(AskError):
|
|
answer_pick(booth, "batch", {"q1": "", "q2": ""})
|
|
|
|
|
|
def test_an_offered_but_invalid_option_is_an_error(tmp_path):
|
|
"""A broken form, not a skipped question."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "winner", _single())
|
|
with pytest.raises(AskError):
|
|
answer_pick(booth, "winner", "C — never offered")
|
|
|
|
|
|
def test_answering_an_unknown_pick_raises(tmp_path):
|
|
"""SR-4: load_ask used to own this AskError. Extracting build_answer moved
|
|
the path to the caller, and a stale form POST must be a 400, not a silent
|
|
no-op."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
with pytest.raises(AskError):
|
|
answer_pick(booth, "ghost", "A — baseline")
|
|
|
|
|
|
def test_re_answering_overwrites(tmp_path):
|
|
"""The mark is the CURRENT judgment, not a log."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "winner", _single())
|
|
answer_pick(booth, "winner", "A — baseline")
|
|
m = answer_pick(booth, "winner", "B — async")
|
|
assert m.answer["label"] == "B — async"
|
|
assert len(marks_for(booth)) == 1
|
|
|
|
|
|
def test_build_answer_matches_the_old_sidecar_document(tmp_path):
|
|
"""INV-4. build_answer is write_answer's logic with the I/O removed; the
|
|
document it produces is the one the sidecar carried."""
|
|
ask = normalize_ask(_single(), "winner")
|
|
doc = build_answer(ask, "B — async", notes="less banding", who="10.0.0.1")
|
|
assert doc["stem"] == "winner"
|
|
assert doc["choice"] == "B — async"
|
|
assert doc["choice_index"] == 1
|
|
assert doc["label"] == "B — async"
|
|
assert doc["notes"] == "less banding"
|
|
assert doc["complete"] is True
|
|
assert doc["unanswered"] == []
|
|
assert doc["answered_by"] == "10.0.0.1"
|
|
assert "answered_at" in doc
|
|
|
|
|
|
def test_build_answer_sources_stem_from_the_ask(tmp_path):
|
|
"""SR-3: the extraction takes a pre-loaded ask and reads `stem` off it."""
|
|
ask = normalize_ask(_multi(), "batch")
|
|
doc = build_answer(ask, {"q1": "keep", "q2": "drop"})
|
|
assert doc["stem"] == "batch"
|
|
assert doc["complete"] is True
|
|
|
|
|
|
# ---- the JSON boundary (SR-8) -----------------------------------------------
|
|
|
|
|
|
def test_as_dict_round_trips_through_json(tmp_path):
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "winner", _single())
|
|
m = marks_for(booth)[0]
|
|
out = json.loads(json.dumps(as_dict(m)))
|
|
assert out["id"] == "winner"
|
|
assert out["shape"] == "pick"
|
|
assert out["answer"] is None
|
|
|
|
|
|
# ---- the stdlib-only invariant (INV-5) --------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize("module", ["marks", "asks", "links", "manifest"])
|
|
def test_stdlib_only(module):
|
|
"""INV-5. scripts/booth imports these under the system python3 with NO venv,
|
|
through a `python3 -c` heredoc that no AST extractor can see — so nothing
|
|
but this test stands between a casual third-party import and `booth ask`
|
|
breaking on every fleet host."""
|
|
# `manifest` also carries a stricter copy in tests/test_manifest.py, which
|
|
# additionally forbids importing `booth.*` — a cross-import between two
|
|
# stdlib-only modules is a second way for this invariant to break.
|
|
src = pathlib.Path(__file__).parent.parent / "booth" / f"{module}.py"
|
|
tree = ast.parse(src.read_text())
|
|
roots = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
roots.update(a.name.split(".")[0] for a in node.names)
|
|
elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module:
|
|
roots.add(node.module.split(".")[0])
|
|
outside = {r for r in roots if r != "booth" and r not in sys.stdlib_module_names}
|
|
assert not outside, f"booth/{module}.py imports non-stdlib: {sorted(outside)}"
|
|
|
|
|
|
# ---- slice 2: flag — the shape with no existing code ------------------------
|
|
|
|
|
|
def test_flag_is_an_upsert_and_unflag_removes(tmp_path):
|
|
"""One item has at most one flag state, so flagging twice is idempotent and
|
|
clearing REMOVES the mark rather than storing `value: false` — an absent
|
|
flag and a false flag are the same judgment."""
|
|
from booth.marks import set_flag
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
m = set_flag(booth, "DSC03389.jpg", True, who="10.0.0.1")
|
|
assert m is not None and m.shape == "flag" and m.flagged is True
|
|
assert m.target == "DSC03389.jpg"
|
|
|
|
set_flag(booth, "DSC03389.jpg", True) # again
|
|
assert len(marks_for(booth)) == 1
|
|
|
|
assert set_flag(booth, "DSC03389.jpg", False) is None
|
|
assert marks_for(booth) == []
|
|
|
|
|
|
def test_unflagging_something_unflagged_is_not_an_error(tmp_path):
|
|
from booth.marks import set_flag
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
assert set_flag(booth, "nope.png", False) is None
|
|
|
|
|
|
def test_flags_on_different_items_coexist(tmp_path):
|
|
from booth.marks import marks_for_target, set_flag
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
set_flag(booth, "a.png", True)
|
|
set_flag(booth, "b.png", True)
|
|
marks = marks_for(booth)
|
|
assert len(marks) == 2
|
|
assert [m.id for m in marks_for_target(marks, "a.png")] == ["flag:a.png"]
|
|
|
|
|
|
def test_a_flag_never_counts_as_open(tmp_path):
|
|
"""Only a pick can owe an answer. A flag is born resolved."""
|
|
from booth.marks import set_flag
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
set_flag(booth, "a.png", True)
|
|
assert open_marks(marks_for(booth)) == []
|
|
|
|
|
|
def test_a_flag_target_cannot_escape_the_booth(tmp_path):
|
|
from booth.marks import set_flag
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
for bad in ("../outside.png", "/etc/passwd", ""):
|
|
with pytest.raises(AskError):
|
|
set_flag(booth, bad, True)
|
|
|
|
|
|
# ---- slice 3: note ----------------------------------------------------------
|
|
|
|
|
|
def test_note_allows_several_per_target(tmp_path):
|
|
"""An item may carry several notes, so each gets a generated id."""
|
|
from booth.marks import marks_for_target, write_note
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
write_note(booth, "a.png", "banding in the gradient")
|
|
write_note(booth, "a.png", "and the highlight clips")
|
|
marks = marks_for(booth)
|
|
assert len(marks) == 2
|
|
assert {m.id for m in marks} == {"note-1", "note-2"}
|
|
assert [m.text for m in marks_for_target(marks, "a.png")] == [
|
|
"banding in the gradient", "and the highlight clips"]
|
|
|
|
|
|
def test_booth_level_note_has_target_none(tmp_path):
|
|
"""The booth itself is a legal target."""
|
|
from booth.marks import marks_for_target, write_note
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
write_note(booth, None, "ship v3, not v4")
|
|
m = marks_for(booth)[0]
|
|
assert m.target is None
|
|
assert marks_for_target(marks_for(booth), None) == [m]
|
|
|
|
|
|
def test_an_empty_note_is_refused(tmp_path):
|
|
"""Nothing to record — the same posture as an empty pick submission."""
|
|
from booth.marks import write_note
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
for blank in ("", " ", "\n\n", None):
|
|
with pytest.raises(AskError):
|
|
write_note(booth, "a.png", blank)
|
|
assert marks_for(booth) == []
|
|
|
|
|
|
def test_a_note_never_counts_as_open(tmp_path):
|
|
from booth.marks import write_note
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
write_note(booth, "a.png", "seen it")
|
|
assert open_marks(marks_for(booth)) == []
|
|
|
|
|
|
def test_delete_mark_is_the_operators_undo(tmp_path):
|
|
from booth.marks import delete_mark, write_note
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
write_note(booth, "a.png", "wrong call")
|
|
assert delete_mark(booth, "note-1") is True
|
|
assert marks_for(booth) == []
|
|
assert delete_mark(booth, "note-1") is False
|
|
|
|
|
|
# ---- the lock (INV-6) -------------------------------------------------------
|
|
|
|
|
|
def test_concurrent_writers_do_not_lose_a_mark(tmp_path):
|
|
"""INV-6. Two processes writing the same booth's marks at once: both land.
|
|
Processes, not threads — the flock is what is under test, and a thread-level
|
|
test would pass on the GIL alone."""
|
|
import subprocess
|
|
import sys as _sys
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
repo = str(pathlib.Path(__file__).parent.parent)
|
|
prog = (
|
|
"import sys; sys.path.insert(0, %r);"
|
|
"from booth.marks import set_flag;"
|
|
"set_flag(%r, sys.argv[1], True)" % (repo, str(booth))
|
|
)
|
|
procs = [subprocess.Popen([_sys.executable, "-c", prog, f"item{i}.png"])
|
|
for i in range(8)]
|
|
for p in procs:
|
|
assert p.wait() == 0
|
|
|
|
got = {m.target for m in marks_for(booth)}
|
|
assert got == {f"item{i}.png" for i in range(8)}, f"lost a mark: {sorted(got)}"
|
|
|
|
|
|
# ---- slice 4: the legacy sidecars — imported, NEVER deleted ----------------
|
|
|
|
|
|
def _sidecar(booth, stem, doc, mtime=None):
|
|
booth.mkdir(parents=True, exist_ok=True)
|
|
p = booth / f"{stem}{ASK_SUFFIX}"
|
|
p.write_text(json.dumps(doc))
|
|
if mtime is not None:
|
|
import os as _os
|
|
_os.utime(p, (mtime, mtime))
|
|
return p
|
|
|
|
|
|
def test_import_reads_a_sidecar_into_marks(tmp_path):
|
|
from booth.marks import import_legacy_asks
|
|
|
|
booth = tmp_path / "b"
|
|
_sidecar(booth, "winner", _single())
|
|
got = import_legacy_asks(booth)
|
|
|
|
assert [m.id for m in got] == ["winner"]
|
|
m = marks_for(booth)[0]
|
|
assert m.shape == "pick"
|
|
assert m.prompt == "Which render wins?"
|
|
assert m.answer is None
|
|
|
|
|
|
def test_import_is_idempotent_and_keeps_the_sidecar(tmp_path):
|
|
"""INV-7. The ROADMAP's non-goal is explicit: no migration that deletes
|
|
anything. Four of these are live and unanswered right now."""
|
|
from booth.marks import import_legacy_asks
|
|
|
|
booth = tmp_path / "b"
|
|
p = _sidecar(booth, "winner", _single())
|
|
import_legacy_asks(booth)
|
|
assert import_legacy_asks(booth) == [] # second run is a no-op
|
|
assert len(marks_for(booth)) == 1
|
|
assert p.is_file(), "the importer deleted the operator's sidecar"
|
|
|
|
|
|
def test_import_carries_an_existing_answer(tmp_path):
|
|
from booth.asks import ANSWER_SUFFIX, build_answer, normalize_ask
|
|
from booth.marks import import_legacy_asks
|
|
|
|
booth = tmp_path / "b"
|
|
_sidecar(booth, "winner", _single())
|
|
# The legacy answer sidecar, written the way the retired `write_answer` did.
|
|
doc = build_answer(normalize_ask(_single(), "winner"), "B — async", notes="less banding")
|
|
(booth / f"winner{ANSWER_SUFFIX}").write_text(json.dumps(doc))
|
|
import_legacy_asks(booth)
|
|
|
|
m = marks_for(booth)[0]
|
|
assert m.answer["label"] == "B — async"
|
|
assert m.answer["notes"] == "less banding"
|
|
assert open_marks(marks_for(booth)) == []
|
|
|
|
|
|
def test_import_does_not_clobber_a_newer_answer(tmp_path):
|
|
"""A stem already present as a mark is skipped outright."""
|
|
from booth.marks import import_legacy_asks
|
|
|
|
booth = tmp_path / "b"
|
|
_sidecar(booth, "winner", _single())
|
|
declare_pick(booth, "winner", _single())
|
|
answer_pick(booth, "winner", "A — baseline")
|
|
|
|
assert import_legacy_asks(booth) == []
|
|
assert marks_for(booth)[0].answer["label"] == "A — baseline"
|
|
|
|
|
|
def test_import_preserves_mtime_ordering(tmp_path):
|
|
"""SR-5. list_asks ordered by file mtime; marks_for orders by the stored
|
|
`created`. Seed it from the sidecar's mtime or the four live asks silently
|
|
reorder on import."""
|
|
from booth.marks import import_legacy_asks
|
|
|
|
booth = tmp_path / "b"
|
|
_sidecar(booth, "later", _single(), mtime=2_000_000_000)
|
|
_sidecar(booth, "earlier", _single(), mtime=1_000_000_000)
|
|
import_legacy_asks(booth)
|
|
assert [m.id for m in marks_for(booth)] == ["earlier", "later"]
|
|
|
|
|
|
def test_import_surfaces_a_broken_sidecar_instead_of_dropping_it(tmp_path):
|
|
"""A question the session believes it posted stays visible."""
|
|
from booth.marks import import_legacy_asks
|
|
|
|
booth = tmp_path / "b"
|
|
_sidecar(booth, "broke", {"prompt": "no options here"})
|
|
import_legacy_asks(booth)
|
|
m = marks_for(booth)[0]
|
|
assert m.id == "broke"
|
|
assert m.error is not None
|
|
|
|
|
|
def test_import_ignores_a_bad_stem(tmp_path):
|
|
"""A filename that could not have been written by `booth ask` is left alone
|
|
rather than becoming a mark with an unusable id."""
|
|
from booth.marks import import_legacy_asks
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
(booth / f"..evil{ASK_SUFFIX}").write_text(json.dumps(_single()))
|
|
assert import_legacy_asks(booth) == []
|
|
|
|
|
|
def test_import_on_a_booth_with_nothing_is_empty(tmp_path):
|
|
from booth.marks import import_legacy_asks
|
|
|
|
assert import_legacy_asks(tmp_path / "nope") == []
|
|
(tmp_path / "b").mkdir()
|
|
assert import_legacy_asks(tmp_path / "b") == []
|
|
|
|
|
|
# ---- the deterministic-order invariant (operator directive, 2026-09-21) ------
|
|
|
|
|
|
def test_marks_order_is_deterministic_across_reads(tmp_path):
|
|
"""Every ordered collection the Booth renders needs a STATED rule, because
|
|
the operator judges positionally — "the third one", "the one after the
|
|
banded one" — and an order that moves between renders misfiles the judgment
|
|
instead of crashing. Marks order by `(created, id)`."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "b-pick", _single())
|
|
from booth.marks import set_flag, write_note
|
|
write_note(booth, "z.png", "later note")
|
|
set_flag(booth, "a.png", True)
|
|
write_note(booth, None, "booth note")
|
|
|
|
first = [m.id for m in marks_for(booth)]
|
|
for _ in range(5):
|
|
assert [m.id for m in marks_for(booth)] == first
|
|
|
|
|
|
def test_same_second_marks_tie_break_on_id(tmp_path):
|
|
"""The `created` stamp is second-resolution, so two marks written inside one
|
|
second would order by whatever json listed them. The id is the tie-break."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
stamp = "2026-09-21T12:00:00-07:00"
|
|
(booth / MARKS_FILE).write_text(json.dumps({"version": 1, "marks": [
|
|
{"id": "zeta", "shape": "note", "target": None, "created": stamp, "text": "z"},
|
|
{"id": "alpha", "shape": "note", "target": None, "created": stamp, "text": "a"},
|
|
]}))
|
|
assert [m.id for m in marks_for(booth)] == ["alpha", "zeta"]
|
|
# and reversing the stored order changes nothing
|
|
(booth / MARKS_FILE).write_text(json.dumps({"version": 1, "marks": [
|
|
{"id": "alpha", "shape": "note", "target": None, "created": stamp, "text": "a"},
|
|
{"id": "zeta", "shape": "note", "target": None, "created": stamp, "text": "z"},
|
|
]}))
|
|
assert [m.id for m in marks_for(booth)] == ["alpha", "zeta"]
|
|
|
|
|
|
def test_open_marks_preserves_the_read_order(tmp_path):
|
|
"""The filtered view must not re-order — the panel and the badge have to
|
|
agree with the list they came from."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
for mid in ("c", "a", "b"):
|
|
declare_pick(booth, mid, _single())
|
|
marks = marks_for(booth)
|
|
assert [m.id for m in open_marks(marks)] == [m.id for m in marks]
|
|
|
|
|
|
# ---- INV-3: the judgment travels, like the caption --------------------------
|
|
|
|
|
|
def _png(p):
|
|
p.write_bytes(
|
|
b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
|
|
b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00"
|
|
b"\x01\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path):
|
|
from fastapi.testclient import TestClient
|
|
|
|
from booth.app import create_app
|
|
return TestClient(create_app(tmp_path, ttl_hours=24, start_sweeper=False)), tmp_path
|
|
|
|
|
|
def test_zoom_carries_the_marks(client):
|
|
"""U1's rule, extended from the caption to the judgment: full size is where
|
|
the operator is actually deciding, so the flag state and the notes have to
|
|
be there."""
|
|
from booth.marks import set_flag, write_note
|
|
|
|
c, data = client
|
|
b = data / "b"
|
|
b.mkdir()
|
|
_png(b / "a.png")
|
|
set_flag(b, "a.png", True)
|
|
write_note(b, "a.png", "banding in the gradient")
|
|
|
|
html = c.get("/b/b/view?f=a.png").text
|
|
assert "banding in the gradient" in html
|
|
assert "✔ flagged" in html
|
|
assert 'action="/b/b/flag"' in html
|
|
|
|
|
|
def test_the_tile_carries_the_marks(client):
|
|
from booth.marks import set_flag, write_note
|
|
|
|
c, data = client
|
|
b = data / "b"
|
|
b.mkdir()
|
|
_png(b / "a.png")
|
|
_png(b / "b.png")
|
|
set_flag(b, "a.png", True)
|
|
write_note(b, "b.png", "this one is soft")
|
|
|
|
html = c.get("/b/b/").text
|
|
assert "✔ flagged" in html and "○ flag" in html # a flagged one and an unflagged one
|
|
assert "this one is soft" in html
|
|
assert 'id="item-a.png"' in html
|
|
|
|
|
|
def test_flag_and_note_round_trip_through_the_browser(client):
|
|
c, data = client
|
|
b = data / "b"
|
|
b.mkdir()
|
|
_png(b / "a.png")
|
|
|
|
assert c.post("/b/b/flag", data={"target": "a.png", "on": "1"},
|
|
follow_redirects=False).status_code == 303
|
|
assert c.post("/b/b/note", data={"target": "a.png", "text": "soft"},
|
|
follow_redirects=False).status_code == 303
|
|
doc = c.get("/b/b/marks.json").json()
|
|
assert {m["shape"] for m in doc["marks"]} == {"flag", "note"}
|
|
assert doc["open"] == [] # neither owes an answer
|
|
|
|
# and the operator can withdraw one
|
|
note_id = next(m["id"] for m in doc["marks"] if m["shape"] == "note")
|
|
assert c.post("/b/b/unmark", data={"mark": note_id},
|
|
follow_redirects=False).status_code == 303
|
|
assert [m["shape"] for m in c.get("/b/b/marks.json").json()["marks"]] == ["flag"]
|
|
|
|
|
|
def test_a_no_op_write_does_not_touch_the_booth(tmp_path):
|
|
"""A booth's TTL is measured from its newest mtime, INCLUDING dotfiles — so
|
|
writing `.marks.json` resets the clock. Marking IS activity and should; a
|
|
write that changes nothing is not activity and must not. Unflagging
|
|
something never flagged would otherwise keep a dead booth alive."""
|
|
from booth.marks import delete_mark, set_flag
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
assert set_flag(booth, "ghost.png", False) is None
|
|
assert delete_mark(booth, "nothing") is False
|
|
assert not (booth / MARKS_FILE).exists(), "a no-op write created the mark file"
|
|
|
|
|
|
def test_a_real_write_then_a_no_op_leaves_the_file_alone(tmp_path):
|
|
import os
|
|
from booth.marks import set_flag
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
set_flag(booth, "a.png", True)
|
|
path = booth / MARKS_FILE
|
|
os.utime(path, (1_000_000_000, 1_000_000_000))
|
|
before = path.stat().st_mtime
|
|
|
|
set_flag(booth, "a.png", True) # idempotent: already flagged
|
|
assert path.stat().st_mtime == before, "an idempotent flag rewrote the file"
|
|
|
|
|
|
# ---- findings from the cross-frontier contract panel, 2026-09-22 -------------
|
|
#
|
|
# Heid panel (thread 01M33VSNFER4N1554G0Y0VC9C8). Four arms, artifact-only.
|
|
|
|
|
|
def test_a_write_over_a_corrupt_marks_file_refuses_instead_of_replacing(tmp_path):
|
|
"""DATA LOSS, shipped in v0.2.0. Found by Kimi (flag 2), converged with Hulda.
|
|
|
|
`marks_for` is deliberately lenient — an unparseable file reads as "no marks"
|
|
so a review page still loads. The write path inherited that leniency through
|
|
the same reader, so the next flag toggle appended one entry to an empty list
|
|
and atomically replaced the file: every judgment in that booth gone, from one
|
|
click, silently.
|
|
|
|
The read stays lenient and the WRITE goes strict. That asymmetry is the fix —
|
|
a page that renders without an annotation is recoverable, a file that
|
|
overwrote the operator's judgment is not, and this repo's standing rule is
|
|
that nothing deletes his data.
|
|
"""
|
|
from booth.marks import MarksCorrupt, set_flag, write_note
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
write_note(booth, "a.png", "judgment one")
|
|
write_note(booth, "b.png", "judgment two")
|
|
raw = (booth / MARKS_FILE).read_text()
|
|
(booth / MARKS_FILE).write_text(raw[: len(raw) // 2]) # truncated mid-write
|
|
|
|
with pytest.raises(MarksCorrupt):
|
|
set_flag(booth, "c.png", True)
|
|
|
|
# The damaged bytes are still on disk — untouched, recoverable by hand.
|
|
assert (booth / MARKS_FILE).read_text() == raw[: len(raw) // 2]
|
|
# And the read path is still lenient, so the page renders rather than 500s.
|
|
assert marks_for(booth) == []
|
|
|
|
|
|
def test_an_absent_or_empty_marks_file_is_not_corrupt(tmp_path):
|
|
"""The strict write path must not mistake "nothing yet" for "damaged"."""
|
|
from booth.marks import set_flag
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
assert set_flag(booth, "a.png", True) is not None # no file at all
|
|
(booth / MARKS_FILE).write_text("")
|
|
assert set_flag(booth, "b.png", True) is not None # zero bytes
|
|
(booth / MARKS_FILE).write_text('{"version": 1, "marks": []}')
|
|
assert set_flag(booth, "c.png", True) is not None # valid but empty
|
|
|
|
|
|
def test_a_pick_can_target_one_item(tmp_path):
|
|
"""Found by Hulda (flag 1), converged with Regin.
|
|
|
|
`Mark.target` carries an item rel, `marks_for_target` retrieves by it, and
|
|
the panel template already renders "on <item>" for a pick — but
|
|
`declare_pick` had no target parameter, so a session could not actually
|
|
produce one. A question about ONE artifact is the 2026-09-09 ruling's whole
|
|
point; the record supported it and the door was missing.
|
|
"""
|
|
from booth.marks import marks_for_target
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "which-crop", _single(), target="v3/DSC03389.jpg")
|
|
m = marks_for(booth)[0]
|
|
assert m.target == "v3/DSC03389.jpg"
|
|
assert [x.id for x in marks_for_target(marks_for(booth), "v3/DSC03389.jpg")] == ["which-crop"]
|
|
# and it still answers normally
|
|
answer_pick(booth, "which-crop", "A — baseline")
|
|
assert marks_for(booth)[0].answer["complete"] is True
|
|
|
|
|
|
def test_a_pick_target_cannot_escape_the_booth(tmp_path):
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
for bad in ("../outside.png", "/etc/passwd"):
|
|
with pytest.raises(AskError):
|
|
declare_pick(booth, "p", _single(), target=bad)
|
|
|
|
|
|
def test_redeclaring_a_pick_may_move_its_target(tmp_path):
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "p", _single(), target="a.png")
|
|
declare_pick(booth, "p", _single(), target="b.png")
|
|
assert marks_for(booth)[0].target == "b.png"
|
|
|
|
|
|
def test_import_adopts_a_legacy_answer_for_an_already_declared_pick(tmp_path):
|
|
"""Found by Gróa (flag 10).
|
|
|
|
The idempotence rule skipped any stem already present as a mark. If a
|
|
session had re-declared that stem through marks (so the mark exists, still
|
|
unanswered) while the operator's answer sat in the legacy sidecar, the import
|
|
skipped and that answer was stranded on disk forever — with the read path
|
|
forbidden from looking at sidecars. Adopting the answer preserves both rules:
|
|
idempotent, and never clobbers a NEWER judgment.
|
|
"""
|
|
from booth.asks import ANSWER_SUFFIX, build_answer, normalize_ask
|
|
from booth.marks import import_legacy_asks
|
|
|
|
booth = tmp_path / "b"
|
|
_sidecar(booth, "winner", _single())
|
|
doc = build_answer(normalize_ask(_single(), "winner"), "B — async", notes="from the sidecar")
|
|
(booth / f"winner{ANSWER_SUFFIX}").write_text(json.dumps(doc))
|
|
declare_pick(booth, "winner", _single()) # re-declared, unanswered
|
|
assert marks_for(booth)[0].answer is None
|
|
|
|
import_legacy_asks(booth)
|
|
got = marks_for(booth)[0]
|
|
assert got.answer is not None, "the legacy answer was stranded"
|
|
assert got.answer["choice"] == "B — async"
|
|
assert open_marks(marks_for(booth)) == []
|
|
|
|
|
|
def test_import_never_overwrites_an_answer_made_through_marks(tmp_path):
|
|
"""The other half of the same rule: a judgment recorded SINCE the sidecar
|
|
outranks it, and adoption must not reach back over it."""
|
|
from booth.asks import ANSWER_SUFFIX, build_answer, normalize_ask
|
|
from booth.marks import import_legacy_asks
|
|
|
|
booth = tmp_path / "b"
|
|
_sidecar(booth, "winner", _single())
|
|
old = build_answer(normalize_ask(_single(), "winner"), "A — baseline")
|
|
(booth / f"winner{ANSWER_SUFFIX}").write_text(json.dumps(old))
|
|
declare_pick(booth, "winner", _single())
|
|
answer_pick(booth, "winner", "B — async") # the operator changed his mind
|
|
|
|
import_legacy_asks(booth)
|
|
assert marks_for(booth)[0].answer["choice"] == "B — async"
|
|
|
|
|
|
def test_the_doc_view_carries_the_marks(client):
|
|
"""INV-3's third surface — flagged 4/4 by the panel as named in the rule but
|
|
covered by no test, so shipping it unmarked would have passed."""
|
|
from booth.marks import write_note
|
|
|
|
c, data = client
|
|
b = data / "b"
|
|
b.mkdir()
|
|
(b / "notes.md").write_text("# report\n\nprose here\n")
|
|
write_note(b, "notes.md", "this section is wrong")
|
|
|
|
html = c.get("/b/b/view?f=notes.md").text
|
|
assert "prose here" in html
|
|
assert "this section is wrong" in html
|
|
|
|
|
|
def test_a_corrupt_marks_file_gives_the_browser_a_409_not_a_500(client):
|
|
"""The request was fine and the service is fine — the state on disk is not,
|
|
and the refusal is deliberate. A 500 would read as "the Booth is broken" and
|
|
send the operator looking for something to restart."""
|
|
c, data = client
|
|
b = data / "b"
|
|
b.mkdir()
|
|
from booth.marks import write_note
|
|
write_note(b, "a.png", "keep me")
|
|
(b / MARKS_FILE).write_text("{truncated")
|
|
|
|
r = c.post("/b/b/flag", data={"target": "a.png", "on": "1"}, follow_redirects=False)
|
|
assert r.status_code == 409
|
|
body = r.json()
|
|
assert "cannot be read" in body["error"] and body["fix"]
|
|
# the page still renders, so the operator can see the booth at all
|
|
assert c.get("/b/b/").status_code == 200
|
|
assert c.get("/b/b/marks.json").status_code == 200
|
|
|
|
|
|
# ---- findings from the cross-frontier BUG-HUNT panel, 2026-09-22 -------------
|
|
#
|
|
# Heid panel (thread 01M33XEC1H0298C0D968FWBN7A). Four arms, artifact-only,
|
|
# diff-scoped. The headline was 4/4 convergent and none of it had a guard: the
|
|
# panel's own mutation tables showed the lock lifecycle SURVIVED every existing
|
|
# test, because `test_a_no_op_write_does_not_touch_the_booth` asserts only that
|
|
# `.marks.json` is absent and never looks at the lock or at the clock the
|
|
# sweeper actually reads.
|
|
|
|
|
|
def test_the_lock_file_is_never_unlinked(tmp_path):
|
|
"""The lock must outlive the operation that created it.
|
|
|
|
`flock` binds to an INODE, not to a path. Unlinking `.marks.lock` while a
|
|
second writer is blocked on it leaves that writer holding an exclusive lock
|
|
on a deleted inode — and the next writer along creates a FRESH lock file and
|
|
takes it immediately. Two processes then run the read-modify-write
|
|
concurrently and the later `os.replace` drops the earlier one's mark, with
|
|
no error anywhere. Both of them obeyed the protocol.
|
|
|
|
The cleanup existed to keep a no-op from leaving a lock file as its only
|
|
trace. That is a tidiness goal, and it bought a lost-update race.
|
|
"""
|
|
from booth.marks import MARKS_LOCK, set_flag
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
assert set_flag(booth, "ghost.png", False) is None # a no-op
|
|
assert (booth / MARKS_LOCK).exists(), "the no-op path unlinked the lock file"
|
|
|
|
|
|
def test_a_no_op_does_not_reset_the_ttl_clock(tmp_path):
|
|
"""The property the no-op guard actually exists for, asserted against the
|
|
clock the sweeper reads instead of against one file's absence.
|
|
|
|
Creating or removing a directory entry bumps the DIRECTORY's mtime, and
|
|
`_newest_mtime` seeds from exactly that. So `touch` + `unlink` of the lock
|
|
reset the booth's age to zero while leaving no trace behind — the comment on
|
|
the create-only guard reasons about the lock FILE's mtime and misses that
|
|
the directory moved underneath it. Repeated, it kept a dead booth alive
|
|
forever, which is the precise outcome the guard was written to prevent.
|
|
"""
|
|
import os
|
|
|
|
from booth.app import booth_age_seconds
|
|
from booth.marks import delete_mark, set_flag
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
old = 1_000_000_000
|
|
os.utime(booth, (old, old))
|
|
|
|
set_flag(booth, "ghost.png", False) # no-op: never flagged
|
|
delete_mark(booth, "nothing") # no-op: no such mark
|
|
|
|
age = booth_age_seconds(booth, now=old + 90_000)
|
|
assert age > 86_400, f"a no-op reset the TTL clock (age fell to {age:.0f}s)"
|
|
|
|
|
|
def test_a_real_mark_still_resets_the_ttl_clock(tmp_path):
|
|
"""The other half of the same rule, so the fix cannot overshoot into
|
|
'marking is never activity'. Marking IS activity and must reset the clock;
|
|
only a write that changes nothing must not."""
|
|
import os
|
|
|
|
from booth.app import booth_age_seconds
|
|
from booth.marks import set_flag
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
old = 1_000_000_000
|
|
os.utime(booth, (old, old))
|
|
|
|
set_flag(booth, "a.png", True) # a real mark
|
|
|
|
assert booth_age_seconds(booth, now=old + 90_000) < 86_400
|
|
|
|
|
|
def test_a_non_string_note_text_does_not_crash_the_read(tmp_path):
|
|
"""`_clean_text` did `(text or "").replace(...)`, so a stored `text` that is
|
|
valid JSON but not a string raised AttributeError out of the READ path.
|
|
|
|
That is not a marks bug, it is an INDEX bug: `list_booths` reads every
|
|
booth's marks on every page load, so one poisoned file took down `/` and
|
|
`/healthz` for all 25 booths. The module's stated posture is that a mark it
|
|
cannot read renders as broken, never as a 500.
|
|
"""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
(booth / MARKS_FILE).write_text(json.dumps({
|
|
"version": 1,
|
|
"marks": [{"id": "n1", "shape": "note", "text": 7,
|
|
"created": "2026-09-21T00:00:00+00:00"}],
|
|
}))
|
|
|
|
marks = marks_for(booth)
|
|
assert len(marks) == 1
|
|
assert marks[0].error, "a poisoned note read clean instead of reading broken"
|
|
|
|
|
|
def test_a_non_string_created_does_not_crash_the_sort(tmp_path):
|
|
"""`marks_for` sorts on `(created, id)`. A stored `created` of the wrong type
|
|
made that comparison raise TypeError — same blast radius as the note above,
|
|
reached through the sort rather than through hydration."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
(booth / MARKS_FILE).write_text(json.dumps({
|
|
"version": 1,
|
|
"marks": [
|
|
{"id": "a", "shape": "note", "text": "fine",
|
|
"created": "2026-09-21T00:00:00+00:00"},
|
|
{"id": "b", "shape": "note", "text": "also fine", "created": 17},
|
|
],
|
|
}))
|
|
|
|
marks = marks_for(booth)
|
|
assert len(marks) == 2
|
|
# An unreadable mark loses its `created` and so sorts FIRST — the stated
|
|
# rule is `("", id)` against `(created, id)`. A mark nobody can read is the
|
|
# one that wants looking at, and the alternative is it landing at an
|
|
# arbitrary position in the middle of the panel.
|
|
assert [m.id for m in marks] == ["b", "a"]
|
|
assert marks[0].error and not marks[1].error
|
|
|
|
|
|
def test_legacy_import_order_survives_same_second_mtimes(tmp_path):
|
|
"""ROADMAP states the legacy import's order is `(mtime, name)`. It was
|
|
stamping `created` at whole-second resolution, so two sidecars written in
|
|
the same second lost the fractional part that distinguished them and
|
|
`marks_for`'s `(created, id)` tie-break silently re-sorted them into
|
|
alphabetical order — reversing the pair the importer had just ordered.
|
|
|
|
Deterministic order is a v1 invariant precisely because the operator refers
|
|
to things positionally. An order that is stated and not kept is worse than
|
|
one that was never claimed.
|
|
"""
|
|
import os
|
|
|
|
from booth.marks import import_legacy_asks
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
for stem in ("zeta", "alpha"):
|
|
(booth / f"{stem}{ASK_SUFFIX}").write_text(json.dumps(_single()))
|
|
# Same whole second, different fractions: `zeta` is OLDER and must come first.
|
|
os.utime(booth / f"zeta{ASK_SUFFIX}", (1_700_000_000.10, 1_700_000_000.10))
|
|
os.utime(booth / f"alpha{ASK_SUFFIX}", (1_700_000_000.90, 1_700_000_000.90))
|
|
|
|
imported = [m.id for m in import_legacy_asks(booth)]
|
|
assert imported == ["zeta", "alpha"], "the importer's own order is wrong"
|
|
assert [m.id for m in marks_for(booth)] == imported, (
|
|
"the read path re-sorted what the importer ordered"
|
|
)
|
|
|
|
|
|
def test_the_index_survives_a_poisoned_marks_file(client):
|
|
"""The blast radius, asserted where it actually hurts.
|
|
|
|
`list_booths` reads every booth's marks on every index load and `/healthz`
|
|
does the same. One hand-edited or foreign-written `.marks.json` therefore
|
|
took down the front page for all 25 booths — the single-booth failure the
|
|
lenient reader exists to contain, escaping the booth it belongs to.
|
|
"""
|
|
c, data = client
|
|
good = data / "good"
|
|
good.mkdir()
|
|
_png(good / "a.png")
|
|
bad = data / "bad"
|
|
bad.mkdir()
|
|
(bad / MARKS_FILE).write_text(json.dumps({
|
|
"version": 1,
|
|
"marks": [{"id": "n1", "shape": "note", "text": {"oops": True}, "created": 3}],
|
|
}))
|
|
|
|
assert c.get("/").status_code == 200
|
|
assert c.get("/healthz").status_code == 200
|
|
assert c.get("/b/bad/").status_code == 200
|
|
|
|
|
|
def test_answer_treats_a_non_string_notes_field_as_no_notes(client):
|
|
"""`booth_note` guards `text` with `isinstance(..., str)`; `booth_answer`
|
|
passed `notes` straight to `_clean_notes`, which calls `.replace` on it. A
|
|
multipart FILE part named `notes` is a str to nobody, so the route 500'd on
|
|
hostile-but-legal input where its sibling handled the same class of value.
|
|
|
|
Both routes now read the field the same way: a value that is not text is no
|
|
value. The CHOICE is the judgment and it still lands — throwing the whole
|
|
answer away over a junk optional field would be the wrong trade."""
|
|
c, data = client
|
|
b = data / "b"
|
|
b.mkdir()
|
|
declare_pick(b, "winner", _single())
|
|
|
|
r = c.post(
|
|
"/b/b/answer",
|
|
data={"ask": "winner", "choice": "A — baseline"},
|
|
files={"notes": ("n.txt", b"surprise", "text/plain")},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 303
|
|
mark = next(m for m in marks_for(b) if m.id == "winner")
|
|
assert mark.answer["choice"] == "A — baseline"
|
|
assert not mark.answer.get("notes")
|
|
|
|
|
|
def test_an_inline_doc_tile_offers_a_note_control(client):
|
|
"""Three item branches, two of them call `marknotes`. The doc branch got the
|
|
flag button and not the note field, so the operator could point at a report
|
|
and not write down why — on the one item kind whose whole purpose is prose.
|
|
|
|
This is the exact failure the `blurtoggle` macro comment names ("patched two
|
|
of three"), recurring on the macro that was written to prevent it.
|
|
"""
|
|
c, data = client
|
|
b = data / "b"
|
|
b.mkdir()
|
|
(b / "report.md").write_text("# report\n\nprose here\n")
|
|
|
|
html = c.get("/b/b/").text
|
|
assert 'value="report.md"' in html, "the doc tile has no mark controls at all"
|
|
# `marknotes`' add-field, which only that macro emits. The booth-level panel
|
|
# has its own note form, so the presence of /note on the page proves nothing.
|
|
assert 'placeholder="a note on this item"' in html, (
|
|
"an inline doc tile has no way to add a note"
|
|
)
|
|
|
|
|
|
def test_the_marks_panel_survives_a_booth_that_also_has_a_link_board(client):
|
|
"""The board booth renders as a board instead of a gallery, which is right —
|
|
but the suppression was unconditional, so a pick declared on a booth that
|
|
happens to carry a `links.md` had no form to answer it and no way to say so."""
|
|
c, data = client
|
|
b = data / "b"
|
|
b.mkdir()
|
|
(b / "links.md").write_text("- [a thing](http://example.invalid) <sub>· who · when</sub>\n")
|
|
declare_pick(b, "winner", _single())
|
|
|
|
html = c.get("/b/b/").text
|
|
assert "Which render wins?" in html, "a pick on a board booth was unanswerable"
|
|
|
|
|
|
def test_the_zoom_view_does_not_navigate_away_from_a_note_being_typed(client):
|
|
"""The viewer's arrow keys move between images and Escape goes back. The
|
|
note textarea landed in the same page, and the handler is on `document`, so
|
|
an arrow key meant for the caret threw away the draft instead of moving it.
|
|
|
|
Asserted structurally: the handler must bail on events from an editable
|
|
target. There is no browser in this suite, and a guard nobody can test is
|
|
exactly how this shipped."""
|
|
c, data = client
|
|
b = data / "b"
|
|
b.mkdir()
|
|
_png(b / "a.png")
|
|
|
|
js = c.get("/b/b/view?f=a.png").text
|
|
assert "isEditable" in js, "the viewer's key handler has no editing guard"
|
|
|
|
|
|
@pytest.mark.parametrize("route", ["booth_answer", "booth_note", "booth_flag",
|
|
"booth_unmark", "booth_import_asks"])
|
|
def test_mark_writes_do_not_block_the_event_loop(route):
|
|
"""Every mark write takes a blocking `flock` and does synchronous disk I/O.
|
|
In an `async def` handler that runs ON the event loop, so a lock held by
|
|
another process — the CLI mid-`marks-import`, a second browser tab — freezes
|
|
every other request, including the index and `/healthz`.
|
|
|
|
Structural, like `test_stdlib_only`, and for the same reason: the failure is
|
|
a property of where the call runs, which no single-process response
|
|
assertion can see. The rule is that an async mark-write handler hands the
|
|
locked section to a worker thread and never calls the writer inline.
|
|
"""
|
|
src = pathlib.Path(__file__).parent.parent / "booth" / "app.py"
|
|
fn = next(
|
|
n for n in ast.walk(ast.parse(src.read_text()))
|
|
if isinstance(n, ast.AsyncFunctionDef) and n.name == route
|
|
)
|
|
writers = {"answer_pick", "write_note", "set_flag", "delete_mark",
|
|
"import_legacy_asks"}
|
|
for node in ast.walk(fn):
|
|
if not isinstance(node, ast.Call):
|
|
continue
|
|
name = getattr(node.func, "id", None) or getattr(node.func, "attr", None)
|
|
if name in writers:
|
|
pytest.fail(f"{route} calls {name}() on the event loop; "
|
|
"dispatch it through run_in_threadpool")
|
|
|
|
|
|
def test_an_unreadable_mark_is_visible_on_the_page(client):
|
|
"""Surviving the poisoned file is half of it. A note whose stored `text` is
|
|
unreadable hydrates with empty text, and the panel rendered that as an empty
|
|
`<pre>` with a withdraw button beside it — which looks exactly like a note
|
|
the operator wrote and then cleared.
|
|
|
|
`_hydrate`'s own docstring forbids this for picks ("a broken question the
|
|
session believes it posted has to be visible — silently hiding it is the one
|
|
outcome nobody can debug"). It is the same argument for every shape."""
|
|
c, data = client
|
|
b = data / "b"
|
|
b.mkdir()
|
|
(b / MARKS_FILE).write_text(json.dumps({
|
|
"version": 1,
|
|
"marks": [{"id": "n1", "shape": "note", "text": {"oops": True},
|
|
"created": "2026-09-21T00:00:00+00:00"}],
|
|
}))
|
|
|
|
html = c.get("/b/b/").text
|
|
assert "⚠ broken" in html, "an unreadable mark rendered as an empty note"
|
|
assert "n1" in html
|