Vuong Hoang
c015a917ee
Two four-arm artifact-only rounds landed together: the contract paraphrase (against the pre-seam-review capture) and the code-vs-contract conformance review (against the amended one), correctly firewalled from each other. The conformance round found ZERO drift in the strict sense — the code is a clause-for-clause implementation of the contract — and the weight of both rounds landed one layer down, in what green tests structurally cannot report. Full triage in persistent-memory.d/. A LIVE HOLE IN RELEASED CODE, FOUND ON THE SIBLING MODULE v0.2.2 adopted the RecursionError finding from the bug-hunt round and closed half of it: `_hydrate_safe` guards hydration, but `json.loads` runs above it in `_read_raw`, whose catch list covers neither RecursionError nor MemoryError. A 400 KB file of nothing but brackets in any ONE booth therefore still returned 500 for `/` and `/healthz` across every booth on the service. Confirmed by running it before believing it. Both modules now bound the read by `stat` before touching the bytes and catch both classes anyway, so raising a bound later cannot quietly re-open the hole. The strict half of the marks asymmetry refuses everything the lenient half tolerates, or a file that reads as "no marks" gets replaced by a write that believed it. THE WHY-WIPE `booth new x --why "..."` then `booth add x out/*.png` erased the sentence the first command existed to record. Omitted flags meant empty strings and empty strings overwrote. Two arms predicted it from the contract's wording alone; every test here passed --why on both calls and so could not see it. Omitted now means unchanged and an explicit --why "" still clears — the shell carries the distinction by leaving the variable UNSET, not empty. --title WAS WRITE-ONLY Stored, flag-surfaced, rendered nowhere. 4/4, and independently top-ranked by every arm of the paraphrase round. It lands on the booth page heading with the directory name beside it, because the directory name is the identity the operator navigates by and refers to positionally. THREE TESTS THAT COULD NOT FAIL - test_the_write_is_atomic asserted no *.tmp survived, which a plain write_text passes. It asserts the inode changes now. (The first replacement was ALSO vacuous — it spied on os.open, which Path.write_text reaches through io.open in C and never touches. Recorded in the test, because writing a second vacuous test while fixing the first is exactly the failure this round is about.) - The INV-3 preservation test passed against an implementation that regenerated `created` every time, because _now() is whole-second resolution and back-to-back writes share a stamp. Seeded from 2019 now. - test_announcing_is_activity passed whether or not _newest_mtime counted the manifest, because writing it bumps the directory mtime either way. The directory's clock is put back, leaving the file as the only thing that can keep the booth alive. ALSO - The title fallback skipped the normalizer the explicit value gets; a directory name may legally carry a newline and run to 255 bytes. - Every writer derived the same .booth.json.tmp. Marks are protected from that by their flock; the manifest has none, so uniqueness stands in. - test_stdlib_only was blind to relative imports in all four modules. - INV-1 had no guard at all; INV-5 named two different promises; the negative render states were asserted on the index only. Contract amended throughout: the 4 GB case is a stat-checked bound rather than a return constraint, every field of an error-carrying record has a stated value, INV-1 no longer contradicts INV-3, repo-wide rules are named in words instead of by a colliding number, and touches admits the macro partial the implementation added. 329 tests.
1250 lines
47 KiB
Python
1250 lines
47 KiB
Python
"""Marks: one primitive for operator judgment attached to an artifact.
|
|
|
|
`pick` (the session asks), `note` (the operator tells), `flag` (the operator
|
|
points at the good ones) — three shapes, one storage model, one read path.
|
|
|
|
See docs/contracts/u2_marks.contract.md.
|
|
"""
|
|
import ast
|
|
import json
|
|
import pathlib
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from booth.asks import ASK_SUFFIX, AskError, build_answer, normalize_ask
|
|
from booth.marks import (
|
|
MARKS_FILE,
|
|
Mark,
|
|
answer_pick,
|
|
as_dict,
|
|
declare_pick,
|
|
marks_for,
|
|
open_marks,
|
|
)
|
|
|
|
|
|
def _single():
|
|
return {"prompt": "Which render wins?", "options": ["A — baseline", "B — async"]}
|
|
|
|
|
|
def _multi():
|
|
return {
|
|
"title": "R18 batch review",
|
|
"questions": [
|
|
{"key": "q1", "prompt": "Render 1?", "options": ["keep", "drop"]},
|
|
{"key": "q2", "prompt": "Render 2?", "options": ["keep", "drop"]},
|
|
],
|
|
}
|
|
|
|
|
|
# ---- slice 1: storage, the record, the read path ----------------------------
|
|
|
|
|
|
def test_pick_round_trips_through_marks_json(tmp_path):
|
|
"""A declared pick lands in ONE file and reads back normalized."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "winner", _single())
|
|
|
|
assert (booth / MARKS_FILE).is_file()
|
|
marks = marks_for(booth)
|
|
assert len(marks) == 1
|
|
m = marks[0]
|
|
assert isinstance(m, Mark)
|
|
assert (m.id, m.shape, m.target) == ("winner", "pick", None)
|
|
assert m.answer is None
|
|
assert m.prompt == "Which render wins?"
|
|
assert m.multi is False
|
|
# normalize_ask emits `notes` (bool); the record carries it as notes_enabled (SR-2)
|
|
assert m.notes_enabled is True
|
|
assert m.notes_label == "notes"
|
|
assert [o["id"] for o in m.options] == ["A — baseline", "B — async"]
|
|
assert len(m.questions) == 1 and m.questions[0]["key"] is None
|
|
|
|
|
|
def test_marks_json_is_a_dotfile_so_it_is_not_an_item(tmp_path):
|
|
"""The dotfile skip in booth_items already covers it. Asserted, not assumed."""
|
|
from booth.items import booth_items
|
|
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
(booth / "a.png").write_bytes(b"\x89PNG")
|
|
declare_pick(booth, "winner", _single())
|
|
|
|
assert MARKS_FILE.startswith(".")
|
|
assert [i.rel for i in booth_items(booth)] == ["a.png"]
|
|
|
|
|
|
def test_corrupt_marks_json_renders_as_empty(tmp_path):
|
|
"""A booth with no marks and a booth with a broken mark file both render as
|
|
'no marks'. Neither is a 500."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
(booth / MARKS_FILE).write_text("{not json at all")
|
|
assert marks_for(booth) == []
|
|
|
|
|
|
def test_missing_booth_and_missing_file_are_empty(tmp_path):
|
|
assert marks_for(tmp_path / "nope") == []
|
|
(tmp_path / "b").mkdir()
|
|
assert marks_for(tmp_path / "b") == []
|
|
|
|
|
|
def test_broken_pick_declaration_surfaces_error(tmp_path):
|
|
"""A declaration that cannot be rendered comes back with `error` set, so the
|
|
page can SAY so rather than silently hiding a question the session believes
|
|
it posted. Preserved from list_asks."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
(booth / MARKS_FILE).write_text(json.dumps({
|
|
"version": 1,
|
|
"marks": [{"id": "bad", "shape": "pick", "target": None,
|
|
"created": "2026-09-21T10:00:00-07:00",
|
|
"declaration": {"prompt": "no options"}, "answer": None}],
|
|
}))
|
|
m = marks_for(booth)[0]
|
|
assert m.error is not None
|
|
assert m.id == "bad"
|
|
|
|
|
|
def test_declare_pick_validates_before_writing(tmp_path):
|
|
"""A session cannot land a question the renderer would refuse."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
with pytest.raises(AskError):
|
|
declare_pick(booth, "bad", {"prompt": "only one", "options": ["just me"]})
|
|
assert not (booth / MARKS_FILE).exists()
|
|
|
|
|
|
def test_declare_pick_rejects_a_bad_id(tmp_path):
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
with pytest.raises(AskError):
|
|
declare_pick(booth, "../escape", _single())
|
|
|
|
|
|
def test_redeclaring_a_pick_clears_its_answer(tmp_path):
|
|
"""The question changed, so the old judgment is not an answer to it."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "winner", _single())
|
|
answer_pick(booth, "winner", "A — baseline")
|
|
assert marks_for(booth)[0].answer is not None
|
|
|
|
declare_pick(booth, "winner", {"prompt": "Which one now?", "options": ["X", "Y"]})
|
|
marks = marks_for(booth)
|
|
assert len(marks) == 1
|
|
assert marks[0].answer is None
|
|
assert marks[0].prompt == "Which one now?"
|
|
|
|
|
|
def test_marks_are_oldest_first_by_created(tmp_path):
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "first", _single())
|
|
declare_pick(booth, "second", _single())
|
|
assert [m.id for m in marks_for(booth)] == ["first", "second"]
|
|
|
|
|
|
# ---- openness: THE one predicate (INV-2) ------------------------------------
|
|
|
|
|
|
def test_open_marks_counts_an_unanswered_pick(tmp_path):
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "winner", _single())
|
|
assert [m.id for m in open_marks(marks_for(booth))] == ["winner"]
|
|
|
|
|
|
def test_open_marks_drops_a_complete_answer(tmp_path):
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "winner", _single())
|
|
answer_pick(booth, "winner", "A — baseline")
|
|
assert open_marks(marks_for(booth)) == []
|
|
|
|
|
|
def test_a_partial_answer_counts_as_open(tmp_path):
|
|
"""SR-7, the declared behaviour change. Today the index badge tests
|
|
`answer is None` and so reports a half-answered four-question pick as
|
|
closed, while the panel renders it `◐ partial`. A booth that still owes an
|
|
answer is open — which is also what makes U4's pin rule correct."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "batch", _multi())
|
|
answer_pick(booth, "batch", {"q1": "keep"}) # q2 left alone
|
|
|
|
m = marks_for(booth)[0]
|
|
assert m.answer is not None
|
|
assert m.answer["complete"] is False
|
|
assert m.answer["unanswered"] == ["q2"]
|
|
assert [x.id for x in open_marks(marks_for(booth))] == ["batch"]
|
|
|
|
|
|
# ---- the pick semantics, preserved by NOT rewriting them (INV-4) ------------
|
|
|
|
|
|
def test_partial_answer_is_recorded_not_refused(tmp_path):
|
|
"""Operator ruling 2026-09-09, as a regression test."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "batch", _multi())
|
|
m = answer_pick(booth, "batch", {"q1": "keep", "q2": ""})
|
|
assert m.answer["answers"]["q1"]["label"] == "keep"
|
|
assert "q2" not in m.answer["answers"]
|
|
assert m.answer["unanswered"] == ["q2"]
|
|
assert m.answer["complete"] is False
|
|
|
|
|
|
def test_nothing_to_record_is_still_refused(tmp_path):
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "batch", _multi())
|
|
with pytest.raises(AskError):
|
|
answer_pick(booth, "batch", {"q1": "", "q2": ""})
|
|
|
|
|
|
def test_an_offered_but_invalid_option_is_an_error(tmp_path):
|
|
"""A broken form, not a skipped question."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "winner", _single())
|
|
with pytest.raises(AskError):
|
|
answer_pick(booth, "winner", "C — never offered")
|
|
|
|
|
|
def test_answering_an_unknown_pick_raises(tmp_path):
|
|
"""SR-4: load_ask used to own this AskError. Extracting build_answer moved
|
|
the path to the caller, and a stale form POST must be a 400, not a silent
|
|
no-op."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
with pytest.raises(AskError):
|
|
answer_pick(booth, "ghost", "A — baseline")
|
|
|
|
|
|
def test_re_answering_overwrites(tmp_path):
|
|
"""The mark is the CURRENT judgment, not a log."""
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "winner", _single())
|
|
answer_pick(booth, "winner", "A — baseline")
|
|
m = answer_pick(booth, "winner", "B — async")
|
|
assert m.answer["label"] == "B — async"
|
|
assert len(marks_for(booth)) == 1
|
|
|
|
|
|
def test_build_answer_matches_the_old_sidecar_document(tmp_path):
|
|
"""INV-4. build_answer is write_answer's logic with the I/O removed; the
|
|
document it produces is the one the sidecar carried."""
|
|
ask = normalize_ask(_single(), "winner")
|
|
doc = build_answer(ask, "B — async", notes="less banding", who="10.0.0.1")
|
|
assert doc["stem"] == "winner"
|
|
assert doc["choice"] == "B — async"
|
|
assert doc["choice_index"] == 1
|
|
assert doc["label"] == "B — async"
|
|
assert doc["notes"] == "less banding"
|
|
assert doc["complete"] is True
|
|
assert doc["unanswered"] == []
|
|
assert doc["answered_by"] == "10.0.0.1"
|
|
assert "answered_at" in doc
|
|
|
|
|
|
def test_build_answer_sources_stem_from_the_ask(tmp_path):
|
|
"""SR-3: the extraction takes a pre-loaded ask and reads `stem` off it."""
|
|
ask = normalize_ask(_multi(), "batch")
|
|
doc = build_answer(ask, {"q1": "keep", "q2": "drop"})
|
|
assert doc["stem"] == "batch"
|
|
assert doc["complete"] is True
|
|
|
|
|
|
# ---- the JSON boundary (SR-8) -----------------------------------------------
|
|
|
|
|
|
def test_as_dict_round_trips_through_json(tmp_path):
|
|
booth = tmp_path / "b"
|
|
booth.mkdir()
|
|
declare_pick(booth, "winner", _single())
|
|
m = marks_for(booth)[0]
|
|
out = json.loads(json.dumps(as_dict(m)))
|
|
assert out["id"] == "winner"
|
|
assert out["shape"] == "pick"
|
|
assert out["answer"] is None
|
|
|
|
|
|
# ---- the stdlib-only invariant (INV-5) --------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize("module", ["marks", "asks", "links", "manifest"])
|
|
def test_stdlib_only(module):
|
|
"""INV-5. scripts/booth imports these under the system python3 with NO venv,
|
|
through a `python3 -c` heredoc that no AST extractor can see — so nothing
|
|
but this test stands between a casual third-party import and `booth ask`
|
|
breaking on every fleet host."""
|
|
# `manifest` also carries a stricter copy in tests/test_manifest.py, which
|
|
# additionally forbids importing `booth.*` — a cross-import between two
|
|
# stdlib-only modules is a second way for this invariant to break.
|
|
src = pathlib.Path(__file__).parent.parent / "booth" / f"{module}.py"
|
|
tree = ast.parse(src.read_text())
|
|
roots = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
roots.update(a.name.split(".")[0] for a in node.names)
|
|
elif isinstance(node, ast.ImportFrom):
|
|
# `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)
|