vh
fa5d46443d
Code fixes: - The narrow-screen fold was specified and never built (4/4). The tray and notes are now closed <details> in the aside; above 1000px CSS alone (::details-content) shows them and hides the summary. There is no script. Browser-tested at 390 and 1400, JS on and off. - The lightbox gated on parsed board rows, not page identity (3/4). It now uses is_board, the lesson the bench panel already carried. - wants_json returned True at the first good entry, so a malformed later entry was never read (3/4). It now parses every entry first; any error is False. - One flag predicate, flagged_targets. It serves the Desk count, the tray, the filmstrip, the tape and the review button. An unreadable flag entry counts nowhere. - The header's open count and lifetime line, and the no-set marks panel, are now regions (they were stale after an in-place answer). - Inline group headers render only when every group is one contiguous run. Interleaved directories no longer reprint or misfile headers. - A booth held unreadable has no open_since, even with a readable pick beside the damage. - The swap marks an absent region is-stale instead of leaving it looking current. It carries disclosure state (except the sent form's). The failure message is readable for 0.9 s before the reload. Contract amended where the code was right and the text was not: the wants_json and record_seen signatures, landed_at's three refinements, the group position being ring-based, the end of the set offering every other open pick, the Space-key player exception, and the fold mechanism. New tests cover the parse order; a board with media; the header region; the no-set panel; interleaved groups; mixed damage; the flag predicate; the review recording .viewed; the fold at two widths with JS on and off; the status message before the reload; a lost response after a landed write (exactly one note); a stale absent region; stage node identity across a swap; and F with a radio focused. The lost-response and stale tests turn red under their mutations. 724 passed.
676 lines
32 KiB
Python
676 lines
32 KiB
Python
"""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'<figure class="item[^"]*"[^>]*>.*?</figure>', 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'<section class="desk-sec"[^>]*data-section="(\w+)"[^>]*>(.*?)</section>',
|
|
body, re.S):
|
|
out[sec] = re.findall(r'<article class="desk-row[^"]*" data-booth="([^"]+)"', inner)
|
|
return out
|
|
|
|
|
|
def test_the_desk_triages_needs_you_then_new_then_everything_else(tmp_path):
|
|
"""The tracer for C4: three sections, always in this order, each booth in
|
|
exactly one of them."""
|
|
from booth.marks import declare_pick
|
|
t0 = time.time() - 10_000 # in the PAST: a future stamp outranks every real write and hides the bug
|
|
asks = _booth(tmp_path, "asks", {"a.png": PNG})
|
|
declare_pick(asks, "q", {"prompt": "Which?", "options": ["x", "y"]})
|
|
_booth(tmp_path, "fresh", {"a.png": PNG}) # never looked at
|
|
seen = _booth(tmp_path, "seen", {"a.png": PNG})
|
|
_at(seen / "a.png", t0)
|
|
(seen / ".viewed").write_bytes(b"")
|
|
_at(seen / ".viewed", t0 + 60) # looked AFTER it landed
|
|
body = _client(tmp_path).get("/").text
|
|
assert _desk(body) == {"needs": ["asks"], "new": ["fresh"], "rest": ["seen"]}
|
|
assert body.index('data-section="needs"') < body.index('data-section="new"') \
|
|
< body.index('data-section="rest"')
|
|
|
|
|
|
def _set_created(booth: pathlib.Path, mark_id: str, created: str) -> 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}) <sub>· {who} · 2026-09-01 10:00</sub>\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".*?</section>', body, re.S).group(0)
|
|
assert "live one" in benches and "old one" not in benches
|
|
marks = re.search(r'data-panel="bookmarks".*?</section>', 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".*?</section>', 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'<article class="desk-row[^"]*" data-booth="g".*?</article>', body, re.S).group(0)
|
|
imgs = re.findall(r'<img class="([^"]*)" loading="lazy" src="/b/g/([^"]+)"', row)
|
|
assert imgs == [("", "a.png"), ("blurred-thumb", "b.png"), ("", "c.png"), ("", "d.png")]
|
|
assert "2 flagged" in row
|
|
|
|
|
|
def test_a_booth_without_images_shows_its_kind_instead(tmp_path):
|
|
_booth(tmp_path, "songs", {"a.mp3": b"ID3", "b.mp3": b"ID3"})
|
|
row = re.search(r'data-booth="songs".*?</article>', _client(tmp_path).get("/").text, re.S).group(0)
|
|
assert "♪ audio" in row and "<img" not in row
|
|
|
|
|
|
# ---- C5: the lightbox ---------------------------------------------------------
|
|
|
|
def _region(body: str, rid: str) -> str:
|
|
"""The element carrying data-region=rid, through its matching close tag
|
|
(same-name nesting counted, so a region holding spans or divs is whole)."""
|
|
m = re.search(r'<(\w+)[^>]*data-region="%s"[^>]*>' % re.escape(rid), body)
|
|
assert m, f"no region {rid}"
|
|
tag, depth, pos = m.group(1), 1, m.end()
|
|
for t in re.finditer(r"<(/?)%s\b[^>]*>" % tag, body[pos:]):
|
|
depth += -1 if t.group(1) else 1
|
|
if depth == 0:
|
|
return body[m.start():pos + t.end()]
|
|
raise AssertionError(f"region {rid} never closes")
|
|
|
|
|
|
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('<aside class="verdict"')
|
|
assert 'action="/b/g/answer"' in aside and "Which one?" in aside
|
|
assert body.count('action="/b/g/answer"') == 1, "the pick renders once, in the aside"
|
|
assert 'class="lightbox"' in body and 'id="grid"' in body
|
|
|
|
|
|
def test_the_flag_tray_lists_flags_by_tile_number_not_by_click_order(tmp_path):
|
|
"""Flag #3 first, then #1: the tray reads #1, #3. Today's panel list is in
|
|
click order `(created, id)`; the tray is the SET's order, the declared
|
|
change. Each entry is the original shown small, blurred if the item is."""
|
|
from booth.app import set_blurred
|
|
b = _booth(tmp_path, "g", {"a.png": PNG, "b.png": PNG, "c.png": PNG})
|
|
set_flag(b, "c.png", True)
|
|
time.sleep(0.01)
|
|
set_flag(b, "a.png", True)
|
|
set_blurred(b, "a.png", True)
|
|
aside = _region(_client(tmp_path).get("/b/g/").text, "verdict")
|
|
tray = re.findall(r'<a class="tray-item( is-blurred)?" href="view\?f=([^"]+)"[^>]*>.*?#(\d+)<', aside, re.S)
|
|
assert tray == [(" is-blurred", "a.png", "1"), ("", "c.png", "3")]
|
|
|
|
|
|
def test_groups_get_an_inline_header_that_is_not_a_tile(tmp_path):
|
|
"""When the rail finds grouping informative, each group's first tile is
|
|
preceded by a header — a div, never a figure.item, so the tile sequence
|
|
the keyboard and the order check walk is exactly the items."""
|
|
files = {f"{g}-{n}.png": PNG for g in ("aa", "bb", "cc") for n in (1, 2)}
|
|
_booth(tmp_path, "g", files)
|
|
body = _client(tmp_path).get("/b/g/").text
|
|
grid = body[body.index('id="grid"'):]
|
|
heads = re.findall(r'<div class="grp-head"[^>]*><span class="grp-key">(\w+)</span> <span class="grp-n">(\d+)</span>', grid)
|
|
assert heads == [("aa", "2"), ("bb", "2"), ("cc", "2")]
|
|
assert len(re.findall(r'<figure class="item', grid)) == 6
|
|
# and no header at all when grouping is not informative
|
|
_booth(tmp_path, "flat", {f"DSC{n}.jpg": PNG for n in range(4)})
|
|
assert 'class="grp-head"' not in _client(tmp_path).get("/b/flat/").text
|
|
|
|
|
|
def test_every_mark_dependent_element_is_a_swappable_region(tmp_path):
|
|
"""The in-place script replaces regions by id; a stale flag count or tile
|
|
after an in-place flag is the failure this pins."""
|
|
b = _booth(tmp_path, "g", {"a b.png": PNG, "c.png": PNG})
|
|
body = _client(tmp_path).get("/b/g/").text
|
|
for rid in ("verdict", "filters", "item-a%20b.png", "item-c.png", "status"):
|
|
assert f'data-region="{rid}"' in body, rid
|
|
# the standing board has no lightbox and therefore no verdict aside
|
|
_booth(tmp_path, "links", {"links.md": b"- [x](http://x/) <sub>\xc2\xb7 a \xc2\xb7 2026-09-01 10:00</sub>\n"})
|
|
board = _client(tmp_path).get("/b/links/").text
|
|
assert 'data-region="verdict"' not in board and 'class="lightbox"' not in board
|
|
|
|
|
|
# ---- C6: the review -----------------------------------------------------------
|
|
|
|
def test_an_audio_item_is_reviewed_like_a_picture(tmp_path):
|
|
"""The tracer for C6. A track gets the review page — its native player on
|
|
the stage, no Fit/1:1 toggle (that is for images only), and the judgment
|
|
rail with a flag that lands back here."""
|
|
_booth(tmp_path, "g", {"a.png": PNG, "b.mp3": b"ID3", "c.mp3": b"ID3"})
|
|
r = _client(tmp_path).get("/b/g/view?f=b.mp3", follow_redirects=False)
|
|
assert r.status_code == 200
|
|
body = r.text
|
|
stage = body[body.index('id="vstage"'):]
|
|
assert len(re.findall(r"<audio\b[^>]*\bcontrols\b", stage.split("</div>")[0])) == 1
|
|
assert 'id="vtoggle"' not in body and ">1:1<" not in body
|
|
rail = _region(body, "rail")
|
|
assert 'name="back" value="view"' in rail and 'name="f" value="b.mp3"' in rail
|
|
assert "2 of 3" in rail and "#2" in rail
|
|
|
|
|
|
def test_filmstrip_and_tape_are_the_review_ring_with_seen_and_flags(tmp_path):
|
|
"""One list, three surfaces: the filmstrip and the tape are the ring in
|
|
set order. The tape counts seen ∩ ring — including the item being looked
|
|
at, which is recorded before the page renders; a doc is never in it."""
|
|
b = _booth(tmp_path, "g", {"a.png": PNG, "b.md": b"# b", "c.png": PNG, "d.mp3": b"ID3"})
|
|
set_flag(b, "d.mp3", True)
|
|
c = _client(tmp_path)
|
|
c.get("/b/g/view?f=a.png")
|
|
body = c.get("/b/g/view?f=c.png").text
|
|
film = re.findall(r'<a class="film-f([^"]*)"\s+href="\?f=([^"]+)"', _region(body, "film"))
|
|
assert [(rel, cls.split()) for cls, rel in film] == [
|
|
("a.png", []), ("c.png", ["is-current"]), ("d.mp3", ["is-flagged"])]
|
|
assert re.findall(r'class="film-ord">#(\d)<', body) == ["1", "3", "4"] # whole-set numbers
|
|
tape = _region(body, "tape")
|
|
assert re.findall(r'<a class="tape-s([^"]*)"', tape) == [" is-seen", " is-current", " is-flagged"]
|
|
assert "2 of 3 seen" in tape
|
|
|
|
|
|
def test_a_question_about_this_item_is_answerable_here_and_the_rest_wait_for_the_end(tmp_path):
|
|
"""The rail offers a pick TARGETING the item; booth-level picks are a count
|
|
and a link — until the last item, where the end of the set offers them
|
|
all, each landing back on the review."""
|
|
from booth.marks import declare_pick
|
|
b = _booth(tmp_path, "g", {"a.png": PNG, "b.png": PNG})
|
|
declare_pick(b, "about-a", {"prompt": "Is a sharp?", "options": ["yes", "no"]}, target="a.png")
|
|
declare_pick(b, "overall", {"prompt": "Ship the set?", "options": ["yes", "no"]})
|
|
c = _client(tmp_path)
|
|
first = _region(c.get("/b/g/view?f=a.png").text, "rail")
|
|
assert "Is a sharp?" in first and "Ship the set?" not in first
|
|
assert "1 more open question on this booth" in first
|
|
assert first.count('name="back" value="view"') >= 2 # flag + the pick
|
|
last = _region(c.get("/b/g/view?f=b.png").text, "rail")
|
|
assert "End of the set" in last
|
|
assert "Ship the set?" in last and "Is a sharp?" in last
|
|
form = re.search(r'<form class="mark-form"[^>]*>.*?Ship the set\?|Ship the set\?.*?</form>', last, re.S)
|
|
assert form and 'name="f" value="b.png"' in last
|
|
|
|
|
|
def test_only_a_picture_gets_the_fit_toggle_and_blur_stays_honest(tmp_path):
|
|
from booth.app import set_blurred
|
|
b = _booth(tmp_path, "g", {"a.png": PNG, "v.webm": b"\x1aE"})
|
|
set_blurred(b, "a.png", True)
|
|
c = _client(tmp_path)
|
|
pic = c.get("/b/g/view?f=a.png").text
|
|
assert 'id="vtoggle"' in pic and ">Fit<" in pic and ">1:1<" in pic
|
|
assert 'class="vstage fit is-blurred"' in pic and "blur is cosmetic" in pic
|
|
vid = c.get("/b/g/view?f=v.webm").text
|
|
assert 'id="vtoggle"' not in vid and re.search(r"<video\b[^>]*\bcontrols\b", vid)
|
|
|
|
|
|
# ---- C7: copy and brand -------------------------------------------------------
|
|
|
|
def test_no_emblem_in_the_chrome(tmp_path):
|
|
"""Ruling `emblem=no`: the top bar carries the brand dot and the name, and
|
|
no image at all."""
|
|
_booth(tmp_path, "g", {"a.png": PNG})
|
|
c = _client(tmp_path)
|
|
for path in ("/", "/b/g/", "/b/g/view?f=a.png"):
|
|
body = c.get(path).text
|
|
m = re.search(r'<header class="topbar">.*?</header>', body, re.S)
|
|
if m:
|
|
assert "<img" not in m.group(0) and "<svg" not in m.group(0), path
|
|
|
|
|
|
# ---- fixups from the heid code-review panel (round "Wren") --------------------
|
|
|
|
@pytest.mark.parametrize("accept,want", [
|
|
("application/json", True),
|
|
("application/json;q=0.5", True),
|
|
("application/json;q=abc", False), # malformed q alone
|
|
("application/json, application/json;q=broken", False), # malformed AFTER a good one
|
|
("application/json;q=broken, application/json", False),
|
|
("application/json, text/plain;q=nope", False), # any unparseable entry
|
|
])
|
|
def test_wants_json_fails_closed_on_any_unparseable_entry(accept, want):
|
|
"""C3: any header that fails to parse is False, whatever order its entries
|
|
come in. The first cut returned True as soon as it met a good JSON entry,
|
|
so a malformed one after it was never read (Wren W3, 3/4)."""
|
|
from booth.app import wants_json
|
|
assert wants_json(accept) is want
|
|
|
|
|
|
def test_a_board_booth_keeps_its_single_column_even_with_media_in_it(tmp_path):
|
|
"""C5: ANYTHING with links.md is a board — page identity, not page content
|
|
(the lesson `is_board` already carries). A board with an image and a
|
|
links.md that parses to no rows must not get the lightbox (Wren W2, 3/4)."""
|
|
_booth(tmp_path, "links", {"links.md": b"just prose, no rows\n", "a.png": PNG})
|
|
body = _client(tmp_path).get("/b/links/").text
|
|
assert 'class="lightbox"' not in body and 'data-region="verdict"' not in body
|
|
|
|
|
|
def test_the_header_count_and_lifetime_line_are_a_region_too(tmp_path):
|
|
"""Answering the last open pick in place must not leave "1 open" and
|
|
"held until answered" stale in the header (Wren, hulda) — they depend on
|
|
marks, so by C3's rule they are a region."""
|
|
from booth.marks import declare_pick
|
|
b = _booth(tmp_path, "g", {"a.png": PNG})
|
|
declare_pick(b, "q", {"prompt": "?", "options": ["x", "y"]})
|
|
status = _region(_client(tmp_path).get("/b/g/").text, "booth-status")
|
|
assert "1 open" in status and "held until answered" in status
|
|
|
|
|
|
def test_a_booth_with_marks_and_no_items_keeps_its_panel_in_a_region(tmp_path):
|
|
"""No set, so no lightbox — but the panel's forms are in-place, so the
|
|
panel must be a region or a note written there shows nowhere (Wren, hulda)."""
|
|
from booth.marks import write_note
|
|
b = _booth(tmp_path, "g", {})
|
|
write_note(b, None, "only a note")
|
|
body = _client(tmp_path).get("/b/g/").text
|
|
assert "only a note" in _region(body, "marks-panel")
|
|
|
|
|
|
def test_interleaved_groups_get_no_inline_headers(tmp_path):
|
|
"""Groups come from basenames, the order from full paths, so groups can
|
|
interleave: d1/aa, d1/bb, d2/aa, d2/bb. Re-printing 'aa' and 'bb' would
|
|
claim runs that are not there, and one header per group would file d2/aa
|
|
under 'bb'. Headers render only when every group is one contiguous run
|
|
(Wren, hulda). The rail's jump links are unaffected."""
|
|
_booth(tmp_path, "g", {"d1/aa-1.png": PNG, "d1/bb-1.png": PNG,
|
|
"d2/aa-2.png": PNG, "d2/bb-2.png": PNG})
|
|
body = _client(tmp_path).get("/b/g/").text
|
|
assert 'class="grp-head"' not in body
|
|
assert 'class="rail-g"' in body
|
|
|
|
|
|
def test_a_booth_with_damaged_marks_sorts_after_every_dated_question(tmp_path):
|
|
"""Mixed damage: one readable open pick (the OLDEST stamp here) plus one
|
|
entry that cannot be read. The booth is held 'unreadable', and the
|
|
contract puts unreadable booths after every booth with a dated question —
|
|
the damage is the thing to fix, not the age of the pick beside it (Wren,
|
|
groa)."""
|
|
import json
|
|
from booth.marks import declare_pick
|
|
mixed = _booth(tmp_path, "mixed", {"a.png": PNG})
|
|
clean = _booth(tmp_path, "clean", {"a.png": PNG})
|
|
declare_pick(mixed, "q", {"prompt": "?", "options": ["x", "y"]})
|
|
declare_pick(clean, "q", {"prompt": "?", "options": ["x", "y"]})
|
|
_set_created(mixed, "q", "2026-01-01T00:00:00+00:00")
|
|
_set_created(clean, "q", "2026-09-01T00:00:00+00:00")
|
|
doc = json.loads((mixed / ".marks.json").read_text())
|
|
doc["marks"].append({"id": "bad", "shape": "note", "created": 7, "text": "x"})
|
|
(mixed / ".marks.json").write_text(json.dumps(doc))
|
|
body = _client(tmp_path).get("/").text
|
|
assert _desk(body)["needs"] == ["clean", "mixed"]
|
|
|
|
|
|
def test_one_flag_predicate_everywhere_and_a_damaged_flag_counts_nowhere(tmp_path):
|
|
"""The Desk count, the tray, the filmstrip and the review button all read
|
|
ONE predicate: a readable flag mark on the item. An unreadable flag entry
|
|
is judgment we cannot see, and must not inflate a count it cannot be shown
|
|
in (Wren W4, groa/kimi)."""
|
|
import json
|
|
b = _booth(tmp_path, "g", {"a.png": PNG, "b.png": PNG})
|
|
set_flag(b, "a.png", True)
|
|
doc = json.loads((b / ".marks.json").read_text())
|
|
doc["marks"].append({"id": "flag:b.png", "shape": "flag", "target": "b.png", "created": 9})
|
|
(b / ".marks.json").write_text(json.dumps(doc))
|
|
c = _client(tmp_path)
|
|
row = re.search(r'data-booth="g".*?</article>', c.get("/").text, re.S).group(0)
|
|
assert "1 flagged" in row
|
|
review = c.get("/b/g/view?f=b.png").text
|
|
assert "○ flag" in _region(review, "rail")
|
|
assert re.findall(r'class="film-f([^"]*)"\s+href="\?f=([^"]+)"', review)[1][0].split() == ["is-current"]
|
|
|
|
|
|
def test_a_full_size_look_also_counts_as_looking_at_the_booth(tmp_path):
|
|
"""C2: the review route calls record_view as well as record_seen, so the
|
|
Desk's "new since you looked" clears when the booth is reviewed at full
|
|
size, not only when its grid is opened (Wren, groa: untested)."""
|
|
b = _booth(tmp_path, "g", {"a.png": PNG})
|
|
assert not (b / ".viewed").exists()
|
|
_client(tmp_path).get("/b/g/view?f=a.png")
|
|
assert (b / ".viewed").exists() and (b / ".seen").exists()
|