"""R2 — the review flow: the Desk, the lightbox, the review.
Contract: docs/contracts/r2_flow.contract.md. Tests are grouped by the
contract's components (C1-C7) and named for the behaviour they pin.
"""
from __future__ import annotations
import pathlib
import re
import time
import sys
import pytest
from fastapi.testclient import TestClient
sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
from booth.app import create_app # noqa: E402
from booth.items import booth_items # noqa: E402
from booth.marks import set_flag # noqa: E402
PNG = b"\x89PNG\r\n\x1a\n"
def _booth(root: pathlib.Path, name: str, files: dict[str, bytes]) -> pathlib.Path:
b = root / name
b.mkdir()
for rel, data in files.items():
p = b / rel
p.parent.mkdir(parents=True, exist_ok=True)
p.write_bytes(data)
return b
def _client(root: pathlib.Path) -> TestClient:
return TestClient(create_app(root, ttl_hours=24, start_sweeper=False))
def _ordinals(body: str) -> dict[str, str]:
"""rel -> the ordinal text its tile prints, in render order."""
out = {}
for fig in re.findall(r']*>.*?', body, re.S):
rel = re.search(r'data-item="([^"]+)"', fig).group(1)
m = re.search(r'class="ord"[^>]*>#(\d+)<', fig)
out[rel] = m.group(1) if m else None
return out
# ---- C1: Item.ordinal ------------------------------------------------------
def test_a_filtered_tile_keeps_its_number_in_the_whole_set(tmp_path):
"""The tracer. b.png is the second item of three; under ?filter=flagged it
is the ONLY tile rendered and must still print #2, because the number is a
property of the item, not of the view."""
b = _booth(tmp_path, "g", {"a.png": PNG, "b.png": PNG, "c.png": PNG})
set_flag(b, "b.png", True)
body = _client(tmp_path).get("/b/g/?filter=flagged").text
assert _ordinals(body) == {"b.png": "2"}
def test_ordinals_count_rendered_items_only(tmp_path):
"""A caption sidecar is not an item and takes no number, so the numbers
stay contiguous over what the operator can see. And an undecodable name
that the quote() guard skips takes none either — it is not rendered."""
b = _booth(tmp_path, "g", {"a.png": PNG, "a.png.txt": b"cap", "b.png": PNG,
"c.png": PNG})
import os
os.close(os.open(bytes(b) + b"/m\xff.png", os.O_CREAT | os.O_WRONLY, 0o644))
items = booth_items(b)
assert [(it.rel, it.ordinal) for it in items] == [("a.png", 1), ("b.png", 2), ("c.png", 3)]
def test_ordinals_pad_to_the_width_of_the_whole_set(tmp_path):
"""Twelve items: numbers are written two wide, so a column of them lines
up — and a filter showing only the first does not narrow it to #1."""
b = _booth(tmp_path, "g", {f"{n:02d}.png": PNG for n in range(1, 13)})
set_flag(b, "01.png", True)
c = _client(tmp_path)
assert _ordinals(c.get("/b/g/").text)["01.png"] == "01"
assert _ordinals(c.get("/b/g/").text)["12.png"] == "12"
assert _ordinals(c.get("/b/g/?filter=flagged").text) == {"01.png": "01"}
# ---- C2: review_chain and .seen -------------------------------------------
def test_the_review_ring_is_the_item_order_filtered_to_media(tmp_path):
"""Images, video and audio, in set order. A doc is not in the ring: it
keeps its reading page."""
from booth.items import review_chain
b = _booth(tmp_path, "g", {"a.png": PNG, "b.mp3": b"ID3", "c.md": b"# c",
"d.webm": b"\x1aE", "e.zip": b"PK"})
assert review_chain(booth_items(b)) == ["a.png", "b.mp3", "d.webm"]
def test_the_review_route_rings_through_audio_in_set_order(tmp_path):
"""From the only image, "next" is the audio track that follows it in the
set — today's image-only ring had nowhere to go."""
_booth(tmp_path, "g", {"a.png": PNG, "b.mp3": b"ID3", "c.md": b"# c"})
body = _client(tmp_path).get("/b/g/view?f=a.png").text
assert 'class="vnav vnext" href="?f=b.mp3"' in body
assert 'class="vnav vprev" href="?f=b.mp3"' in body # a two-item ring wraps
def test_a_full_size_look_is_recorded_as_seen_and_a_non_item_is_not(tmp_path):
"""`.seen` answers WHICH items were looked at full size. Gated on the item
record like `record_view`: pointing `f` at a dotfile the service itself
wrote is not a look at anything."""
from booth.items import read_seen
b = _booth(tmp_path, "g", {"a.png": PNG, "b.png": PNG, "c.png": PNG})
(b / ".marks.lock").write_bytes(b"")
c = _client(tmp_path)
for f in ("a.png", "c.png", "a.png", ".marks.lock"):
c.get(f"/b/g/view?f={f}")
assert read_seen(b) == {"a.png", "c.png"}
assert (b / ".seen").read_text() == "a.png\nc.png\n" # sorted, deduplicated
def test_seen_is_pruned_to_live_items_at_the_next_write(tmp_path):
"""A deleted file drops out of `.seen` the next time anything is seen, so
the marker never outgrows the booth and never counts a ghost."""
from booth.items import read_seen
b = _booth(tmp_path, "g", {"a.png": PNG, "b.png": PNG})
c = _client(tmp_path)
c.get("/b/g/view?f=a.png")
(b / "a.png").unlink()
c.get("/b/g/view?f=b.png")
assert read_seen(b) == {"b.png"}
def test_a_planted_seen_symlink_is_replaced_not_written_through(tmp_path):
"""Any fleet session can write into a booth. A `.seen` symlink aimed at a
file outside must not turn a page view into a write at that path."""
outside = tmp_path / "victim.txt"
outside.write_text("untouched")
b = _booth(tmp_path, "g", {"a.png": PNG})
(b / ".seen").symlink_to(outside)
r = _client(tmp_path).get("/b/g/view?f=a.png")
assert r.status_code == 200
assert outside.read_text() == "untouched"
assert not (b / ".seen").is_symlink()
def test_a_look_that_cannot_be_recorded_still_serves_the_page(tmp_path):
"""NEVER RAISES: a booth the service cannot write to costs the marker, not
the page."""
b = _booth(tmp_path, "g", {"a.png": PNG})
b.chmod(0o555)
try:
r = _client(tmp_path).get("/b/g/view?f=a.png")
finally:
b.chmod(0o755)
assert r.status_code == 200
assert not (b / ".seen").exists()
# ---- C3: in-place judgment --------------------------------------------------
GOLDEN = pathlib.Path(__file__).parent / "golden" / "r2_mark_303.json"
def _seed(root: pathlib.Path) -> TestClient:
"""The golden's fixture, byte for byte (see golden_gen in the R2 notes)."""
from booth.marks import declare_pick, write_note
root.mkdir(parents=True, exist_ok=True)
b = _booth(root, "g", {"a.png": PNG, "b b.png": PNG, "c.md": PNG})
declare_pick(b, "q", {"prompt": "Which?", "options": ["x", "y"]}, target="a.png")
set_flag(b, "a.png", True)
write_note(b, "a.png", "seed")
return TestClient(create_app(root, ttl_hours=24, start_sweeper=False),
follow_redirects=False)
def test_every_pre_r2_request_shape_gets_a_byte_identical_303(tmp_path):
"""INV-4. The golden was recorded from the PRE-R2 code: every mark route,
with `back` absent and `back=marks`, under nine Accept headers that must
NOT count as asking for JSON. Status, every header, and the body must match
exactly — the no-JS guarantee lives in these bytes."""
import json
cases = json.loads(GOLDEN.read_text())
assert len(cases) == 108
for i, case in enumerate(cases):
c = _seed(tmp_path / str(i))
headers = {} if case["accept"] is None else {"accept": case["accept"]}
r = c.post(case["path"], data=case["form"], headers=headers)
got = {"status": r.status_code,
"headers": sorted([k.lower(), v] for k, v in r.headers.items()),
"body": r.content.decode("latin-1")}
want = {k: case[k] for k in ("status", "headers", "body")}
assert got == want, (case["path"], case["form"], case["accept"])
@pytest.mark.parametrize("path,form", [
("/b/g/answer", {"ask": "q", "choice": "y"}),
("/b/g/note", {"target": "a.png", "text": "in place"}),
("/b/g/flag", {"target": "b b.png", "on": "1"}),
("/b/g/unmark", {"mark": "note-1"}),
])
@pytest.mark.parametrize("accept", ["application/json", "text/html, application/json;q=0.5"])
def test_an_explicit_json_accept_gets_204_and_the_write_still_lands(tmp_path, path, form, accept):
"""The in-place path: same write as the form, no redirect, no body."""
from booth.marks import marks_for
c = _seed(tmp_path)
before = [(m.id, m.shape, m.answer, m.text) for m in marks_for(tmp_path / "g")]
r = c.post(path, data=form, headers={"accept": accept})
assert r.status_code == 204 and r.content == b""
assert "location" not in r.headers
after = [(m.id, m.shape, m.answer, m.text) for m in marks_for(tmp_path / "g")]
assert after != before, "the write must happen exactly as for the form"
@pytest.mark.parametrize("f,landing", [
("a.png", "/b/g/view?f=a.png#rail"),
("b b.png", "/b/g/view?f=b%20b.png#rail"),
("c.md", "/b/g/#item-b%20b.png"), # a doc is not in the review ring
("gone.png", "/b/g/#item-b%20b.png"), # not an item
("", "/b/g/#item-b%20b.png"),
("../../etc/passwd", "/b/g/#item-b%20b.png"),
])
def test_back_view_lands_on_the_review_only_for_a_media_item(tmp_path, f, landing):
"""The JS-off fix for the bounce: a flag set at full size lands back at
full size. Anything that is not a media item in this booth falls back to
the booth page exactly as a form with no `back` does."""
c = _seed(tmp_path)
r = c.post("/b/g/flag", data={"target": "b b.png", "on": "1", "back": "view", "f": f})
assert r.status_code == 303
assert r.headers["location"] == landing
# ---- C4: the Desk -------------------------------------------------------------
def _at(path: pathlib.Path, t: float) -> None:
import os
os.utime(path, (t, t))
def _desk(body: str) -> dict[str, list[str]]:
"""section -> the booths it renders, in render order."""
out = {}
for sec, inner in re.findall(r']*data-section="(\w+)"[^>]*>(.*?)',
body, re.S):
out[sec] = re.findall(r' None:
import json
doc = json.loads((booth / ".marks.json").read_text())
for m in doc["marks"]:
if m["id"] == mark_id:
m["created"] = created
(booth / ".marks.json").write_text(json.dumps(doc))
def test_needs_you_orders_by_the_parsed_stamp_not_the_string(tmp_path):
"""`created` is a string. As text, 11:00-07:00 sorts before 12:30-05:00;
as time it is 18:00Z against 17:30Z, so the second question is OLDER and
leads. An unparseable stamp, and a booth whose marks cannot be read, sort
after every parseable one; name breaks the tie."""
from booth.marks import declare_pick
for n in ("alpha", "bravo", "charlie", "delta"):
b = _booth(tmp_path, n, {"a.png": PNG})
if n != "delta":
declare_pick(b, "q", {"prompt": "?", "options": ["x", "y"]})
_set_created(tmp_path / "alpha", "q", "2026-09-22T11:00:00-07:00")
_set_created(tmp_path / "bravo", "q", "2026-09-22T12:30:00-05:00")
_set_created(tmp_path / "charlie", "q", "last tuesday")
(tmp_path / "delta" / ".marks.json").write_text("{not json")
body = _client(tmp_path).get("/").text
assert _desk(body)["needs"] == ["bravo", "alpha", "charlie", "delta"]
assert "marks unreadable" in body
def test_flags_and_notes_alone_do_not_make_a_booth_need_you(tmp_path):
"""Needs-you means a question TO the operator. Flags and notes are the
operator's own judgment."""
from booth.marks import write_note
b = _booth(tmp_path, "judged", {"a.png": PNG})
set_flag(b, "a.png", True)
write_note(b, None, "done here")
assert "needs" not in _desk(_client(tmp_path).get("/").text)
def test_new_since_you_looked_reads_content_not_activity(tmp_path):
"""INV-5, the two clocks. A flag made after the last look is ACTIVITY and
must not make a booth look new; a file landed after the last look is
CONTENT and must. Newest content first."""
t0 = time.time() - 10_000 # in the PAST: a future stamp outranks every real write and hides the bug
judged = _booth(tmp_path, "judged", {"a.png": PNG})
delivered = _booth(tmp_path, "delivered", {"a.png": PNG})
later = _booth(tmp_path, "later", {"a.png": PNG})
for b in (judged, delivered, later):
_at(b / "a.png", t0)
(b / ".viewed").write_bytes(b"")
_at(b / ".viewed", t0 + 10)
set_flag(judged, "a.png", True) # activity after the look
(delivered / "b.png").write_bytes(PNG)
_at(delivered / "b.png", t0 + 20) # content after the look
(later / "b.png").write_bytes(PNG)
_at(later / "b.png", t0 + 30) # ...and later still
desk = _desk(_client(tmp_path).get("/").text)
assert desk["new"] == ["later", "delivered"]
assert desk["rest"] == ["judged"]
def test_an_empty_section_renders_nothing_and_a_full_one_renders(tmp_path):
"""The negative half of the kept-lane pair, carried forward: a section with
no booths has no heading and no box. Checked against the element, never a
bare word the stylesheet also contains."""
c = _client(tmp_path)
body = c.get("/").text
for sec in ("needs", "new", "rest"):
assert f'data-section="{sec}"' not in body
assert 'data-panel="benches"' not in body and 'data-panel="bookmarks"' not in body
_booth(tmp_path, "fresh", {"a.png": PNG})
body = c.get("/").text
assert 'data-section="new"' in body
assert 'data-section="needs"' not in body and 'data-section="rest"' not in body
def test_a_look_then_a_judgment_leaves_the_booth_out_of_new(tmp_path):
"""Through the routes, not hand-set markers. Every Booth write that CREATES
a dotfile — `.viewed`, the marks file's temp-and-replace — bumps the booth
DIRECTORY's mtime. A `landed_at` that read the directory would make the
flag you set after looking read as a fresh delivery."""
t0 = time.time() - 10_000 # in the PAST: a future stamp outranks every real write and hides the bug
b = _booth(tmp_path, "g", {"a.png": PNG})
_at(b / "a.png", t0)
_at(b, t0)
c = _client(tmp_path)
assert _desk(c.get("/").text) == {"new": ["g"]}
c.get("/b/g/")
time.sleep(0.02)
# Not following the 303: following it GETs the booth page, which records a
# fresh look and would hide the defect this pins. A session writing a mark
# from the CLI never looks at the page at all.
c.post("/b/g/flag", data={"target": "a.png", "on": "1"}, follow_redirects=False)
assert _desk(c.get("/").text) == {"rest": ["g"]}
def _link(desc: str, url: str, who: str = "x-dev") -> str:
return f"- [{desc}]({url}) · {who} · 2026-09-01 10:00\n"
def test_the_side_column_shows_live_benches_and_non_booth_bookmarks(tmp_path):
"""Benches: non-retired, registry order. Bookmarks: the board the CLI
writes, booth URLs left out (a booth announces itself on the Desk), pinned
first then newest, capped at eight with the way to the rest."""
from booth.benches import set_bench_state, upsert_bench
upsert_bench(tmp_path, "http://h:1/", "live one", "a-dev")
retired, _ = upsert_bench(tmp_path, "http://h:2/", "old one", "a-dev")
set_bench_state(tmp_path, retired.id, "retired")
rows = "".join(_link(f"ref {n}", f"http://ref/{n}") for n in range(10))
rows += _link("a booth", "http://10.0.0.1:8090/b/somebooth/")
board = _booth(tmp_path, "links", {"links.md": rows.encode()})
(board / ".forever").write_bytes(b"")
body = _client(tmp_path).get("/").text
benches = re.search(r'data-panel="benches".*?', body, re.S).group(0)
assert "live one" in benches and "old one" not in benches
marks = re.search(r'data-panel="bookmarks".*?', body, re.S).group(0)
shown = re.findall(r'class="desk-mark[^"]*" href="([^"]+)"', marks)
assert shown == [f"http://ref/{n}" for n in (9, 8, 7, 6, 5, 4, 3, 2)] # newest first, 8
assert "all 10 on the board" in marks
def test_a_damaged_bench_registry_says_so_rather_than_rendering_empty(tmp_path):
(tmp_path / ".benches.json").write_text("{broken")
body = _client(tmp_path).get("/").text
panel = re.search(r'data-panel="benches".*?', body, re.S)
assert panel and "could not be read" in panel.group(0)
def test_a_row_previews_four_images_keeps_blur_and_counts_flags(tmp_path):
"""The originals shown small (no generated thumbnail), the first four in
item order, a blurred one still blurred. The flag count is on the row."""
from booth.app import set_blurred
b = _booth(tmp_path, "g", {f"{n}.png": PNG for n in "abcde"})
set_blurred(b, "b.png", True)
set_flag(b, "c.png", True)
set_flag(b, "e.png", True)
body = _client(tmp_path).get("/").text
row = re.search(r'', body, re.S).group(0)
imgs = re.findall(r'', _client(tmp_path).get("/").text, re.S).group(0)
assert "♪ audio" in row and " str:
m = re.search(r'<(\w+)[^>]*data-region="%s"[^>]*>' % re.escape(rid), body)
assert m, f"no region {rid}"
tag = m.group(1)
# regions in these templates do not nest a same-named tag inside themselves
end = body.index(f"{tag}>", m.end())
return body[m.start():end]
def test_the_verdict_sits_beside_the_set_on_a_gallery_booth(tmp_path):
"""The tracer for C5: the open question, the flags and the notes live in
one aside next to the grid — not in a panel above it that scrolls away."""
from booth.marks import declare_pick
b = _booth(tmp_path, "g", {"a.png": PNG, "b.png": PNG})
declare_pick(b, "q", {"prompt": "Which one?", "options": ["a", "b"]})
body = _client(tmp_path).get("/b/g/").text
aside = _region(body, "verdict")
assert aside.startswith('