vh
64f64889a2
The operator, on the live Desk: "how is this last activity first?" It was not, usefully. The section sorted by `_newest_mtime`, which counts a look (`.viewed`), so opening a booth moved it up. Tonight two post-deploy checks fetched every booth page within half a second, which recorded 22 looks at once and collapsed the section into reverse name order through the (mtime, name) tie-break. Meanwhile each row shows "updated X ago", which is `landed_at`, a different clock from the one the list was sorted by. Operator ruling: "last activity can just be last time the booth was updated, not necessarily operator's last activity." The section now sorts by `(-landed_at, name)`, the date the row shows, labelled "last updated first". Looking, flagging and blurring no longer move a booth. `list_booths` keeps its own order for its other readers, and `_newest_mtime` still feeds lifetime. The r2_flow contract (§3, the ordering table, INV-5) and ROADMAP's ordering row are amended to match. Two tests and two r2_flow.toml rows cover it (25/25).
1194 lines
59 KiB
Python
1194 lines
59 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", "c.png"]' # sorted, deduplicated, JSON
|
|
|
|
|
|
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_everything_else_is_ordered_by_last_update_not_by_looking(tmp_path):
|
|
"""Operator, 2026-09-23: "last activity can just be last time the booth was
|
|
updated". The row shows "updated X ago" (`landed_at`), but the section
|
|
sorted by `_newest_mtime`, which counts a look, so opening a booth moved it
|
|
up, and a script that fetched every booth reshuffled the whole section into
|
|
reverse name order. Defeating change: `rest` left in `list_booths` order."""
|
|
t0 = time.time() - 100_000
|
|
for name, landed in (("aaa-old", t0), ("mmm-mid", t0 + 250), ("zzz-new", t0 + 500)):
|
|
b = _booth(tmp_path, name, {"a.png": PNG})
|
|
_at(b / "a.png", landed)
|
|
(b / ".viewed").write_bytes(b"")
|
|
_at(b / ".viewed", t0 + 1000) # every one looked at since it landed
|
|
_at(b, t0)
|
|
_at(tmp_path / "aaa-old" / ".viewed", t0 + 5000) # ...and the OLDEST looked at last
|
|
body = _client(tmp_path).get("/").text
|
|
assert _desk(body)["rest"] == ["zzz-new", "mmm-mid", "aaa-old"]
|
|
assert "last updated first" in body
|
|
|
|
|
|
def test_everything_else_breaks_an_update_tie_by_name(tmp_path):
|
|
"""CLAUDE.md invariant 6: two booths landed by one rsync share an mtime."""
|
|
t0 = time.time() - 100_000
|
|
for name in ("bravo", "alpha", "charlie"):
|
|
b = _booth(tmp_path, name, {"a.png": PNG})
|
|
_at(b / "a.png", t0)
|
|
(b / ".viewed").write_bytes(b"")
|
|
_at(b / ".viewed", t0 + 10)
|
|
assert _desk(_client(tmp_path).get("/").text)["rest"] == ["alpha", "bravo", "charlie"]
|
|
|
|
|
|
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 first four in item order, a blurred one still blurred, the flag
|
|
count on the row.
|
|
|
|
⚠ THE SOURCE CHANGED AFTER R2 AND THE ASSERTION MOVED WITH IT. This
|
|
originally read "the originals shown small (no generated thumbnail)", which
|
|
was true when written and is what the operator rejected: the strip is four
|
|
images per booth on the page he opens first, and on the live set that was
|
|
the heaviest surface in the service. The strip now asks for `?thumb=1`.
|
|
|
|
The URL carries `?thumb=1` from the extension alone, without a disk read —
|
|
so a tiny stub like this fixture still gets the parameter, and the route
|
|
simply serves the original when there is nothing worth generating."""
|
|
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?thumb=1"), ("blurred-thumb", "b.png?thumb=1"),
|
|
("", "c.png?thumb=1"), ("", "d.png?thumb=1")]
|
|
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()
|
|
|
|
|
|
def test_a_flag_on_a_file_that_is_gone_stays_visible_and_withdrawable(tmp_path):
|
|
"""Nyx N3 (2/4): the tray only shows live items, and `tray` being always
|
|
defined killed the old list fallback — so a flag whose file was deleted
|
|
rendered NOWHERE on the booth page while the Desk still counted it. It is
|
|
now listed apart, with its withdraw control; the Desk counts live items."""
|
|
b = _booth(tmp_path, "g", {"a.png": PNG, "b.png": PNG})
|
|
set_flag(b, "a.png", True)
|
|
set_flag(b, "b.png", True)
|
|
(b / "b.png").unlink()
|
|
c = _client(tmp_path)
|
|
aside = _region(c.get("/b/g/").text, "verdict")
|
|
orphans = re.search(r'class="orphan-flags".*?</ul>', aside, re.S)
|
|
assert orphans and "b.png" in orphans.group(0)
|
|
assert 'action="/b/g/unmark"' in orphans.group(0)
|
|
row = re.search(r'data-booth="g".*?</article>', c.get("/").text, re.S).group(0)
|
|
assert "1 flagged" in row
|
|
|
|
|
|
def test_a_planted_fifo_or_device_seen_marker_cannot_hang_the_review(tmp_path):
|
|
"""Nyx N4 (3/4): `.seen` was read with an unbounded, symlink-following
|
|
read_text(). A FIFO with no writer blocked the worker forever; a symlink to
|
|
/dev/zero read until memory ran out. The read now refuses anything that is
|
|
not a small regular file, without following a link."""
|
|
import os
|
|
import threading
|
|
b = _booth(tmp_path, "g", {"a.png": PNG})
|
|
os.mkfifo(b / ".seen")
|
|
out = {}
|
|
t = threading.Thread(target=lambda: out.setdefault(
|
|
"r", _client(tmp_path).get("/b/g/view?f=a.png")), daemon=True)
|
|
t.start()
|
|
t.join(timeout=5)
|
|
assert "r" in out, "the review hung on a FIFO .seen"
|
|
assert out["r"].status_code == 200
|
|
h = _booth(tmp_path, "h", {"a.png": PNG})
|
|
(h / ".seen").symlink_to("/dev/zero")
|
|
assert _client(tmp_path).get("/b/h/view?f=a.png").status_code == 200
|
|
|
|
|
|
def test_seen_round_trips_names_with_spaces_and_newlines(tmp_path):
|
|
"""Nyx (groa, hulda): one stripped line per rel lost ` a.png` and split a
|
|
name holding a newline into two identities. The marker is a JSON array."""
|
|
from booth.items import read_seen
|
|
b = _booth(tmp_path, "g", {"a.png": PNG, " a.png": PNG, "x\ny.png": PNG})
|
|
c = _client(tmp_path)
|
|
c.get("/b/g/view", params={"f": " a.png"})
|
|
c.get("/b/g/view", params={"f": "x\ny.png"})
|
|
assert read_seen(b) == {" a.png", "x\ny.png"}
|
|
|
|
|
|
|
|
def test_a_deeply_nested_seen_marker_reads_as_nothing_seen(tmp_path):
|
|
"""A JSON array nested past the parser's recursion limit raises
|
|
RecursionError, which is not a ValueError: a 100 KB file of `[` planted as
|
|
`.seen` escaped the never-raises read and 500'd every review of the booth.
|
|
It reads as nothing seen, and the next look rewrites it."""
|
|
from booth.items import read_seen
|
|
b = _booth(tmp_path, "g", {"a.png": PNG})
|
|
(b / ".seen").write_text("[" * 100_000)
|
|
assert read_seen(b) == set()
|
|
assert _client(tmp_path).get("/b/g/view?f=a.png").status_code == 200
|
|
assert read_seen(b) == {"a.png"}
|
|
|
|
def test_the_content_clock_reads_the_booth_not_what_its_links_point_at(tmp_path):
|
|
"""Nyx (groa, regin): stat() followed a symlink, so a link to a busy file
|
|
outside the booth made the booth read as newly delivered on every load; and
|
|
one unreadable entry (a symlink loop) made the whole booth read as landed
|
|
NOW, forever. The link's own mtime counts; an unreadable entry is skipped."""
|
|
import os
|
|
t0 = time.time() - 10_000 # in the PAST: a future stamp outranks every real write and hides the bug
|
|
outside = tmp_path / "busy.log"
|
|
outside.write_text("x")
|
|
b = _booth(tmp_path, "g", {"a.png": PNG})
|
|
(b / "linked.png").symlink_to(outside)
|
|
(b / "loop.png").symlink_to(b / "loop.png")
|
|
for p in (b / "a.png", b / "linked.png", b / "loop.png"):
|
|
os.utime(p, (t0, t0), follow_symlinks=False)
|
|
_at(b, t0)
|
|
c = _client(tmp_path)
|
|
c.get("/b/g/") # look at it
|
|
os.utime(outside, None) # the outside file keeps moving
|
|
assert _desk(c.get("/").text).get("rest") == ["g"]
|
|
|
|
|
|
|
|
def test_one_unreadable_entry_costs_that_entry_not_the_booth(tmp_path):
|
|
"""Nyx (groa): one entry the walk can list but not stat made the whole
|
|
booth read as landed NOW on every load, pinned in 'new' forever. A
|
|
directory readable but not searchable is that entry: its names list, and
|
|
every lstat under it is EACCES. (The symlink loop that first showed this
|
|
stopped being a fixture for it once the clock moved to lstat, which reads
|
|
a loop without following it.)"""
|
|
import os
|
|
t0 = time.time() - 10_000
|
|
b = _booth(tmp_path, "g", {"a.png": PNG})
|
|
sub = b / "d"
|
|
sub.mkdir()
|
|
(sub / "x.png").write_bytes(PNG)
|
|
for p in (b / "a.png", sub / "x.png", sub):
|
|
_at(p, t0)
|
|
_at(b, t0)
|
|
c = _client(tmp_path)
|
|
c.get("/b/g/") # look at it
|
|
sub.chmod(0o644) # r--: listable, nothing inside stat-able
|
|
try:
|
|
with pytest.raises(PermissionError):
|
|
(sub / "x.png").lstat() # the fixture is live, not assumed
|
|
assert _desk(c.get("/").text).get("rest") == ["g"]
|
|
finally:
|
|
sub.chmod(0o755)
|
|
|
|
def test_a_nul_in_the_review_path_is_a_404_not_a_500(tmp_path):
|
|
"""Nyx (groa, seat-probed): Path raises ValueError on an embedded NUL, and
|
|
the route caught only OSError. Every other hostile `f` is a 404."""
|
|
_booth(tmp_path, "g", {"a.png": PNG})
|
|
assert _client(tmp_path).get("/b/g/view?f=a%00.png").status_code == 404
|
|
|
|
|
|
@pytest.mark.parametrize("q", ["inf", "1e999", "nan", "-inf"])
|
|
def test_a_non_finite_q_is_malformed(q):
|
|
"""Nyx (regin): float() parses inf and 1e999, and inf > 0 — a malformed
|
|
header slipped through to the 204. Non-finite q is malformed: False."""
|
|
from booth.app import wants_json
|
|
assert wants_json(f"application/json;q={q}") is False
|
|
|
|
|
|
def test_a_sound_only_booth_can_open_the_review(tmp_path):
|
|
"""Nyx (groa): only the image tile linked to view?f=, so a booth of tracks
|
|
had no way into the review, the tape or `.seen`. Every media tile links in
|
|
(and Enter on the grid cursor follows that link)."""
|
|
_booth(tmp_path, "g", {"a.mp3": b"ID3", "b.webm": b"\x1aE"})
|
|
body = _client(tmp_path).get("/b/g/").text
|
|
for rel in ("a.mp3", "b.webm"):
|
|
fig = re.search(r'<figure[^>]*data-item="%s".*?</figure>' % re.escape(rel), body, re.S).group(0)
|
|
assert f'href="view?f={rel}"' in fig, rel
|
|
|
|
|
|
def test_the_desk_never_makes_a_non_web_url_clickable(tmp_path):
|
|
"""Nyx (kimi): bookmark and bench URLs are agent-written and land in href.
|
|
Autoescape does nothing about a `javascript:` scheme. The Desk links only
|
|
http(s) and shows anything else as plain text."""
|
|
import json
|
|
rows = (_link("evil", "javascript:alert`1`")
|
|
+ _link("fine", "https://example.test/"))
|
|
board = _booth(tmp_path, "links", {"links.md": rows.encode()})
|
|
(board / ".forever").write_bytes(b"")
|
|
# The bench WRITE path refuses a non-web URL; a hand-edited registry does
|
|
# not pass through it, and the reader takes any text.
|
|
(tmp_path / ".benches.json").write_text(json.dumps({
|
|
"evil": {"url": "javascript:alert(1)", "name": "evil bench"},
|
|
"http://h:1/": {"url": "http://h:1/", "name": "fine bench"}}))
|
|
body = _client(tmp_path).get("/").text
|
|
panel = re.search(r'data-panel="bookmarks".*?</section>', body, re.S).group(0)
|
|
assert 'href="javascript:' not in panel
|
|
assert 'href="https://example.test/"' in panel and "evil" in panel
|
|
benches = re.search(r'data-panel="benches".*?</section>', body, re.S).group(0)
|
|
assert 'href="javascript:' not in benches
|
|
assert 'href="http://h:1/"' in benches and "evil bench" in benches
|
|
|
|
|
|
def _regions(body: str) -> list[str]:
|
|
"""Every data-region element's full markup."""
|
|
return [_region(body, rid) for rid in dict.fromkeys(re.findall(r'data-region="([^"]+)"', body))]
|
|
|
|
|
|
def test_the_booth_blur_toggle_works_without_js_and_lands_back_on_the_review(tmp_path):
|
|
"""r2b D2b: the operator's control for booth-dev's whole-booth marker. A
|
|
plain form — scripts off, it still works — whose label says what IS
|
|
(read from the server), in the booth header and the review's top bar; the
|
|
review's carries `back` and lands on the same item. The Desk row says
|
|
`blurred` so a fogged strip says why."""
|
|
b = _booth(tmp_path, "g", {"a.png": PNG, "b.png": PNG})
|
|
c = _client(tmp_path)
|
|
page = c.get("/b/g/").text
|
|
form = re.search(r'<form[^>]*action="/b/g/blurbooth".*?</form>', page, re.S).group(0)
|
|
assert 'method="post"' in form, "a GET form would not change anything with scripts off"
|
|
assert 'name="on" value="1"' in form and "blur booth" in form
|
|
# a REGION: its label is server state, so an in-place save refreshes it
|
|
assert form in _region(page, "blur-booth")
|
|
assert "badge-blur" not in re.search(r'data-booth="g".*?</article>', c.get("/").text, re.S).group(0)
|
|
|
|
r = c.post("/b/g/blurbooth", data={"on": "1"}, follow_redirects=False)
|
|
assert r.status_code == 303 and (b / ".blurbooth").exists()
|
|
page = c.get("/b/g/").text
|
|
form = re.search(r'<form[^>]*action="/b/g/blurbooth".*?</form>', page, re.S).group(0)
|
|
assert 'name="on" value="0"' in form and "booth blurred" in form
|
|
row = re.search(r'data-booth="g".*?</article>', c.get("/").text, re.S).group(0)
|
|
assert re.search(r'<span class="badge[^"]*badge-blur[^"]*"[^>]*>[^<]*blurred</span>', row)
|
|
|
|
view = c.get("/b/g/view?f=b.png").text
|
|
vbar = re.search(r'<div class="vbar">.*?</div>\s*\n', view, re.S).group(0)
|
|
vform = re.search(r'<form[^>]*action="/b/g/blurbooth".*?</form>', vbar, re.S).group(0)
|
|
assert 'method="post"' in vform
|
|
assert 'name="back" value="b.png"' in vform and 'name="on" value="0"' in vform
|
|
r = c.post("/b/g/blurbooth", data={"on": "0", "back": "b.png"}, follow_redirects=False)
|
|
assert r.headers["location"] == "/b/g/view?f=b.png" and not (b / ".blurbooth").exists()
|
|
|
|
|
|
def test_reveal_all_is_in_the_markup_only_when_something_is_blurred_and_always_hidden(tmp_path):
|
|
"""r2b D2: the control is server markup only when the booth has a blurred
|
|
item, always with `hidden` (the script unhides it; with JS off it never
|
|
shows), and always OUTSIDE every data-region so no in-place swap replaces
|
|
it. Every page rendered for one booth carries `data-booth` on <html>; the
|
|
index carries none, so nothing there can be revealed."""
|
|
from booth.app import set_blurred
|
|
b = _booth(tmp_path, "g", {"a.png": PNG, "b.png": PNG, "n.md": b"# n"})
|
|
c = _client(tmp_path)
|
|
assert not re.search(r"<button[^>]*data-reveal-all", c.get("/b/g/").text)
|
|
set_blurred(b, "a.png", True)
|
|
for url in ("/b/g/", "/b/g/view?f=a.png", "/b/g/view?f=b.png"):
|
|
page = c.get(url).text
|
|
ctl = re.findall(r'<button[^>]*data-reveal-all[^>]*>', page)
|
|
assert len(ctl) == 1 and " hidden" in ctl[0], (url, ctl)
|
|
assert all(not re.search(r"<button[^>]*data-reveal-all", r) for r in _regions(page)), url
|
|
for url in ("/b/g/", "/b/g/view?f=a.png", "/b/g/view?f=n.md", "/b/g/marks"):
|
|
assert re.search(r'<html lang="en" data-booth="g">', c.get(url).text), url
|
|
assert re.search(r'<html lang="en">', c.get("/").text)
|
|
|
|
|
|
def test_under_a_fogged_booth_each_items_blur_control_tells_the_truth(tmp_path):
|
|
"""r2b D2b, found rendering it: in a fogged booth every item reports
|
|
`blurred`, so an item blurred ONLY by the booth offered "◉ blurred" and an
|
|
un-blur that visibly did nothing (the booth still fogged it). Its control
|
|
now says it is blurred with the booth and offers no per-item action; an item
|
|
blurred in its own right keeps its own un-blur."""
|
|
from booth.app import set_blurred
|
|
b = _booth(tmp_path, "g", {"a.png": PNG, "b.png": PNG})
|
|
set_blurred(b, "a.png", True)
|
|
(b / ".blurbooth").write_bytes(b"")
|
|
page = _client(tmp_path).get("/b/g/").text
|
|
fig = lambda rel: re.search(r'<figure[^>]*data-item="%s".*?</figure>' % re.escape(rel), page, re.S).group(0)
|
|
own, booth = fig("a.png"), fig("b.png")
|
|
assert re.search(r'action="/b/g/blur".*?name="on" value="0".*?◉ blurred', own, re.S)
|
|
assert 'action="/b/g/blur"' not in booth
|
|
assert "◉ booth" in booth
|
|
|
|
|
|
|
|
def test_a_board_with_files_gets_the_blur_controls_its_labels_point_at(tmp_path):
|
|
"""heid code-review (4/4): both booth-wide controls sat inside the board
|
|
suppression meant for the one-click wipe, so a links board holding a fogged
|
|
picture showed "◉ booth — un-blur the booth in the header" with no such
|
|
control in the header. Only the wipe is board-suppressed."""
|
|
b = _booth(tmp_path, "links", {"links.md": b"- [x](https://example.test/)\n", "a.png": PNG})
|
|
(b / ".blurbooth").write_bytes(b"")
|
|
page = _client(tmp_path).get("/b/links/").text
|
|
assert re.search(r'<form[^>]*action="/b/links/blurbooth"', page)
|
|
assert re.search(r"<button[^>]*data-reveal-all", page)
|
|
|
|
|
|
def test_reveal_all_renders_where_it_can_act(tmp_path):
|
|
"""heid code-review (3/4): the header offers Reveal all when ANY item is
|
|
blurred; the review only when an item of the review RING is — a blurred doc
|
|
is not on the review page, so a control there would act on nothing."""
|
|
from booth.app import set_blurred
|
|
b = _booth(tmp_path, "g", {"a.png": PNG, "n.md": b"# n"})
|
|
set_blurred(b, "n.md", True)
|
|
c = _client(tmp_path)
|
|
assert re.search(r"<button[^>]*data-reveal-all", c.get("/b/g/").text)
|
|
assert not re.search(r"<button[^>]*data-reveal-all", c.get("/b/g/view?f=a.png").text)
|
|
|
|
|
|
def test_a_blurred_docs_own_page_is_blurred_too(tmp_path):
|
|
"""heid code-review (hulda): the full-page doc view never read `blurred`,
|
|
so a blurred doc rendered clear at the size where it is most readable.
|
|
Its body is blurred there too, with its own JS-only reveal."""
|
|
from booth.app import set_blurred
|
|
b = _booth(tmp_path, "g", {"n.md": b"# secret", "o.md": b"# open"})
|
|
set_blurred(b, "n.md", True)
|
|
c = _client(tmp_path)
|
|
blurred, clear = c.get("/b/g/view?f=n.md").text, c.get("/b/g/view?f=o.md").text
|
|
assert re.search(r'class="docbody is-blurred"', blurred)
|
|
assert re.search(r'<button[^>]*id="docreveal"[^>]*hidden', blurred)
|
|
assert 'class="docbody"' in clear and not re.search(r'<button[^>]*id="docreveal"', clear)
|
|
|
|
|
|
def test_fogging_never_writes_through_a_planted_marker_link(tmp_path):
|
|
"""heid bug-hunt (kimi, hulda): `marker.touch()` followed a planted
|
|
`.blurbooth` symlink — a click of the new browser control rewrote an
|
|
outside file's mtime, or CREATED a dangling target. The same class
|
|
`record_view` was hardened against. A link already there reads as fogged
|
|
(is_booth_blurred counts it), so fogging has nothing to write."""
|
|
import os
|
|
outside = tmp_path / "outside.txt"
|
|
outside.write_text("x")
|
|
os.utime(outside, (1_000_000, 1_000_000))
|
|
b = _booth(tmp_path, "g", {"a.png": PNG})
|
|
(b / ".blurbooth").symlink_to(outside)
|
|
h = _booth(tmp_path, "h", {"a.png": PNG})
|
|
(h / ".blurbooth").symlink_to(tmp_path / "created-by-a-click")
|
|
c = _client(tmp_path)
|
|
assert c.post("/b/g/blurbooth", data={"on": "1"}, follow_redirects=False).status_code == 303
|
|
assert c.post("/b/h/blurbooth", data={"on": "1"}, follow_redirects=False).status_code == 303
|
|
assert outside.stat().st_mtime == 1_000_000
|
|
assert not (tmp_path / "created-by-a-click").exists()
|
|
c.post("/b/g/blurbooth", data={"on": "0"}, follow_redirects=False)
|
|
assert not (b / ".blurbooth").is_symlink() and outside.exists() # unlinked, target untouched
|
|
|
|
|
|
def test_the_fog_landing_is_built_from_the_ring_never_echoed(tmp_path):
|
|
"""heid bug-hunt (kimi, regin, groa): `back` went into the 303 unchecked, so
|
|
a stale or foreign value landed on a 404. Like the mark routes' back=view:
|
|
the review only for an item of the review ring, else the booth page."""
|
|
_booth(tmp_path, "g", {"a.png": PNG, "n.md": b"# n"})
|
|
c = _client(tmp_path)
|
|
loc = lambda back: c.post("/b/g/blurbooth", data={"on": "1", "back": back},
|
|
follow_redirects=False).headers["location"]
|
|
assert loc("a.png") == "/b/g/view?f=a.png"
|
|
assert loc("gone.png") == "/b/g/"
|
|
assert loc("n.md") == "/b/g/"
|
|
assert loc("") == "/b/g/"
|
|
|
|
|
|
def _css_blocks(css: str) -> list[tuple[str, list[str]]]:
|
|
"""[(media prelude | selector, declarations)] in source order. Enough of a
|
|
parser for the vendored sheet: blocks are one level deep, or one inside a
|
|
single @media."""
|
|
out, media = [], ""
|
|
for m in re.finditer(r'(@media[^{]+)\{|([^{}@]+?)\s*\{([^{}]*)\}|\}', css):
|
|
if m.group(1):
|
|
media = " ".join(m.group(1).split())
|
|
elif m.group(2) is not None:
|
|
key = (media + " | " + " ".join(m.group(2).split()).split("*/")[-1].strip()).strip()
|
|
body = re.sub(r"/\*.*?\*/", "", m.group(3), flags=re.S) # a comment hides the declaration after it
|
|
decls = [d.strip() for d in body.split(";") if d.strip()]
|
|
out.append((key, decls))
|
|
else:
|
|
media = ""
|
|
return out
|
|
|
|
|
|
def test_a_forced_theme_and_the_os_theme_are_the_same_declarations():
|
|
"""r2b D3: light is written twice — under the OS query for
|
|
`:root:not([data-theme="dark"])`, and bare for `:root[data-theme="light"]`
|
|
— and light-hc twice likewise. The two copies of each must be the same
|
|
declarations, value for value, or forcing a theme would give a different
|
|
theme than the OS asking for it.
|
|
|
|
And each must declare exactly the property set of the block it overrides,
|
|
read from the UNMOVED dark block (SVOS's light and dark declare the same
|
|
properties, as do its two high-contrast blocks). A check that only compared
|
|
the re-scoped copies with each other would pass a transform that dropped
|
|
the same declaration from both (heid contract panel, 4/4)."""
|
|
css = (pathlib.Path(__file__).parent.parent / "booth/templates/_svos_tokens.css").read_text()
|
|
blocks = _css_blocks(css)
|
|
by = {}
|
|
for key, decls in blocks:
|
|
by.setdefault(key, []).append(decls)
|
|
props = lambda decls: {d.split(":", 1)[0].strip() for d in decls}
|
|
dark = by["| :root"][1] # primitives, DARK, art layer
|
|
art = by["| :root"][2]
|
|
[os_light] = by['@media (prefers-color-scheme: light) | :root:not([data-theme="dark"])']
|
|
[forced_light] = by['| :root[data-theme="light"]']
|
|
[dhc] = by["@media (prefers-contrast: more) | :root"]
|
|
[os_lhc] = by['@media (prefers-contrast: more) and (prefers-color-scheme: light) | :root:not([data-theme="dark"])']
|
|
[forced_lhc] = by['@media (prefers-contrast: more) | :root[data-theme="light"]']
|
|
assert "color-scheme: dark" in dark and len(dark) > 30 # the parser found the real dark block
|
|
assert os_light == forced_light
|
|
assert os_lhc == forced_lhc
|
|
# Light overrides every dark property plus the art layer's four light-only
|
|
# values. That extra set is written here, from SVOS, NOT derived from the
|
|
# copies: deriving it from them let a transform that dropped an art-light
|
|
# value from BOTH copies pass (heid code-review, 2/4).
|
|
art_light = {"--shadow-sm", "--shadow-md", "--shadow-lg", "--glow-armed"}
|
|
assert art_light <= props(art)
|
|
assert props(forced_light) == props(dark) | art_light
|
|
assert props(forced_lhc) == props(dhc)
|
|
|
|
|
|
def _row(body: str, name: str) -> str:
|
|
return re.search(r'<article class="desk-row[^"]*" data-booth="%s".*?</article>' % re.escape(name), body, re.S).group(0)
|
|
|
|
|
|
def test_the_lifetime_pill_class_is_kept_held_or_counting(tmp_path):
|
|
"""r2b D1 (operator: 'make it obvious which are kept and which are
|
|
ephemeral'): the lifetime is a pill in the right column, its CLASS chosen
|
|
by state and its words exactly the lifetime macro's."""
|
|
from booth.marks import declare_pick
|
|
k = _booth(tmp_path, "kept1", {"a.png": PNG})
|
|
(k / ".forever").write_bytes(b"")
|
|
h = _booth(tmp_path, "held", {"a.png": PNG})
|
|
declare_pick(h, "q", {"prompt": "Which?", "options": ["x", "y"]})
|
|
_booth(tmp_path, "loose", {"a.png": PNG})
|
|
broken = _booth(tmp_path, "broken", {"a.png": PNG})
|
|
(broken / ".marks.json").write_text("{not json")
|
|
body = _client(tmp_path).get("/").text
|
|
pill = lambda n: re.search(r'<span class="life (life-\w+)"[^>]*>(.*?)</span>\s*</div>', _row(body, n), re.S)
|
|
assert pill("kept1").group(1) == "life-kept" and "kept" in pill("kept1").group(2)
|
|
assert pill("held").group(1) == "life-held" and "held until answered" in pill("held").group(2)
|
|
assert pill("loose").group(1) == "life-count" and "expires in" in pill("loose").group(2)
|
|
assert pill("broken").group(1) == "life-held" and "marks unreadable" in pill("broken").group(2)
|
|
# and the lifetime left the facts line, which is facts only
|
|
assert "expires in" not in re.search(r'class="desk-facts".*?</div>', _row(body, "loose"), re.S).group(0)
|
|
|
|
|
|
def test_created_and_updated_are_dated_facts_and_none_says_nothing(tmp_path, monkeypatch):
|
|
"""r2b D1b (operator: 'creation and update dates on the booths'): created is
|
|
a date, updated an age, each a <time> with its full stamp as the title, on
|
|
the Desk row's facts line and in the booth header. A birth time the
|
|
filesystem cannot give renders NOTHING — never a guess (booth-dev)."""
|
|
import os
|
|
import booth.app as app_mod
|
|
t0 = time.time() - 5 * 86400
|
|
b = _booth(tmp_path, "g", {"a.png": PNG})
|
|
os.utime(b / "a.png", (t0, t0))
|
|
_booth(tmp_path, "nobirth", {"a.png": PNG})
|
|
real = app_mod.birth_time
|
|
monkeypatch.setattr(app_mod, "birth_time", lambda p: None if p.name == "nobirth" else t0 - 86400)
|
|
c = _client(tmp_path)
|
|
facts = lambda n: re.search(r'class="desk-facts".*?</div>', _row(c.get("/").text, n), re.S).group(0)
|
|
g = facts("g")
|
|
assert re.search(r'<time[^>]*datetime="\d{4}-\d\d-\d\dT[^"]+"[^>]*title="created [^"]+"[^>]*>created [^<]+</time>', g)
|
|
assert re.search(r'<time[^>]*title="updated [^"]+"[^>]*>updated 5d ago</time>', g)
|
|
assert "created" not in facts("nobirth")
|
|
head = c.get("/b/g/").text
|
|
assert re.search(r'<time[^>]*>created [^<]+</time>', head) and re.search(r'<time[^>]*>updated 5d ago</time>', head)
|
|
monkeypatch.setattr(app_mod, "birth_time", real)
|
|
|
|
|
|
def test_a_forced_theme_is_applied_before_first_paint(tmp_path):
|
|
"""r2b D3 / INV-6: the script that applies a stored theme runs in <head>
|
|
BEFORE any stylesheet, so a forced theme never flashes the other one. The
|
|
toggle is markup on every page, always `hidden` until the script shows it."""
|
|
_booth(tmp_path, "g", {"a.png": PNG})
|
|
c = _client(tmp_path)
|
|
for url in ("/", "/b/g/"):
|
|
page = c.get(url).text
|
|
head = page[:page.index("</head>")]
|
|
assert head.index("booth.theme") < head.index("<style>"), url
|
|
assert re.search(r'<div class="theme"[^>]*hidden', page), url
|
|
|
|
|
|
def test_a_date_no_calendar_can_hold_renders_nothing_and_never_500s(tmp_path, monkeypatch):
|
|
"""heid bug-hunt (4/4): the date filters raised on a timestamp outside the
|
|
calendar's range, and the Desk renders every row in one response — one
|
|
poisoned clock 500'd the index for every booth. A date that cannot be
|
|
rendered renders nothing, like a birth time the disk cannot give."""
|
|
import booth.app as app_mod
|
|
_booth(tmp_path, "g", {"a.png": PNG})
|
|
_booth(tmp_path, "ok", {"a.png": PNG})
|
|
monkeypatch.setattr(app_mod, "birth_time", lambda p: 1e20 if p.name == "g" else None)
|
|
monkeypatch.setattr(app_mod, "_content_mtime", lambda p: -1e20 if p.name == "g" else time.time() - 3600)
|
|
c = _client(tmp_path)
|
|
r = c.get("/")
|
|
assert r.status_code == 200
|
|
assert "created" not in re.search(r'class="desk-facts".*?</div>', _row(r.text, "g"), re.S).group(0)
|
|
assert c.get("/b/g/").status_code == 200
|
|
|
|
|
|
@pytest.mark.parametrize("age,words", [(30, "just now"), (59 * 60, "59m ago"), (23 * 3600, "23h ago"),
|
|
(30 * 3600, "1d ago"), (47 * 3600, "1d ago"), (5 * 86400, "5d ago")])
|
|
def test_an_age_is_said_in_its_largest_whole_unit(age, words):
|
|
"""heid bug-hunt (groa): hours ran on to 47h, so "1d ago" never appeared,
|
|
against the helper's own docstring."""
|
|
from booth.app import date_ago
|
|
assert date_ago(age) == words
|
|
|
|
|
|
def test_updated_shows_whenever_it_differs_from_created_and_a_future_one_says_its_date(tmp_path, monkeypatch):
|
|
"""heid bug-hunt (hulda, groa): "updated" was dropped for content OLDER than
|
|
the booth — copied files keep their mtimes while the folder is born now —
|
|
and a future content clock read "updated just now". Either direction of a
|
|
real gap shows; a clock ahead of now shows its date, never an age."""
|
|
import booth.app as app_mod
|
|
now = time.time()
|
|
_booth(tmp_path, "copied", {"a.png": PNG})
|
|
_booth(tmp_path, "ahead", {"a.png": PNG})
|
|
monkeypatch.setattr(app_mod, "birth_time", lambda p: now - 3600)
|
|
monkeypatch.setattr(app_mod, "_content_mtime",
|
|
lambda p: now - 30 * 86400 if p.name == "copied" else now + 5 * 86400)
|
|
body = _client(tmp_path).get("/").text
|
|
facts = lambda n: re.search(r'class="desk-facts".*?</div>', _row(body, n), re.S).group(0)
|
|
assert re.search(r">updated 30d ago</time>", facts("copied"))
|
|
ahead = facts("ahead")
|
|
assert "just now" not in ahead and re.search(r">updated \d{1,2} [A-Z][a-z]{2}( \d{4})?</time>", ahead)
|