"""U4 — derived lifetime.
A booth's lifetime stops being a boolean somebody remembered to press and
becomes a fact derived from the booth's own state:
KEPT `.forever` present never swept
HELD an open pick, or marks we cannot read never swept
EPHEMERAL otherwise swept past the TTL
See `docs/contracts/u4_derived_lifetime.contract.md`. The invariant ids in the
test names are that contract's.
"""
import json
import os
import pathlib
import stat
import subprocess
import time
import pytest
from fastapi.testclient import TestClient
from booth.app import (
KEEP_MARKER,
is_kept,
VIEW_MARKER,
booth_age_seconds,
create_app,
hold_reason,
is_expired,
list_booths,
record_view,
sweep_once,
zip_booth,
)
from booth.items import booth_items
from booth.marks import (answer_pick, declare_pick, delete_mark, hold_read, marks_for, open_marks,
read_error, set_flag, write_note)
def _touch(path, when=None):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(b"x")
if when is not None:
os.utime(path, (when, when))
def _stale(path, seconds=10_000):
"""Age a booth and everything in it well past any test TTL.
`follow_symlinks=False`, because a booth may hold a dangling one and
`os.utime` follows by default — which is the helper raising on a fixture the
tests deliberately build."""
t = time.time() - seconds
for p in sorted(path.rglob("*"), reverse=True):
try:
os.utime(p, (t, t), follow_symlinks=False)
except (OSError, NotImplementedError):
pass
os.utime(path, (t, t))
def _pick(booth, mark_id="q1", target=None):
"""Declare an unanswered pick — the thing that holds a booth."""
return declare_pick(booth, mark_id,
{"prompt": "which one?", "options": ["a", "b"]}, target=target)
@pytest.fixture
def client(tmp_path):
app = create_app(tmp_path, ttl_hours=24, start_sweeper=False)
return TestClient(app), tmp_path
# ---- record_view / VIEW_MARKER ---------------------------------------------
def test_record_view_writes_the_marker(tmp_path):
booth = tmp_path / "b"
_touch(booth / "a.png")
record_view(booth)
assert (booth / VIEW_MARKER).exists()
def test_a_view_resets_the_clock_through_the_existing_age_rule(tmp_path):
"""INV-1. No new arithmetic: `.viewed` is a dotfile and NOT a `.lock`
dotfile, so `_newest_mtime` already counts it. `booth_age_seconds` is
untouched by this unit."""
booth = tmp_path / "b"
_touch(booth / "a.png")
_stale(booth)
assert booth_age_seconds(booth) > 3600
record_view(booth)
assert booth_age_seconds(booth) < 60
def test_the_view_marker_is_not_special_cased_in_the_age_rule(tmp_path):
"""INV-1's ACTUAL falsifier, and the reason the test above is not enough.
Cross-frontier review, 2026-09-22: an implementation that special-cases
`.viewed` inside `_newest_mtime` — the new arithmetic INV-1 forbids —
passes a "the clock moved" assertion perfectly well. What it cannot pass is
equivalence: a booth holding `.viewed` and a booth holding any other
non-lock dotfile of the SAME mtime must report the same age, because the
rule is about the tree, not about this file.
"""
when = time.time() - 5_000
viewed = tmp_path / "viewed"
_touch(viewed / "a.png", when=when)
_touch(viewed / VIEW_MARKER, when=when)
other = tmp_path / "other"
_touch(other / "a.png", when=when)
_touch(other / ".anything-else", when=when)
for b in (viewed, other):
os.utime(b, (when, when))
now = time.time()
assert abs(booth_age_seconds(viewed, now=now)
- booth_age_seconds(other, now=now)) < 0.001
# ...and a `.lock` dotfile is the one that must NOT count, or the exemption
# the view rule rides on has stopped being an exemption.
locked = tmp_path / "locked"
_touch(locked / "a.png", when=when)
_touch(locked / ".marks.lock")
os.utime(locked, (when, when))
assert booth_age_seconds(locked, now=now) > 3600
def test_a_viewed_booth_is_not_swept(tmp_path):
booth = tmp_path / "b"
_touch(booth / "a.png")
_stale(booth)
record_view(booth)
assert sweep_once(tmp_path, ttl_seconds=3600) == []
assert booth.exists()
def test_the_view_marker_is_not_an_item(tmp_path):
"""A dotfile, so it costs nothing in counts, galleries or zips — the same
reason `.forever`, `.marks.json` and `.booth.json` cost nothing."""
booth = tmp_path / "b"
_touch(booth / "a.png")
record_view(booth)
assert [it.rel for it in booth_items(booth)] == ["a.png"]
assert list_booths(tmp_path, ttl_seconds=3600)[0]["count"] == 1
assert VIEW_MARKER not in zip_booth(booth).decode("latin-1")
def test_record_view_never_raises_on_a_read_only_booth(tmp_path):
"""INV-5. A read-only mount, a booth we do not own, a full disk: the cost
is the timestamp, never the page."""
booth = tmp_path / "b"
_touch(booth / "a.png")
os.chmod(booth, stat.S_IRUSR | stat.S_IXUSR)
try:
record_view(booth) # must not raise
assert not (booth / VIEW_MARKER).exists()
finally:
os.chmod(booth, stat.S_IRWXU)
def test_a_read_only_booth_still_serves_all_three_view_pages(client):
"""INV-5's ACTUAL falsifier — the one the unit test above does not reach.
Cross-frontier review, 2026-09-22: `record_view` swallowing OSError proves
nothing about the ROUTES. An implementation that keeps the swallow and then
touches the marker a second time outside the guard 500s every one of these
and still passes the unit test. The invariant is about the pages.
"""
c, data = client
booth = data / "b"
_touch(booth / "a.png")
os.chmod(booth, stat.S_IRUSR | stat.S_IXUSR)
try:
assert c.get("/b/b/").status_code == 200
assert c.get("/b/b/view?f=a.png").status_code == 200
assert c.get("/b/b/marks").status_code == 200
assert not (booth / VIEW_MARKER).exists(), "nothing was written"
finally:
os.chmod(booth, stat.S_IRWXU)
def test_record_view_never_raises_on_a_missing_booth(tmp_path):
record_view(tmp_path / "gone") # must not raise
# ---- which routes count as a view ------------------------------------------
def test_the_booth_page_records_a_view(client):
c, data = client
_touch(data / "b" / "a.png")
assert c.get("/b/b/").status_code == 200
assert (data / "b" / VIEW_MARKER).exists()
def test_the_zip_download_records_a_view(client):
"""Same route, same deliberate act — and the marker is written before the
early return, not after it."""
c, data = client
_touch(data / "b" / "a.png")
assert c.get("/b/b/?download=1").status_code == 200
assert (data / "b" / VIEW_MARKER).exists()
def test_the_zoom_page_records_a_view(client):
c, data = client
_touch(data / "b" / "a.png")
assert c.get("/b/b/view?f=a.png").status_code == 200
assert (data / "b" / VIEW_MARKER).exists()
def test_the_marks_page_records_a_view(client):
c, data = client
_touch(data / "b" / "a.png")
assert c.get("/b/b/marks").status_code == 200
assert (data / "b" / VIEW_MARKER).exists()
def test_the_legacy_asks_url_records_a_view_through_the_redirect(client):
"""`/asks` is a 308 to `/marks`; it needs no call of its own, and adding
one would double-count."""
c, data = client
_touch(data / "b" / "a.png")
assert c.get("/b/b/asks").status_code == 200 # followed
assert (data / "b" / VIEW_MARKER).exists()
def test_a_verbatim_booth_records_a_view(client):
"""The index.html branch returns before the gallery is built, so the
marker has to be written above that fork."""
c, data = client
(data / "b").mkdir()
(data / "b" / "index.html").write_text("hi")
assert c.get("/b/b/").status_code == 200
assert (data / "b" / VIEW_MARKER).exists()
@pytest.mark.parametrize("path", ["/", "/healthz"])
def test_browsing_the_index_is_not_a_view(client, path):
"""INV-7 / the IA's rule: a view is a DELIBERATE act, so it cannot be
triggered by browsing the index."""
c, data = client
_touch(data / "b" / "a.png")
_stale(data / "b")
before = booth_age_seconds(data / "b")
c.get(path)
assert not (data / "b" / VIEW_MARKER).exists()
assert booth_age_seconds(data / "b") >= before - 1
def test_polling_marks_json_is_not_a_view(client):
"""INV-7. An agent must not be able to hold its own booth open by polling
for the answer it is waiting on."""
c, data = client
_touch(data / "b" / "a.png")
_stale(data / "b")
before = booth_age_seconds(data / "b")
for _ in range(3):
assert c.get("/b/b/marks.json").status_code == 200
assert not (data / "b" / VIEW_MARKER).exists()
# The falsifier is the AGE, not the marker: a handler that wrote any other
# non-dot file would hold the booth open just as effectively.
assert booth_age_seconds(data / "b") >= before - 1
def test_fetching_a_file_is_not_a_view(client):
"""INV-7. Asset GETs are issued BY the page, and a hotlinked image would
otherwise keep a booth alive forever."""
c, data = client
_touch(data / "b" / "a.png")
_stale(data / "b")
before = booth_age_seconds(data / "b")
assert c.get("/b/b/a.png").status_code == 200
assert not (data / "b" / VIEW_MARKER).exists()
assert booth_age_seconds(data / "b") >= before - 1
# ---- is_held ----------------------------------------------------------------
def test_hold_reason_does_no_io_of_its_own(tmp_path):
"""INV-3. A predicate that reached for the filesystem could answer one way
for the card and another for the sweeper; one that cannot, cannot.
Discriminated rather than asserted: the marks handed in belong to a booth
that does not exist on disk. Anything that went looking would raise or
contradict itself.
"""
nowhere = tmp_path / "does-not-exist"
booth = tmp_path / "b"
booth.mkdir()
_pick(booth)
real = marks_for(booth)
assert not nowhere.exists()
assert hold_reason(real, None) == "open"
assert hold_reason([], None) is None
assert hold_reason([], "damaged") == "unreadable"
# ...and the same inputs give the same answer regardless of any booth.
assert hold_reason(real, None) == "open"
def test_hold_read_answers_both_questions_from_one_read(tmp_path):
"""INV-3's other half, and Hulda's finding on 2026-09-22.
`marks_for` then `read_error` is TWO reads, and two reads of one file are
not one read of one state: a write landing between them can yield
`([], None)` — no marks, no error — a pair that described the booth at no
instant and is exactly the pair that deletes. `hold_read` is one read.
It must also agree with `marks_for` on a clean file, or the badge beside
the hold would be counting something else.
"""
booth = tmp_path / "b"
booth.mkdir()
_pick(booth)
marks, error = hold_read(booth)
assert error is None
assert [m.id for m in marks] == [m.id for m in marks_for(booth)]
(booth / ".marks.json").write_text("not json at all")
marks, error = hold_read(booth)
assert marks == []
assert error is not None, "one read reports the damage AND the emptiness"
def test_an_open_pick_holds(tmp_path):
booth = tmp_path / "b"
booth.mkdir()
_pick(booth)
assert hold_reason(marks_for(booth), None) == "open"
def test_an_answered_pick_does_not_hold(tmp_path):
booth = tmp_path / "b"
booth.mkdir()
_pick(booth)
answer_pick(booth, "q1", "a")
assert hold_reason(marks_for(booth), None) is None
def test_a_partially_answered_pick_STILL_holds(tmp_path):
"""Operator-settled 2026-09-21, and the reason `open_marks` exists rather
than an `answer is None` test: a lifetime rule that released a booth on the
first radio click would sweep a review in flight."""
booth = tmp_path / "b"
booth.mkdir()
declare_pick(booth, "batch", {"questions": [
{"key": "one", "prompt": "a?", "options": ["x", "y"]},
{"key": "two", "prompt": "b?", "options": ["x", "y"]},
]})
answer_pick(booth, "batch", {"one": "x"})
assert hold_reason(marks_for(booth), None) == "open"
def test_a_note_does_not_hold(tmp_path):
"""A note is the operator's OUTPUT, not an owed answer."""
booth = tmp_path / "b"
booth.mkdir()
write_note(booth, None, "looks good")
assert hold_reason(marks_for(booth), None) is None
def test_a_flag_does_not_hold(tmp_path):
booth = tmp_path / "b"
_touch(booth / "a.png")
set_flag(booth, "a.png", True)
assert hold_reason(marks_for(booth), None) is None
def test_a_booth_with_no_marks_does_not_hold(tmp_path):
booth = tmp_path / "b"
_touch(booth / "a.png")
assert hold_reason(marks_for(booth), read_error(booth)) is None
def test_unreadable_marks_hold(tmp_path):
"""INV-6. `marks_for` is lenient because a review page that will not render
is worse than one missing an annotation. The same leniency on the DELETE
path would wipe the booth whose judgment we just failed to read."""
booth = tmp_path / "b"
_touch(booth / "a.png")
(booth / ".marks.json").write_text("{ this is not json")
assert marks_for(booth) == [], "the read stays lenient"
assert hold_reason(marks_for(booth), read_error(booth)) == "unreadable", "the reaper does not"
# ---- the sweeper ------------------------------------------------------------
def test_a_held_booth_survives_the_sweep(tmp_path):
"""The core of the unit: a booth the operator still owes an answer to
cannot be swept, however old."""
doomed = tmp_path / "doomed"
_touch(doomed / "a.png")
_stale(doomed)
held = tmp_path / "held"
_touch(held / "a.png")
_pick(held)
_stale(held)
wiped = sweep_once(tmp_path, ttl_seconds=3600)
assert wiped == ["doomed"]
assert held.exists()
def test_a_booth_with_unreadable_marks_survives_the_sweep(tmp_path):
"""INV-6, end to end."""
booth = tmp_path / "b"
_touch(booth / "a.png")
(booth / ".marks.json").write_text("{ truncated")
_stale(booth)
# A doomed sibling, so "spare everything" is not a passing implementation.
doomed = tmp_path / "doomed"
_touch(doomed / "a.png")
_stale(doomed)
assert sweep_once(tmp_path, ttl_seconds=3600) == ["doomed"]
assert booth.exists()
def test_answering_the_last_pick_releases_the_booth(tmp_path):
"""The hold's exit. Answering IS a write, so the booth's clock resets too —
it rejoins the sweep on a fresh clock, exactly as a released board does."""
booth = tmp_path / "b"
_touch(booth / "a.png")
_pick(booth)
_stale(booth)
assert sweep_once(tmp_path, ttl_seconds=3600) == []
answer_pick(booth, "q1", "a")
_stale(booth)
assert sweep_once(tmp_path, ttl_seconds=3600) == ["b"]
def test_held_booth_is_still_reported_expired_by_age(tmp_path):
"""INV-2. `is_expired` stays a pure age question — expiry arithmetic and
reaper policy are kept apart so they cannot drift into each other. This is
the same split `is_kept` already has."""
booth = tmp_path / "b"
_touch(booth / "a.png")
_pick(booth)
_stale(booth)
assert is_expired(booth, ttl_seconds=3600)
assert sweep_once(tmp_path, ttl_seconds=3600) == []
def test_a_held_booth_is_still_deletable(client):
"""INV-2. `sweep_once` is the ONLY caller that honours a hold, exactly as
it is the only caller that honours the keep sentinel. A hold is protection
from the timer, never from the operator."""
c, data = client
booth = data / "b"
_touch(booth / "a.png")
_pick(booth)
c.post("/b/b/delete", follow_redirects=False)
assert not booth.exists()
def test_a_held_booth_is_still_deletable_via_the_api(client):
c, data = client
booth = data / "b"
_touch(booth / "a.png")
_pick(booth)
assert c.delete("/b/b").status_code == 200
assert not booth.exists()
# ---- the index card ---------------------------------------------------------
def test_list_booths_carries_held_and_the_marks_error(tmp_path):
_touch(tmp_path / "plain" / "a.png")
held = tmp_path / "held"
_touch(held / "a.png")
_pick(held)
broken = tmp_path / "broken"
_touch(broken / "a.png")
(broken / ".marks.json").write_text("nope")
by = {b["name"]: b for b in list_booths(tmp_path, ttl_seconds=3600)}
assert by["plain"]["hold"] is None
assert by["held"]["hold"] == "open"
assert by["broken"]["hold"] == "unreadable"
def test_the_card_says_held_instead_of_a_countdown(client):
"""INV-4. A booth that is not counting down always says why. An invisible
rule that silently stopped the clock would be strictly worse than the
boolean it replaces — `.forever` was at least visible as a lane."""
c, data = client
booth = data / "held"
_touch(booth / "a.png")
_pick(booth)
html = c.get("/").text
assert "held until answered" in html
assert "expires in" not in html
def test_the_card_says_so_when_the_marks_are_unreadable(client):
"""The one hold nothing will release on its own, so it must not read as an
ordinary held booth."""
c, data = client
booth = data / "broken"
_touch(booth / "a.png")
(booth / ".marks.json").write_text("{{{")
html = c.get("/").text
assert "marks unreadable" in html
assert "expires in" not in html
def test_an_ordinary_booth_still_counts_down(client):
c, data = client
_touch(data / "plain" / "a.png")
html = c.get("/").text
assert "expires in" in html
assert "held until answered" not in html
def test_the_booth_header_says_held_too(client):
"""The index card and the booth header are two surfaces on one fact, and a
booth URL handed to the operator lands on the SECOND one."""
c, data = client
booth = data / "held"
_touch(booth / "a.png")
_pick(booth)
html = c.get("/b/held/").text
assert "held until answered" in html
assert "expires in" not in html
def test_kept_beats_held_in_the_display(client):
"""A kept booth is exempt either way, and showing two reasons for one
exemption is the two-representations-of-one-state trap."""
c, data = client
booth = data / "both"
_touch(booth / "a.png")
_touch(booth / KEEP_MARKER)
_pick(booth)
html = c.get("/").text
assert "held until answered" not in html
assert "expires in" not in html
# ---- release is activity, stated ------------------------------------------
def test_releasing_a_board_RECORDS_A_VIEW(client):
"""A released board survives another full TTL because RELEASING IT IS
SOMEBODY TOUCHING IT — a rule — and no longer because unlinking a file
happened to bump the directory's mtime, which is not one.
The behaviour is unchanged. Its reason is now stated, which is what U4 owed
the TTL doctrine it inherited.
"""
c, data = client
booth = data / "links"
_touch(booth / "a.png")
_touch(booth / KEEP_MARKER)
c.post("/b/links/unkeep", follow_redirects=False)
assert not (booth / KEEP_MARKER).exists()
assert (booth / VIEW_MARKER).exists(), "release is activity, on purpose"
def test_keeping_a_board_does_not_need_to_record_a_view(client):
"""Keep exempts it outright, so there is nothing for a clock reset to buy —
and writing the marker anyway would put a second mechanism on one job."""
c, data = client
booth = data / "b"
_touch(booth / "a.png")
c.post("/b/b/keep", follow_redirects=False)
assert (booth / KEEP_MARKER).exists()
assert not (booth / VIEW_MARKER).exists()
def test_the_marks_page_carries_the_lifetime_line(client):
"""INV-4's third home, and the one that matters most. A verbatim booth's own
index.html is served untouched, so it has no Booth-rendered header — and a
report that ASKS something is the archetype of a held booth. Without this
the booths most likely to be held would be the ones that never say so."""
c, data = client
booth = data / "report"
booth.mkdir()
(booth / "index.html").write_text("the report")
_pick(booth)
html = c.get("/b/report/marks").text
assert "held until answered" in html
@pytest.mark.parametrize("setup,expect_held", [
("open_pick", True),
("answered_pick", False),
("partial_pick", True),
("note_only", False),
("corrupt", True),
("nothing", False),
])
def test_the_card_and_the_sweeper_never_disagree(tmp_path, setup, expect_held):
"""INV-3, in its falsifiable form. Two surfaces reading one truth is how the
zoom view lost its captions; here the stakes are a deletion, so the check is
behavioural rather than by inspection — what the index reports as held and
what the sweeper refuses to take must agree on every shape of booth."""
booth = tmp_path / "b"
_touch(booth / "a.png")
if setup == "open_pick":
_pick(booth)
elif setup == "answered_pick":
_pick(booth)
answer_pick(booth, "q1", "a")
elif setup == "partial_pick":
declare_pick(booth, "batch", {"questions": [
{"key": "one", "prompt": "a?", "options": ["x", "y"]},
{"key": "two", "prompt": "b?", "options": ["x", "y"]},
]})
answer_pick(booth, "batch", {"one": "x"})
elif setup == "note_only":
write_note(booth, None, "fine")
elif setup == "corrupt":
(booth / ".marks.json").write_text("}{")
_stale(booth)
card = list_booths(tmp_path, ttl_seconds=3600)[0]
survived = sweep_once(tmp_path, ttl_seconds=3600) == []
assert (card["hold"] is not None) is expect_held
assert survived is expect_held, "the card and the reaper must not disagree"
# ---- INV-4, discriminated per surface ---------------------------------------
#
# The first cut of these tests all rendered `GET /` and checked a substring,
# which meant every one of them passed under a change that dropped the line
# from the booth header, from the marks page, or from the board branch. Four
# of four cross-frontier arms found the board branch; the rest of this block
# is the same lesson applied to the surfaces they did not have to find.
def _board(data, name="links", rows=1):
b = data / name
b.mkdir(parents=True, exist_ok=True)
(b / "links.md").write_text("".join(
f"- [row {i}](http://x{i}/) · nobody · 2026-08-23 10:00\n"
for i in range(rows)))
return b
def test_a_BOARD_booths_header_still_states_its_lifetime(client):
"""The drift all four arms found on 2026-09-22.
The booth header's sub-line forks on `{% if board %}`, and the lifetime
macro sat only in the `{% else %}` — so a booth carrying `links.md`
rendered a link count and NOTHING about its lifetime. INV-4 says the
templates have no path that renders neither, and that was a path.
Reachable, not theoretical: the release button on the kept card drops the
sentinel, and a hand-made `links.md` booth never had one. The standing
board being kept by construction is what hid it.
"""
c, data = client
_board(data, rows=2)
html = c.get("/b/links/").text
assert "2 links" in html
assert "expires in" in html, "a non-kept board must say when it goes"
def test_a_held_BOARD_booth_says_held_in_its_header(client):
c, data = client
b = _board(data, rows=1)
_pick(b)
html = c.get("/b/links/").text
assert "held until answered" in html
assert "expires in" not in html
def test_kept_beats_held_in_the_HEADER_not_just_the_index(client):
"""The first version of this test only rendered `/`, where kept cards took
a hardcoded `kept` string and never reached the macro — so swapping the
macro's precedence to held-before-kept passed it. This renders the surface
that actually calls the macro with `kept=True`."""
c, data = client
booth = data / "both"
_touch(booth / "a.png")
_touch(booth / KEEP_MARKER)
_pick(booth)
html = c.get("/b/both/").text
assert "kept" in html
assert "held until answered" not in html
assert "expires in" not in html
def test_the_booth_header_alone_carries_the_hold(client):
"""Rendered without touching the index, so dropping the header's line
cannot be masked by the index card still having one."""
c, data = client
booth = data / "held"
_touch(booth / "a.png")
_pick(booth)
assert "held until answered" in c.get("/b/held/").text
def test_a_kept_board_still_surfaces_unreadable_marks(client):
"""A kept booth is exempt either way, so the hold is moot — but damaged
judgment is not an exemption, it is a thing somebody has to fix, and the
durable boards are where losing the operator's marks costs most. The kept
card used to render a hardcoded `kept` and say nothing.
Cross-frontier paraphrase panel, 2026-09-22: `held` and `marks_error` are
raw facts on the card; only the DISPLAY has a precedence."""
c, data = client
booth = data / "board"
_touch(booth / "a.png")
_touch(booth / KEEP_MARKER)
(booth / ".marks.json").write_text("<<>>")
index = c.get("/").text
assert "marks unreadable" in index
assert "kept" in index
assert "marks unreadable" in c.get("/b/board/").text
def test_a_404_zoom_url_is_not_a_view(client):
"""`record_view` sits BELOW the 404s in the zoom route, so a crawler
walking dead zoom URLs cannot hold a booth open — the same reasoning that
keeps `marks.json` off the view list. Regin and Kimi both raised it."""
c, data = client
booth = data / "b"
_touch(booth / "a.png")
assert c.get("/b/b/view?f=no-such-file.png").status_code == 404
assert not (booth / VIEW_MARKER).exists()
# ---- INV-2, through the CLI the sessions actually call ----------------------
def test_booth_rm_deletes_a_held_booth(tmp_path):
"""INV-2 via `scripts/booth`, not just the HTTP routes. A hold is
protection from the TIMER; nothing about it may reach a deliberate delete,
and `booth rm` is the one 17 handles type."""
script = pathlib.Path(__file__).parent.parent / "scripts" / "booth"
booth = tmp_path / "b"
_touch(booth / "a.png")
_pick(booth)
assert hold_reason(*hold_read(booth)) == "open"
r = subprocess.run([str(script), "rm", "b"], capture_output=True, text=True,
env={**os.environ, "BOOTH_DATA_DIR": str(tmp_path),
"BOOTH_URL": "http://booth.invalid"}, timeout=30)
assert r.returncode == 0, r.stderr
assert not booth.exists()
# ---- what the bug-hunt panel found (2026-09-22) ----------------------------
#
# Four independent paths into one shape: a read that FAILED still resolving to
# "no hold", and therefore to a delete. No single arm found all four. The
# invariant declared to the panel was "a deletion decision must never be made
# from a read that failed"; these are the tests that make it true.
def test_a_pick_that_hydrated_BROKEN_still_holds(tmp_path):
"""The round's strongest finding (Hulda, solo).
`.marks.json` parses, but one mark fails normalization, so `_hydrate_safe`
hands back a Mark carrying `error`. `_is_open` returns False for an errored
pick ON PURPOSE — it can never be answered, which is what the CLI's exit
code 4 means — so the booth read as `not held` and SWEPT, while the panel
beside it rendered the broken mark in full. The fail-safe was built for
file-level damage and missed entry-level.
A mark we cannot read is judgment we cannot see. Deleting the booth it
belongs to is the one thing we must not do with it.
"""
booth = tmp_path / "b"
_touch(booth / "a.png")
(booth / ".marks.json").write_text(json.dumps({
"version": 1,
"marks": [{"id": "q1", "shape": "pick", "created": "2026-09-22T00:00:00.000000+00:00",
"doc": "this should be a dict, not a string"}],
}))
marks = marks_for(booth)
assert len(marks) == 1 and marks[0].error is not None, "the panel still renders it"
assert open_marks(marks) == [], "and it is NOT open — a broken pick cannot be answered"
assert hold_reason(*hold_read(booth)) == "unreadable", "but it still holds"
_stale(booth)
doomed = tmp_path / "doomed"
_touch(doomed / "a.png")
_stale(doomed)
assert sweep_once(tmp_path, ttl_seconds=3600) == ["doomed"]
assert booth.exists()
def test_a_present_but_blank_marks_file_holds(tmp_path):
"""Gróa, solo. `_read_raw_strict` early-returns for whitespace-only content
— "absent, empty and valid-but-empty are all no marks yet", which is right
for the WRITE path it was written for. On the delete path it is not: our
writer never produces a blank marks document, so a blank one that exists
is something that went wrong, and rmtree is not the response to that."""
booth = tmp_path / "b"
_touch(booth / "a.png")
(booth / ".marks.json").write_text(" \n\n ")
_stale(booth)
assert sweep_once(tmp_path, ttl_seconds=3600) == []
assert booth.exists()
def test_a_booth_whose_own_stat_fails_reads_as_FRESH_not_ancient(tmp_path):
"""Hulda, solo. `_newest_mtime` returned 0.0 when the booth's own stat
failed, which made it maximally ancient and therefore the FIRST thing the
sweeper takes — a permissions problem resolving to a deletion. Not knowing
a booth's age is a reason to leave it alone."""
gone = tmp_path / "vanished"
assert booth_age_seconds(gone) < 60, "unknowable age is FRESH, not epoch-old"
def test_a_dangling_symlink_is_still_skipped_not_treated_as_unknowable(tmp_path):
"""The other half of that fix, and the reason it is not a blanket catch: a
dangling symlink and a file removed mid-scan both raise FileNotFoundError,
and neither is a thing with an mtime worth counting. Only OTHER stat
failures mean 'there is something here we cannot read'."""
booth = tmp_path / "b"
_touch(booth / "a.png")
_stale(booth)
(booth / "dangling").symlink_to(tmp_path / "nowhere")
os.utime(booth, (time.time() - 10_000,) * 2)
assert booth_age_seconds(booth) > 3600, "a dangling link does not make a booth fresh"
def test_is_kept_treats_an_unreadable_sentinel_as_KEPT(tmp_path):
"""Gróa, solo. `Path.exists()` maps ELOOP and EACCES to False, so a kept
booth whose sentinel could not be stat'd became sweepable. A symlink
sentinel counts too, dangling or not — somebody put it there to mean keep."""
booth = tmp_path / "b"
_touch(booth / "a.png")
(booth / KEEP_MARKER).symlink_to(tmp_path / "does-not-exist")
_stale(booth)
assert is_kept(booth) is True
assert sweep_once(tmp_path, ttl_seconds=3600) == []
def test_record_view_refuses_to_follow_a_planted_symlink(tmp_path):
"""Three of four arms, independently. `Path.touch()` follows an existing
symlink, so a booth carrying `.viewed -> /anywhere` turned every page view
into an mtime write at an arbitrary path under the service uid — and any
fleet session can write into a booth, because making a folder is the whole
API. A symlink here now raises ELOOP into the swallow: view-recording
quietly stops for that booth, which is the right way to lose this."""
outside = tmp_path / "outside.txt"
outside.write_text("do not touch me")
old = time.time() - 50_000
os.utime(outside, (old, old))
booth = tmp_path / "b"
_touch(booth / "a.png")
(booth / VIEW_MARKER).symlink_to(outside)
record_view(booth) # must not raise, must not follow
assert abs(outside.stat().st_mtime - old) < 1, "the target's mtime is untouched"
def test_a_lock_file_is_not_a_viewable_item(tmp_path):
"""The panel pointed `?f=.marks.lock` at the zoom route and held a booth
open with a file the service created itself. A dotfile is not an item, and
a view of a thing that is not an item is not a view of the booth."""
app = create_app(tmp_path, ttl_hours=24, start_sweeper=False)
c = TestClient(app)
booth = tmp_path / "b"
_touch(booth / "a.png")
_touch(booth / ".marks.lock")
c.get("/b/b/view?f=.marks.lock")
assert not (booth / VIEW_MARKER).exists()
def test_releasing_an_ALREADY_released_booth_is_not_activity(client):
"""An unconditional `record_view` on unkeep made POSTing release at an
already-released booth an endless TTL refresh — contradicting this route's
own no-op promise and diverging from the CLI, which removes the sentinel
without recording anything. Release is activity; releasing nothing is not."""
c, data = client
booth = data / "b"
_touch(booth / "a.png")
assert not (booth / KEEP_MARKER).exists()
r = c.post("/b/b/unkeep", follow_redirects=False)
assert r.status_code == 303, "still a no-op, still not a 500"
assert not (booth / VIEW_MARKER).exists()
def test_a_forever_that_is_a_DIRECTORY_does_not_500_the_release(client):
"""Pre-existing, re-exposed: `unlink` on a directory raises
IsADirectoryError straight through the route, which made the card's release
button permanently dead for that booth."""
c, data = client
booth = data / "b"
_touch(booth / "a.png")
(booth / KEEP_MARKER).mkdir()
r = c.post("/b/b/unkeep", follow_redirects=False)
assert r.status_code == 303
def test_a_VALID_but_empty_marks_document_does_NOT_hold(tmp_path):
"""The other side of the blank-file fix, and the regression it would have
been without this. `_write_raw` writes `{"version": 1, "marks": []}` when
the last mark is deleted — a valid document with nothing in it. If THAT
held, every booth the operator ever finished judging would be immortal.
Blank bytes are damage; an empty list is an answer.
"""
booth = tmp_path / "b"
_touch(booth / "a.png")
_pick(booth)
delete_mark(booth, "q1")
assert (booth / ".marks.json").read_text().strip(), "the file is still there"
_stale(booth)
assert hold_reason(*hold_read(booth)) is None
assert sweep_once(tmp_path, ttl_seconds=3600) == ["b"]