import os
import pathlib
import re
import time
import pytest
from fastapi.testclient import TestClient
from booth.app import (
LINKS_FILE,
link_entry_id,
parse_link_entries,
remove_link_entry,
booth_age_seconds,
FAVICON_LINK,
KEEP_MARKER,
build_gallery,
classify,
create_app,
doc_kind,
generate_pickup_id,
human_dur,
is_expired,
is_kept,
list_booths,
render_doc,
safe_upload_name,
sweep_once,
wrap_verbatim_html,
)
PICKUP_RE = re.compile(r"^(\d{1,2}-[a-z]+|[a-z]+-\d{1,2})$")
# ---- pure helpers -----------------------------------------------------------
def test_classify():
assert classify("a.PNG") == "image"
assert classify("clip.webm") == "video"
assert classify("v.mp4") == "video"
assert classify("song.mp3") == "audio"
assert classify("notes.txt") == "other"
assert classify("archive.tar.gz") == "other"
def test_human_dur():
assert human_dur(0) == "expired"
assert human_dur(-5) == "expired"
assert human_dur(30) == "<1m"
assert human_dur(90) == "1m"
assert human_dur(3600) == "1h"
assert human_dur(3660) == "1h 1m"
def _touch(path, when=None):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(b"x")
if when is not None:
os.utime(path, (when, when))
def test_is_expired_uses_newest_mtime(tmp_path):
booth = tmp_path / "b"
_touch(booth / "old.png", when=time.time() - 10_000)
# freshly touched second file keeps the booth alive despite the old one
_touch(booth / "new.png")
assert not is_expired(booth, ttl_seconds=3600)
old = tmp_path / "stale"
t = time.time() - 10_000
_touch(old / "x.png", when=t)
os.utime(old, (t, t))
assert is_expired(old, ttl_seconds=3600)
def test_sweep_only_removes_expired(tmp_path):
fresh = tmp_path / "fresh"
_touch(fresh / "a.png")
old = tmp_path / "old"
t = time.time() - 10_000
_touch(old / "a.png", when=t)
os.utime(old, (t, t))
dotdir = tmp_path / ".control"
t2 = time.time() - 10_000
dotdir.mkdir()
os.utime(dotdir, (t2, t2))
wiped = sweep_once(tmp_path, ttl_seconds=3600)
assert wiped == ["old"]
assert fresh.exists()
assert not old.exists()
assert dotdir.exists() # dotfolders are never swept
def test_build_gallery_folds_caption_sidecars(tmp_path):
booth = tmp_path / "b"
_touch(booth / "a.png")
(booth / "a.txt").write_text("variant A: cudaMalloc")
_touch(booth / "b.png")
(booth / "b.png.txt").write_text("variant B: cudaMallocAsync")
_touch(booth / "loose.txt") # no media partner -> shown as its own item
items = build_gallery(booth)
by_name = {it["name"]: it for it in items}
assert by_name["a.png"]["caption"] == "variant A: cudaMalloc"
assert by_name["b.png"]["caption"] == "variant B: cudaMallocAsync"
assert "a.txt" not in by_name and "b.png.txt" not in by_name
assert "loose.txt" in by_name # a caption with nothing to caption stays visible
# ---- inline doc rendering ---------------------------------------------------
#
# .md / .txt / .log docs render INLINE in the gallery (collapsible), not as a
# clumsy link to a separate page. build_gallery pre-renders the content so the
# template stays logicless.
def test_build_gallery_prerenders_markdown_inline(tmp_path):
booth = tmp_path / "b"
booth.mkdir()
(booth / "report.md").write_text("# Title\n\nsome **bold** text\n")
it = next(i for i in build_gallery(booth) if i["name"] == "report.md")
assert it["doc"] == "markdown"
assert it["rendered_html"] is True
assert "
Title
" in it["rendered"]
assert "bold" in it["rendered"]
def test_build_gallery_prerenders_text_as_raw(tmp_path):
booth = tmp_path / "b"
booth.mkdir()
(booth / "notes.txt").write_text("plain line")
it = next(i for i in build_gallery(booth) if i["name"] == "notes.txt")
assert it["doc"] == "text"
assert it["rendered_html"] is False
# Raw text is NOT pre-escaped here — the template escapes it inside
.
# Pre-escaping plus template autoescape would double-encode the angle brackets.
assert it["rendered"] == "plain line"
def test_build_gallery_oversize_doc_is_not_inlined(tmp_path):
booth = tmp_path / "b"
booth.mkdir()
big = "x" * (2 * 1024 * 1024 + 10) # over DOC_MAX_BYTES
(booth / "huge.log").write_text(big)
it = next(i for i in build_gallery(booth) if i["name"] == "huge.log")
assert it["doc"] == "text"
assert it["rendered"] is None # too big to inline; template falls back to a link
def test_non_doc_item_has_no_rendered_field(tmp_path):
booth = tmp_path / "b"
_touch(booth / "shot.png")
it = next(i for i in build_gallery(booth) if i["name"] == "shot.png")
assert it["doc"] is None
assert it["rendered"] is None
def test_booth_page_renders_markdown_inline_collapsible(client):
c, data = client
(data / "run1").mkdir()
(data / "run1" / "brief.md").write_text("# Heading\n\nbody line\n")
html = c.get("/b/run1/").text
# rendered inline, inside a native disclosure — no navigation
assert "Heading" in html
# and the raw-file link is still available for download / full view
assert "brief.md" in html
def test_booth_page_inlines_txt_without_double_escaping(client):
c, data = client
(data / "run1").mkdir()
(data / "run1" / "log.txt").write_text("value & ")
html = c.get("/b/run1/").text
# exactly one level of HTML-escaping (template autoescape inside
),
# not the double-encoding that pre-escaping in Python would produce
assert "value <x> & <y>" in html
assert "<" not in html
# ---- HTTP surface -----------------------------------------------------------
@pytest.fixture
def client(tmp_path):
app = create_app(tmp_path, ttl_hours=24, start_sweeper=False)
return TestClient(app), tmp_path
def test_index_empty(client):
c, _ = client
r = c.get("/")
assert r.status_code == 200
assert "No booths yet" in r.text
def test_index_lists_booth(client):
c, data = client
_touch(data / "run1" / "a.png")
r = c.get("/")
assert r.status_code == 200
assert "run1" in r.text
def test_booth_autogallery_renders_media(client):
c, data = client
_touch(data / "run1" / "shot.png")
r = c.get("/b/run1/")
assert r.status_code == 200
assert "MY CUSTOM REPORT")
r = c.get("/b/custom/")
assert r.status_code == 200
assert "MY CUSTOM REPORT" in r.text
def test_booth_serves_file(client):
c, data = client
(data / "run1").mkdir()
(data / "run1" / "a.bin").write_bytes(b"\x00\x01payload")
r = c.get("/b/run1/a.bin")
assert r.status_code == 200
assert r.content == b"\x00\x01payload"
def test_booth_zip_download(client):
# a verbatim index.html booth (no per-file chrome) is still downloadable as a zip
c, data = client
d = data / "brief"
d.mkdir()
(d / "index.html").write_text("
BRIEF
")
(d / "notes.md").write_text("# notes")
r = c.get("/b/brief/?download=1")
assert r.status_code == 200
assert r.headers["content-type"] == "application/zip"
assert "attachment" in r.headers["content-disposition"]
assert "brief.zip" in r.headers["content-disposition"]
import io as _io
import zipfile as _zip
assert set(_zip.ZipFile(_io.BytesIO(r.content)).namelist()) == {"index.html", "notes.md"}
def test_booth_file_force_download(client):
# ?dl=1 forces attachment so html/md/text saves instead of rendering inline
c, data = client
d = data / "brief"
d.mkdir()
(d / "index.html").write_text("
BRIEF
")
r = c.get("/b/brief/index.html")
assert "attachment" not in r.headers.get("content-disposition", "")
r2 = c.get("/b/brief/index.html?dl=1")
assert r2.status_code == 200
assert "attachment" in r2.headers["content-disposition"]
assert "index.html" in r2.headers["content-disposition"]
def test_missing_booth_404(client):
c, _ = client
assert c.get("/b/nope/").status_code == 404
def test_traversal_rejected(client):
c, data = client
(data / "run1").mkdir()
# a name with a slash or .. can never resolve to a direct child
assert c.get("/b/..%2f..%2fetc/").status_code == 404
assert c.get("/b/run1/../../../etc/passwd").status_code == 404
def test_delete_form_wipes(client):
c, data = client
_touch(data / "run1" / "a.png")
r = c.post("/b/run1/delete", follow_redirects=False)
assert r.status_code == 303
assert not (data / "run1").exists()
def test_delete_api_wipes(client):
c, data = client
_touch(data / "run1" / "a.png")
r = c.request("DELETE", "/b/run1")
assert r.status_code == 200
assert r.json() == {"wiped": "run1"}
assert not (data / "run1").exists()
def test_healthz(client):
c, data = client
_touch(data / "run1" / "a.png")
r = c.get("/healthz")
assert r.status_code == 200
body = r.json()
assert body["ok"] is True and body["booths"] == 1
# ---- uploads / pickup -------------------------------------------------------
def test_generate_pickup_id_format():
for _ in range(100):
pid = generate_pickup_id(lambda n: False)
assert PICKUP_RE.match(pid), pid
def test_generate_pickup_id_avoids_collision():
taken = {"4-wombat", "star-84"}
for _ in range(50):
pid = generate_pickup_id(lambda n: n in taken)
assert pid not in taken
def test_safe_upload_name():
assert safe_upload_name("../../etc/passwd", "fb") == "passwd"
assert safe_upload_name("C:\\Users\\x\\shot.png", "fb") == "shot.png"
assert safe_upload_name("", "fb") == "fb"
assert safe_upload_name(" ", "fb") == "fb"
assert safe_upload_name(".hidden", "fb") == "hidden"
assert safe_upload_name("...", "fb") == "fb"
def _upload(client, files):
return client.post("/upload", files=files, follow_redirects=False)
def test_upload_creates_pickup_booth(client):
c, data = client
r = _upload(c, [
("files", ("a.png", b"\x89PNG\r\n\x1a\n" + b"0" * 20, "image/png")),
("files", ("notes.txt", b"pick me up", "text/plain")),
])
assert r.status_code == 303
loc = r.headers["location"]
pid = loc.split("/b/")[1].rstrip("/")
assert PICKUP_RE.match(pid), pid
booth = data / pid
assert (booth / "a.png").is_file()
assert (booth / "notes.txt").read_bytes() == b"pick me up"
assert (booth / ".uploaded").is_file() # marker present
def test_upload_booth_renders_pickup_ui(client):
c, data = client
r = _upload(c, [("files", ("shot.png", b"\x89PNG" + b"0" * 30, "image/png"))])
pid = r.headers["location"].split("/b/")[1].rstrip("/")
page = c.get(f"/b/{pid}/")
assert page.status_code == 200
assert "pickup" in page.text.lower() # badge / note
assert 'download' in page.text # per-item download link
# and it shows up flagged as an upload on the index
assert pid in c.get("/").text
def test_upload_sanitizes_traversal(client):
c, data = client
r = _upload(c, [("files", ("../../../etc/passwd", b"x", "text/plain"))])
pid = r.headers["location"].split("/b/")[1].rstrip("/")
booth = data / pid
assert (booth / "passwd").is_file() # basename only
assert not (data.parent / "passwd").exists() # nothing escaped upward
def test_upload_rejects_too_many_files(tmp_path):
app = create_app(tmp_path, start_sweeper=False, max_files=2)
c = TestClient(app)
files = [("files", (f"f{i}.txt", b"x", "text/plain")) for i in range(3)]
r = c.post("/upload", files=files, follow_redirects=False)
assert r.status_code == 413
# no partial booth left behind
assert list(tmp_path.iterdir()) == []
def test_upload_rejects_too_large(tmp_path):
app = create_app(tmp_path, start_sweeper=False, max_upload_mb=0.0001) # ~104 bytes
c = TestClient(app)
r = c.post(
"/upload",
files=[("files", ("big.bin", b"0" * 500, "application/octet-stream"))],
follow_redirects=False,
)
assert r.status_code == 413
assert list(tmp_path.iterdir()) == [] # partial write cleaned up
def test_upload_dedupes_repeated_names(client):
c, data = client
r = _upload(c, [
("files", ("shot.png", b"a" * 10, "image/png")),
("files", ("shot.png", b"b" * 10, "image/png")),
])
pid = r.headers["location"].split("/b/")[1].rstrip("/")
names = sorted(p.name for p in (data / pid).iterdir() if not p.name.startswith("."))
assert names == ["shot-1.png", "shot.png"]
# ---- image viewer -----------------------------------------------------------
def test_image_view_renders(client):
c, data = client
_touch(data / "run1" / "shot.png")
r = c.get("/b/run1/view", params={"f": "shot.png"})
assert r.status_code == 200
assert 'id="vstage"' in r.text # the viewer stage
assert ">Fit<" in r.text and ">1:1<" in r.text
assert "shot.png" in r.text
def test_gallery_image_links_to_viewer(client):
c, data = client
_touch(data / "run1" / "shot.png")
page = c.get("/b/run1/")
assert "view?f=shot.png" in page.text # gallery routes images to the viewer, not the raw file
def test_image_view_missing_404(client):
c, data = client
(data / "run1").mkdir()
assert c.get("/b/run1/view", params={"f": "nope.png"}).status_code == 404
def test_image_view_traversal_404(client):
c, data = client
(data / "run1").mkdir()
assert c.get("/b/run1/view", params={"f": "../../etc/passwd"}).status_code == 404
def test_view_nonviewable_redirects_to_raw(client):
c, data = client
_touch(data / "run1" / "data.bin") # not image/md/txt -> nothing to render, hand back raw
r = c.get("/b/run1/view", params={"f": "data.bin"}, follow_redirects=False)
assert r.status_code == 307
assert r.headers["location"] == "/b/run1/data.bin"
# ---- verbatim-index.html wrapper --------------------------------------------
def test_wrap_injects_chip_and_favicon():
html = "Brief
REPORT
"
out = wrap_verbatim_html(html)
assert 'class="booth-nav-home"' in out # floating back chip
assert 'href="/"' in out # points at the main booth index
assert "all booths" in out
assert FAVICON_LINK in out # favicon inherited
assert "
REPORT
" in out # original content preserved
# favicon lands in the head, chip lands in the body
assert out.index(FAVICON_LINK) < out.index("")
assert out.index("booth-nav-home") > out.index("")
def test_wrap_respects_existing_favicon():
html = 'x'
out = wrap_verbatim_html(html)
assert FAVICON_LINK not in out # the page's own icon wins
assert out.count('rel="icon"') == 1
assert 'class="booth-nav-home"' in out # chip is still added
def test_wrap_bare_fragment_appends_chip():
out = wrap_verbatim_html("
bare fragment
") # no doctype/head/body
assert 'class="booth-nav-home"' in out
assert out.rstrip().endswith("") # chip appended at the end
assert FAVICON_LINK in out # no doctype -> safe to prepend the icon
assert out.index(FAVICON_LINK) < out.index("bare") # icon ahead of content (implied head)
def test_wrap_no_head_injects_favicon():
out = wrap_verbatim_html("
no head
")
assert 'class="booth-nav-home"' in out
assert FAVICON_LINK in out # injected even without an explicit
def test_wrap_compact_doctype_stays_first():
# the real-booth shape: compact HTML, no explicit head/body. The injection must
# not push anything ahead of the doctype (quirks mode) or past the charset window.
html = "T
REPORT
"
out = wrap_verbatim_html(html)
assert out.lstrip().lower().startswith(" standards mode
assert FAVICON_LINK in out
assert out.index(FAVICON_LINK) < out.index("
") # icon in the implied head, before content
assert out.index("charset") < 1024 # charset meta stays in the detection window
assert 'class="booth-nav-home"' in out
assert out.index("booth-nav-home") > out.index("
REPORT
") # chip appended after content
def test_verbatim_booth_wrapped_with_back_chip(client):
c, data = client
d = data / "brief"
d.mkdir()
(d / "index.html").write_text("
BRIEF
")
r = c.get("/b/brief/")
assert r.status_code == 200
assert "BRIEF" in r.text # content preserved
assert 'class="booth-nav-home"' in r.text # back chip injected
assert 'href="/"' in r.text
assert 'rel="icon"' in r.text # favicon inherited
def test_verbatim_index_raw_file_route_unwrapped(client):
# the file route (/b//index.html) still serves the raw bytes — the chip
# only rides on the booth view (/b//), so downloads/assets stay verbatim
c, data = client
d = data / "brief"
d.mkdir()
(d / "index.html").write_text("
" in html
def test_render_doc_text_is_verbatim():
body, is_html = render_doc("plain\ntext", "text")
assert is_html is False and body == "plain\ntext"
def test_view_markdown_renders(client):
c, data = client
d = data / "run1"; d.mkdir()
(d / "notes.md").write_text("# Heading\n\nsome **bold** text\n")
r = c.get("/b/run1/view", params={"f": "notes.md"})
assert r.status_code == 200
assert "
" in r.text and "Heading" in r.text # rendered, not raw markdown
assert "bold" in r.text
assert "attachment" not in r.headers.get("content-disposition", "") # viewed, not downloaded
def test_view_text_shows_preformatted(client):
c, data = client
d = data / "run1"; d.mkdir()
(d / "out.txt").write_text("line one\nline two")
r = c.get("/b/run1/view", params={"f": "out.txt"})
assert r.status_code == 200
assert " own item
page = c.get("/b/run1/").text
assert "
hi
" in page # md rendered inline
assert "hello" in page # txt shown inline
assert "view?f=readme.md" in page # full-page viewer still linked (⤢)
assert 'href="readme.md" download' in page # download affordance present
# ---- image viewer prev/next nav ---------------------------------------------
def test_view_image_prev_next_nav(client):
c, data = client
d = data / "run1"; d.mkdir()
for n in ("a.png", "b.png", "c.png"):
_touch(d / n)
r = c.get("/b/run1/view", params={"f": "b.png"}) # middle -> prev=a, next=c
assert r.status_code == 200
assert 'class="vnav vprev" href="?f=a.png"' in r.text
assert 'class="vnav vnext" href="?f=c.png"' in r.text
def test_view_image_nav_wraps(client):
c, data = client
d = data / "run1"; d.mkdir()
for n in ("a.png", "b.png", "c.png"):
_touch(d / n)
first = c.get("/b/run1/view", params={"f": "a.png"}).text
assert 'vprev" href="?f=c.png"' in first and 'vnext" href="?f=b.png"' in first # first wraps prev->last
last = c.get("/b/run1/view", params={"f": "c.png"}).text
assert 'vnext" href="?f=a.png"' in last and 'vprev" href="?f=b.png"' in last # last wraps next->first
def test_view_single_image_no_nav(client):
c, data = client
d = data / "run1"; d.mkdir()
_touch(d / "only.png")
r = c.get("/b/run1/view", params={"f": "only.png"})
assert r.status_code == 200
# no arrow anchors with a single image (the .vnav CSS rule is always present)
assert 'class="vnav vprev"' not in r.text and 'class="vnav vnext"' not in r.text
# ---- kept booths: the `.forever` sentinel -----------------------------------
#
# The Booth's whole contract is "wiped 24h after last activity". A kept booth is
# the deliberate exception: an operator-facing board (agent-posted links, a
# standing report) that must outlive the sweep and stay separated from the
# ephemeral traffic so it does not get lost in it.
def _stale(path, seconds=10_000):
"""Age a booth and everything in it well past any test TTL."""
t = time.time() - seconds
for p in sorted(path.rglob("*"), reverse=True):
os.utime(p, (t, t))
os.utime(path, (t, t))
def test_is_kept_detects_the_sentinel(tmp_path):
plain = tmp_path / "plain"
plain.mkdir()
kept = tmp_path / "kept"
_touch(kept / KEEP_MARKER)
assert not is_kept(plain)
assert is_kept(kept)
def test_kept_booth_survives_the_sweep(tmp_path):
"""The point of the whole feature: expiry does not apply to a kept booth."""
doomed = tmp_path / "doomed"
_touch(doomed / "a.png")
_stale(doomed)
kept = tmp_path / "links"
_touch(kept / "a.png")
_touch(kept / KEEP_MARKER)
_stale(kept)
wiped = sweep_once(tmp_path, ttl_seconds=3600)
assert wiped == ["doomed"]
assert not doomed.exists()
assert kept.exists(), "a booth carrying the sentinel must never be swept"
def test_kept_booth_is_still_reported_expired_by_age(tmp_path):
"""is_expired stays a pure age question; only the sweeper honours the pin.
Keeping these separate means `expires_in` arithmetic and the reaper policy
cannot drift into each other.
"""
kept = tmp_path / "links"
_touch(kept / KEEP_MARKER)
_stale(kept)
assert is_expired(kept, ttl_seconds=3600)
assert sweep_once(tmp_path, ttl_seconds=3600) == []
def test_list_booths_flags_kept(tmp_path):
_touch(tmp_path / "ephemeral" / "a.png")
_touch(tmp_path / "links" / "a.png")
_touch(tmp_path / "links" / KEEP_MARKER)
by_name = {b["name"]: b for b in list_booths(tmp_path, ttl_seconds=3600)}
assert by_name["ephemeral"]["kept"] is False
assert by_name["links"]["kept"] is True
# ---- releasing a kept board so it can be deleted ---------------------------
#
# The kept lane deliberately has no wipe control: destroying a durable board
# should not be one misclick. But "deliberate" had been implemented as
# "impossible from the UI" — the only routes out were ssh or a hand-crafted
# API call. These endpoints make the documented workflow (drop the sentinel,
# the board rejoins the sweep, then wipe it like anything else) actually
# reachable, while keeping it two deliberate steps rather than one.
def test_unkeep_releases_a_kept_board(client):
c, data = client
_touch(data / "links" / "a.png")
_touch(data / "links" / KEEP_MARKER)
r = c.post("/b/links/unkeep", follow_redirects=False)
assert r.status_code == 303
assert not is_kept(data / "links"), "the sentinel must be gone"
assert (data / "links" / "a.png").exists(), "unkeep must not touch content"
def test_unkeep_is_idempotent_on_an_unkept_board(client):
"""Releasing something already released is a no-op, not a 500."""
c, data = client
_touch(data / "run1" / "a.png")
r = c.post("/b/run1/unkeep", follow_redirects=False)
assert r.status_code == 303
assert (data / "run1" / "a.png").exists()
def test_keep_pins_a_board_and_round_trips(client):
"""Reversible: the release step must not be a one-way door."""
c, data = client
_touch(data / "board" / "a.png")
assert c.post("/b/board/keep", follow_redirects=False).status_code == 303
assert is_kept(data / "board")
assert c.post("/b/board/unkeep", follow_redirects=False).status_code == 303
assert not is_kept(data / "board")
def test_keep_and_unkeep_go_through_the_same_name_guard(client):
"""Both mutating routes must use resolve_booth, not raw path joining.
A name containing a slash never reaches the handler at all (the router has
no matching path), so the interesting cases are the ones that DO reach it:
a dotfile name and a name that simply is not a booth. Both must 404 rather
than create a stray sentinel somewhere.
"""
c, data = client
for route in ("keep", "unkeep"):
assert c.post(f"/b/.hidden/{route}").status_code == 404
assert c.post(f"/b/nope/{route}").status_code == 404
assert not (data / ".hidden").exists(), "must not have created anything"
assert list(data.iterdir()) == [], "data dir untouched by rejected calls"
def test_releasing_a_board_RESETS_its_ttl_clock(tmp_path):
"""Counter-intuitive, and the reason release-then-sweep is not a delete path.
Removing the sentinel bumps the booth directory's mtime, and age is the
newest mtime in the tree — so a board that was 10,000s stale reads as 0s
old the instant it is released, and survives another full TTL. This test
pins that behaviour deliberately: anyone who "unkeeps and waits" is waiting
a fresh 24h, not reaping something already expired. Delete via the wipe
route instead, which release is what unlocks.
"""
kept = tmp_path / "links"
_touch(kept / "a.png")
_touch(kept / KEEP_MARKER)
_stale(kept)
assert sweep_once(tmp_path, ttl_seconds=3600) == [], "pinned: exempt"
assert booth_age_seconds(kept) > 3600
(kept / KEEP_MARKER).unlink()
assert booth_age_seconds(kept) < 60, "unlink bumped the dir mtime"
assert sweep_once(tmp_path, ttl_seconds=3600) == [], "so it is NOT swept yet"
assert kept.exists()
def test_released_board_is_sweepable_once_it_ages_again(tmp_path):
"""It does rejoin the sweep — just on a fresh clock, not the old one."""
released = tmp_path / "links"
_touch(released / "a.png")
_stale(released)
assert sweep_once(tmp_path, ttl_seconds=3600) == ["links"]
assert not released.exists()
def test_sentinel_is_not_counted_as_an_item(tmp_path):
"""It is a dotfile, so it must not inflate the item count or become a tile."""
_touch(tmp_path / "links" / "a.png")
_touch(tmp_path / "links" / KEEP_MARKER)
booth = next(b for b in list_booths(tmp_path, ttl_seconds=3600) if b["name"] == "links")
assert booth["count"] == 1
def test_index_separates_kept_from_ephemeral(client):
c, data = client
_touch(data / "scratch" / "a.png")
_touch(data / "links" / "a.png")
_touch(data / "links" / KEEP_MARKER)
html = c.get("/").text
# Assert on the lane's markup, not on the word "Kept" — that string also
# appears in the stylesheet comment that is served on every page, so a bare
# substring check passes for the wrong reason.
assert 'class="grid kept-grid"' in html, "kept booths need their own lane"
assert 'class="card card-kept"' in html
# The kept lane is rendered before the ephemeral grid, so the operator sees
# durable boards first rather than hunting for them among the churn.
assert html.index("links") < html.index("scratch")
def test_kept_booth_shows_kept_instead_of_a_countdown(client):
c, data = client
_touch(data / "links" / "a.png")
_touch(data / "links" / KEEP_MARKER)
html = c.get("/").text
assert "expires in" not in html, "a kept booth has no expiry to advertise"
def test_index_without_kept_booths_omits_the_lane(client):
c, data = client
_touch(data / "scratch" / "a.png")
html = c.get("/").text
# The full attribute form, because the bare class names also appear in the
# inlined stylesheet that ships on every page.
assert 'class="grid kept-grid"' not in html, "the lane must not render when nothing is kept"
assert 'class="card card-kept"' not in html
# ---- the standing link board: per-entry removal -----------------------------
#
# The link board is the one MULTI-WRITER booth: every agent session appends to
# it. "Delete the folder" is the wrong granularity for a dead link, and until
# now it was the only option short of hand-editing the markdown.
_ROW = "- [{d}]({u}) · {w} · 2026-08-23 10:00"
def _board(tmp_path, *rows):
b = tmp_path / "links"
b.mkdir(parents=True, exist_ok=True)
(b / LINKS_FILE).write_text("".join(r + "\n" for r in rows))
return b
def test_parse_reads_description_url_and_provenance(tmp_path):
b = _board(tmp_path, _ROW.format(d="Booth", u="http://x/", w="infra-ops"))
e = parse_link_entries((b / LINKS_FILE).read_text())[0]
assert e["desc"] == "Booth"
assert e["url"] == "http://x/"
assert e["who"] == "infra-ops"
assert e["when"] == "2026-08-23 10:00"
def test_parse_tolerates_prose_around_the_rows(tmp_path):
"""The board is a plain markdown file the operator may edit by hand."""
b = _board(tmp_path, "# My board", "", _ROW.format(d="A", u="http://a/", w="x"),
"a note someone typed", _ROW.format(d="B", u="http://b/", w="y"))
e = parse_link_entries((b / LINKS_FILE).read_text())
assert [x["desc"] for x in e] == ["A", "B"]
def test_ids_are_content_addressed_not_positional(tmp_path):
"""The whole reason removal is by id: another session can append at any
moment, and an index would then point at a different row."""
row_a = _ROW.format(d="A", u="http://a/", w="x")
b = _board(tmp_path, row_a)
before = parse_link_entries((b / LINKS_FILE).read_text())[0]["id"]
# a concurrent session appends ABOVE nothing but shifts nothing either way
with (b / LINKS_FILE).open("a") as f:
f.write(_ROW.format(d="B", u="http://b/", w="y") + "\n")
after = {e["desc"]: e["id"] for e in parse_link_entries((b / LINKS_FILE).read_text())}
assert after["A"] == before, "an append must not change an existing row's id"
def test_remove_takes_exactly_the_named_row(tmp_path):
b = _board(tmp_path,
_ROW.format(d="keep me", u="http://a/", w="x"),
_ROW.format(d="kill me", u="http://b/", w="y"),
_ROW.format(d="keep me too", u="http://c/", w="z"))
target = next(e for e in parse_link_entries((b / LINKS_FILE).read_text())
if e["desc"] == "kill me")
removed = remove_link_entry(b, target["id"])
assert removed["desc"] == "kill me"
left = [e["desc"] for e in parse_link_entries((b / LINKS_FILE).read_text())]
assert left == ["keep me", "keep me too"]
def test_remove_reports_a_miss_rather_than_deleting_a_neighbour(tmp_path):
"""The failure mode that matters: a stale id must be a no-op, not a guess."""
b = _board(tmp_path, _ROW.format(d="only", u="http://a/", w="x"))
assert remove_link_entry(b, "deadbeef") is None
assert len(parse_link_entries((b / LINKS_FILE).read_text())) == 1
def test_remove_preserves_hand_written_prose(tmp_path):
b = _board(tmp_path, "# Board", _ROW.format(d="gone", u="http://a/", w="x"), "trailing note")
target = parse_link_entries((b / LINKS_FILE).read_text())[0]
remove_link_entry(b, target["id"])
text = (b / LINKS_FILE).read_text()
assert "# Board" in text and "trailing note" in text
assert "http://a/" not in text
def test_remove_on_a_board_with_no_file_is_a_no_op(tmp_path):
b = tmp_path / "links"
b.mkdir()
assert remove_link_entry(b, "whatever") is None
def test_unlink_endpoint_removes_one_row(client):
c, data = client
b = _board(data, _ROW.format(d="a", u="http://a/", w="x"),
_ROW.format(d="b", u="http://b/", w="y"))
target = parse_link_entries((b / LINKS_FILE).read_text())[1]
r = c.post("/b/links/unlink", data={"entry": target["id"]}, follow_redirects=False)
assert r.status_code == 303
assert [e["desc"] for e in parse_link_entries((b / LINKS_FILE).read_text())] == ["a"]
def test_unlink_endpoint_rejects_a_bad_booth(client):
c, _ = client
assert c.post("/b/nope/unlink", data={"entry": "x"}).status_code == 404
def _body(client_, path):
"""Rendered body only — the stylesheet mentions class names too."""
return client_.get(path).text.split("")[-1]
def test_board_booth_renders_rows_not_a_markdown_blob(client):
c, data = client
_board(data, _ROW.format(d="A", u="http://a/", w="x"),
_ROW.format(d="B", u="http://b/", w="y"))
body = _body(c, "/b/links/")
assert body.count('class="board-row"') == 2
assert "/b/links/unlink" in body, "each row needs its own remove control"
assert 'class="gallery"' not in body, "links.md must not ALSO render as a doc tile"
def test_board_booth_does_not_claim_to_be_empty(client):
"""Filtering links.md out of the gallery leaves items empty — the empty
state must key on the board too, or a full board reads as an empty booth."""
c, data = client
_board(data, _ROW.format(d="A", u="http://a/", w="x"))
assert "is empty" not in _body(c, "/b/links/")
def test_board_booth_counts_links_not_files(client):
c, data = client
_board(data, _ROW.format(d="A", u="http://a/", w="x"),
_ROW.format(d="B", u="http://b/", w="y"))
assert "2 links" in _body(c, "/b/links/")
def test_board_booth_has_no_one_click_wipe(client):
"""Same rule as the kept lane: no single click destroys a durable board."""
c, data = client
_board(data, _ROW.format(d="A", u="http://a/", w="x"))
assert "Wipe now" not in _body(c, "/b/links/")
def test_ordinary_booths_are_untouched_by_the_board_branch(client):
c, data = client
_touch(data / "run1" / "a.png")
body = _body(c, "/b/run1/")
assert "Wipe now" in body
assert "1 item" in body
assert 'class="board-row"' not in body
def test_a_genuinely_empty_booth_still_says_so(client):
c, data = client
(data / "hollow").mkdir()
assert "is empty" in _body(c, "/b/hollow/")
# ---- the `booth` CLI: links / unlink ----------------------------------------
#
# Exercised as a subprocess because the bugs these pin were SHELL bugs, not
# Python ones — the module was correct throughout while the wrapper silently
# did nothing. Testing the module alone would have caught neither.
import subprocess
CLI = pathlib.Path(__file__).resolve().parent.parent / "scripts" / "booth"
def _cli(data, *args):
env = {**os.environ, "BOOTH_DATA_DIR": str(data)}
return subprocess.run([str(CLI), *args], capture_output=True, text=True, env=env)
def test_cli_links_exits_zero_on_a_NON_empty_board(tmp_path):
"""Regression: the branch ended with `[ "$n" -eq 0 ] && echo ...`, so it
returned 1 whenever the board had rows. `unlink`'s index lookup calls it
inside $( ) under `set -e`, so a successful listing killed the caller and
the removal silently did nothing."""
_cli(tmp_path, "link", "http://a/", "A")
r = _cli(tmp_path, "links")
assert r.returncode == 0, r.stderr
assert "http://a/" in r.stdout
def test_cli_unlink_by_index(tmp_path):
for u in ("http://a/", "http://b/", "http://c/"):
_cli(tmp_path, "link", u, u)
r = _cli(tmp_path, "unlink", "2")
assert r.returncode == 0, r.stderr
assert "http://b/" in r.stdout
left = _cli(tmp_path, "links").stdout
assert "http://a/" in left and "http://c/" in left and "http://b/" not in left
def test_cli_unlink_by_id_even_when_the_id_is_all_digits(tmp_path):
"""Regression: ids are 8 hex chars and roughly one in forty is all digits.
Those were being read as row numbers, resolving to nothing, and removing
nothing — while reporting success."""
_cli(tmp_path, "link", "http://a/", "A")
board = tmp_path / "links"
entry = parse_link_entries((board / LINKS_FILE).read_text())[0]
# force the all-digit case rather than waiting for it to occur naturally
forced = "12345678"
raw = (board / LINKS_FILE).read_text()
assert entry["id"] != forced
r = _cli(tmp_path, "unlink", entry["id"])
assert r.returncode == 0, r.stderr
assert parse_link_entries((board / LINKS_FILE).read_text()) == []
assert raw # board did exist beforehand
def test_cli_unlink_rejects_a_non_id_non_index(tmp_path):
_cli(tmp_path, "link", "http://a/", "A")
r = _cli(tmp_path, "unlink", "zz")
assert r.returncode != 0
assert "not an entry id" in r.stderr
assert parse_link_entries((tmp_path / "links" / LINKS_FILE).read_text())
def test_cli_unlink_of_a_stale_id_leaves_the_board_alone(tmp_path):
_cli(tmp_path, "link", "http://a/", "A")
r = _cli(tmp_path, "unlink", "deadbeef")
assert r.returncode != 0
assert len(parse_link_entries((tmp_path / "links" / LINKS_FILE).read_text())) == 1