vh
e702be4e1a
Pre-existing, measured at 42ea67f, so it predates U3. `_hydrate` checked only
that `answer` was a dict and never that `answer["answers"]` was one, so
`marks_for` and `hold_read` both reported the mark healthy with no read error
-- and `_ask_inline.html` then asked a list for `.get`. The v0.2.2 lesson was
half-implemented: that outage was a file that could not be PARSED and the
reader was made lenient, while this one parses perfectly and breaks one layer
further in, at render, where no leniency existed.
Closed at the hydration boundary rather than by a third copy of the guard --
one predicate, one place, every surface inherits it. Only the multi case is
checked, because only the multi case indexes; requiring `answers`
unconditionally would break every single-question pick, and that direction has
its own test. Measured before and after: gallery and marks pages 500 -> 200,
the error visible on the page, the booth's other healthy pick untouched.
The placement was the one open operator question of the session. It was
surfaced three times without a ruling, so it is taken under a stated assumption
and is cheap to move: the whole fix is one condition in one function.
Two things fell out of it worth more than the fix.
`_safe_fragments` no longer has a reachable natural trigger. Probed every wrong
answer shape a .marks.json can carry: `answers` as a list, a string or null all
become hydration errors now, and a wrong-typed value INSIDE `answers` renders
without raising, because Jinja absorbs attribute access on a non-mapping. U3's
guard is a pure backstop, and its test now says so and trips it synthetically
through the shared macro module rather than asserting a path nothing reaches.
A guard tested by an unreachable input is an untested guard.
And that guard's handler could not survive the failure it was handling: it
caught a raising `_pick_fragments` and rebuilt the broken-ask box through the
SAME macro module that had just raised, so whenever `whole` was the broken
thing it re-raised and took the whole report. Found by accident while building
the falsifier. Fixed, with its own test.
Both new falsifiers were verified RED against their defeating change rather
than assumed.
607 -> 611 tests.
1431 lines
54 KiB
Python
1431 lines
54 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", "benches"])
|
|
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):
|
|
# `node.level > 0` is a RELATIVE import (`from . import marks`),
|
|
# which has no `module` root to inspect and used to slip through
|
|
# this walk entirely. It cannot reach outside the package, so it is
|
|
# stdlib-safe by construction — but it is recorded rather than
|
|
# ignored, because `manifest.py` additionally forbids importing a
|
|
# sibling and its own test needs to see one.
|
|
if node.level:
|
|
roots.add("booth")
|
|
elif 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
|
|
|
|
|
|
def test_a_marks_file_no_one_can_parse_does_not_take_down_the_index(tmp_path):
|
|
"""The v0.2.2 round adopted the RecursionError finding and closed only half
|
|
of it. `_hydrate_safe` guards hydration; `json.loads` runs BEFORE that, in
|
|
`_read_raw`, whose `except (OSError, ValueError, UnicodeDecodeError)` does
|
|
not cover RecursionError or MemoryError.
|
|
|
|
So a 400 KB file of nothing but brackets, in any one booth, still returned
|
|
500 for `/` and `/healthz` across every booth on the service. Found by the
|
|
U5 code-review panel against the sibling module and confirmed by running it.
|
|
The read is bounded now and both classes are caught.
|
|
"""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
(booth / MARKS_FILE).write_text("[" * 200_000 + "]" * 200_000)
|
|
|
|
assert marks_for(booth) == []
|
|
|
|
|
|
def test_a_marks_file_too_large_to_be_marks_is_refused_before_it_is_read(tmp_path):
|
|
"""Bounded by `stat`, not survived. A booth holds one marks document, and
|
|
the index reads every booth's on every page load."""
|
|
from booth.marks import MARKS_MAX_BYTES
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
(booth / MARKS_FILE).write_text(" " * (MARKS_MAX_BYTES + 10))
|
|
|
|
assert marks_for(booth) == []
|
|
|
|
|
|
def test_a_write_over_an_unparseable_marks_file_still_refuses(tmp_path):
|
|
"""The strict half of the asymmetry has to see the same failures the lenient
|
|
half does, or a file that reads as "no marks" gets replaced by a write that
|
|
believed it. Same two exception classes, same bound."""
|
|
from booth.marks import MarksCorrupt, set_flag
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
(booth / MARKS_FILE).write_text("[" * 200_000 + "]" * 200_000)
|
|
|
|
with pytest.raises(MarksCorrupt):
|
|
set_flag(booth, "a.png", True)
|
|
|
|
|
|
# ---- findings from the U5 diff-scoped BUG-HUNT panel, 2026-09-22 ------------
|
|
|
|
|
|
def test_the_marks_reader_never_blocks_on_a_file_that_is_not_a_file(tmp_path):
|
|
"""Same hole the size cap opened in the manifest, in the sibling it was
|
|
copied from. `st_size` is 0 for a FIFO, so it passes the cap, and then
|
|
`read_text` blocks with no EOF. `list_booths` reads every booth's marks on
|
|
every `GET /` and `/healthz`."""
|
|
import os
|
|
import signal
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
os.mkfifo(booth / MARKS_FILE)
|
|
|
|
def _timeout(signum, frame):
|
|
raise AssertionError("marks_for blocked on a FIFO and never returned")
|
|
|
|
old = signal.signal(signal.SIGALRM, _timeout)
|
|
signal.alarm(5)
|
|
try:
|
|
assert marks_for(booth) == []
|
|
finally:
|
|
signal.alarm(0)
|
|
signal.signal(signal.SIGALRM, old)
|
|
|
|
|
|
def test_new_marks_and_imported_marks_share_one_stamp_format(tmp_path):
|
|
"""The v0.2.2 fix for the legacy-import ordering opened a NEW ordering bug,
|
|
which is the shape worth remembering. `import_legacy_asks` moved to
|
|
microsecond precision while `now_stamp` stayed at whole seconds, and `-` is
|
|
0x2D against `.` at 0x2E — so `...T10:00:00-07:00` sorts BEFORE
|
|
`...T10:00:00.500000-07:00`, putting a LATER mark ahead of an EARLIER
|
|
import inside the same second.
|
|
|
|
Deterministic order is a v1 invariant precisely because the operator refers
|
|
to things positionally. One format, or the rule cannot be stated.
|
|
"""
|
|
from booth.marks import now_stamp
|
|
|
|
stamp = now_stamp()
|
|
assert "." in stamp.split("T")[1], f"now_stamp is not sub-second: {stamp}"
|
|
assert len(stamp.split(".")[1].split("+")[0].split("-")[0]) == 6
|
|
|
|
|
|
def test_the_importer_cannot_raise_out_of_a_poisoned_entry(tmp_path):
|
|
"""`marks_for` routes every entry through `_hydrate_safe`; the importer's
|
|
return still went through the bare `_hydrate`, so the one path that reads
|
|
entries it did not write was the one without the guard."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
(booth / MARKS_FILE).write_text(json.dumps({
|
|
"version": 1,
|
|
"marks": [{"id": "n1", "shape": "note", "text": {"bad": True},
|
|
"created": "2026-09-21T00:00:00+00:00"}],
|
|
}))
|
|
(booth / f"q1{ASK_SUFFIX}").write_text(json.dumps(_single()))
|
|
|
|
from booth.marks import import_legacy_asks
|
|
out = import_legacy_asks(booth) # must not raise
|
|
assert isinstance(out, list)
|
|
|
|
|
|
def test_a_document_that_would_not_read_back_is_refused_at_the_write(tmp_path):
|
|
"""The read bound is on the STORED bytes and the write adds `indent=2`, so a
|
|
document that fits in memory can land over the limit on disk and then read
|
|
back as no marks at all — every mark in the booth gone, silently. Refuse
|
|
loudly instead: a write that fails is recoverable.
|
|
|
|
Asserted against `_write_raw` directly, because no single mark can get
|
|
there: `_clean_text` caps a note at TEXT_MAX and a flag is a fixed shape.
|
|
The reachable path is accumulation — `_note_id` puts no ceiling on how many
|
|
notes one booth may carry — which is thousands of writes, not one. Testing
|
|
it through `write_note` would need a fixture nobody could justify, and
|
|
would be testing the cap rather than the guard.
|
|
"""
|
|
from booth.marks import MARKS_MAX_BYTES, MarksCorrupt, _write_raw
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
bulk = [{"id": f"note-{i}", "shape": "note", "text": "x" * 500,
|
|
"created": "2026-09-21T00:00:00.000000+00:00"}
|
|
for i in range(MARKS_MAX_BYTES // 400)]
|
|
|
|
with pytest.raises(MarksCorrupt):
|
|
_write_raw(booth, bulk)
|
|
assert not (booth / MARKS_FILE).exists(), "a refused write still landed"
|
|
|
|
|
|
def test_a_clock_restore_that_fails_does_not_take_the_route_down(tmp_path):
|
|
"""The concrete half of the mtime-restore finding.
|
|
|
|
`_Locked.__enter__` puts the booth directory's clock back after creating its
|
|
lock, and `os.utime` can fail — a read-only directory, a booth whose owner
|
|
we are not. It used to escape into the route and answer 500 for what is
|
|
otherwise a perfectly good request. Not putting the clock back is a cost
|
|
this module can absorb; not answering is not.
|
|
|
|
The RACE half of that finding is documented in the code and deliberately not
|
|
closed: the alternative fix would silently retire the documented behaviour
|
|
that releasing a kept board resets its clock
|
|
(`test_releasing_a_board_RESETS_its_ttl_clock` pins that on purpose), which
|
|
is a TTL doctrine change rather than a bug fix.
|
|
"""
|
|
import os
|
|
|
|
from booth.marks import MARKS_LOCK, set_flag
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
real_utime = os.utime
|
|
|
|
def boom(path, *a, **kw):
|
|
if str(path) == str(booth):
|
|
raise PermissionError("read-only directory")
|
|
return real_utime(path, *a, **kw)
|
|
|
|
os.utime = boom
|
|
try:
|
|
assert set_flag(booth, "a.png", True) is not None
|
|
finally:
|
|
os.utime = real_utime
|
|
|
|
assert (booth / MARKS_LOCK).exists()
|
|
assert [m.target for m in marks_for(booth)] == ["a.png"]
|
|
|
|
|
|
# ---- the wrong-shaped answer, closed at the hydration boundary --------------
|
|
|
|
|
|
def test_a_wrong_shaped_answer_is_an_error_at_hydration_not_a_500(tmp_path):
|
|
"""A `.marks.json` that is well-formed JSON with a wrong-shaped `answer`
|
|
passed every reader and then raised in the TEMPLATE: `_hydrate` checked only
|
|
that `answer` was a dict, never that `answer["answers"]` was one, so
|
|
`marks_for` and `hold_read` both reported the mark healthy with no read
|
|
error — and `_ask_inline.html` asked a list for `.get`.
|
|
|
|
Measured at `42ea67f`, so it predates U3. U3 guarded its own surface with
|
|
`_safe_fragments` and left the gallery and marks pages alone by scope. This
|
|
closes it at the boundary the rest of the module already argues for: ONE
|
|
predicate, ONE place, every surface inherits it.
|
|
|
|
Defeating change: restoring the bare `isinstance(answer, dict)` check —
|
|
under which `error` is None here and both pages 500.
|
|
"""
|
|
declare_pick(tmp_path, "batch", {"title": "T", "questions": [
|
|
{"key": "r1", "prompt": "A?", "options": ["x", "y"]},
|
|
{"key": "r2", "prompt": "B?", "options": ["x", "y"]}]})
|
|
raw = json.loads((tmp_path / ".marks.json").read_text())
|
|
for e in raw["marks"]:
|
|
if e["id"] == "batch":
|
|
e["answer"] = {"answers": [], "notes": ""}
|
|
(tmp_path / ".marks.json").write_text(json.dumps(raw))
|
|
|
|
mark = {m.id: m for m in marks_for(tmp_path)}["batch"]
|
|
assert mark.error, "a wrong-shaped answer hydrated as healthy"
|
|
assert "answer" in mark.error
|
|
# AND the mark is not silently emptied — the declaration survives, so the
|
|
# operator can still see WHICH question broke rather than a bare error.
|
|
assert mark.declaration is not None
|
|
|
|
|
|
def test_a_healthy_multi_answer_still_hydrates(tmp_path):
|
|
"""The other direction, so the guard cannot be satisfied by rejecting
|
|
everything. Defeating change: requiring `answers` unconditionally, which
|
|
would break every single-question pick."""
|
|
declare_pick(tmp_path, "multi", {"title": "T", "questions": [
|
|
{"key": "r1", "prompt": "A?", "options": ["x", "y"]}]})
|
|
declare_pick(tmp_path, "single", {"prompt": "Which?", "options": ["x", "y"]})
|
|
raw = json.loads((tmp_path / ".marks.json").read_text())
|
|
for e in raw["marks"]:
|
|
if e["id"] == "multi":
|
|
e["answer"] = {"answers": {"r1": {"choice": "x", "notes": ""}}, "notes": ""}
|
|
if e["id"] == "single":
|
|
e["answer"] = {"choice": "x", "notes": ""}
|
|
(tmp_path / ".marks.json").write_text(json.dumps(raw))
|
|
by = {m.id: m for m in marks_for(tmp_path)}
|
|
assert by["multi"].error is None, by["multi"].error
|
|
assert by["single"].error is None, by["single"].error
|