vh
5e41108cd3
Three defects and a missing test, all surfaced by the cross-frontier contract panel dispatched before implementation and triaged after it (heid, four arms, artifact-only, thread 01M33VSNFER4N1554G0Y0VC9C8). v0.2.0 was already tagged and announced to fifteen handles when they landed, which is the argument for running the gate at all. DATA LOSS. `marks_for` is deliberately lenient — an unparseable `.marks.json` reads as "no marks" so a review page still loads. The write path inherited that leniency through the same reader, so one flag click appended a single entry to an empty list and atomically replaced the file: every mark in the booth gone, silently, from a click. Reproduced first, then fixed. The fix is an asymmetry, not a retreat from leniency. Reads stay lenient; writes go strict through `_read_raw_strict`, which distinguishes bytes-present-but- unreadable from absent and valid-but-empty, and raises `MarksCorrupt`. The damaged bytes are left on disk. Routes answer 409 rather than 500 — the service is fine and the request was well-formed, the state on disk is not — and the body says what to do, because the alternative the operator reaches for otherwise is deleting the file, which is the thing being protected. The CLI says it in one line instead of a traceback. A PICK COULD NOT TARGET AN ITEM. `Mark.target` carried one, `marks_for_target` retrieved by it, and the panel already rendered "on <item>" — but `declare_pick` had no parameter for it, so no session could produce one. A question about one artifact is the whole point of the 2026-09-09 inline-placement ruling; the door was simply missing. THE IMPORTER STRANDED AN ANSWER. A stem already present as a mark was skipped wholesale. If a session had re-declared that stem through marks while the operator's choice sat in the legacy sidecar, that choice was lost permanently — reads are forbidden from looking at sidecars. The declaration is still skipped (idempotence holds) but a legacy answer is now adopted when the existing mark is an unanswered pick, and an answer made through marks is never overwritten. INV-3 NAMED A SURFACE NOTHING TESTED. All four arms converged on it: the rule protects gallery tile, zoom view and doc view; the falsifiable check covered one. The doc view was implemented and untested, so shipping it unmarked would have passed. Three tests now, one per surface. The contract carries the full triage, including two findings accepted and NOT closed: INV-2's and INV-5's checks comply in letter — openness can be re-derived without spelling the grepped pattern, and importlib inside a function defeats the AST walk. Both describe a future careless change, and the honest statement is that these checks raise the cost of drifting rather than making it impossible. Recorded rather than papered over. Also pins the three prose ambiguities the panel found, normatively and once each: what counts as open, the three distinct broken-declaration cases, and INV-6, which had named a helper that does not exist and forbidden the calls that helper must make. 253 tests.
884 lines
31 KiB
Python
884 lines
31 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"])
|
|
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."""
|
|
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
|