import os import re import time import pytest from fastapi.testclient import TestClient from booth.app import ( 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 "&lt;" 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("

BRIEF

") r = c.get("/b/brief/index.html") assert r.status_code == 200 assert "booth-nav-home" not in r.text # ---- .md / .txt in-booth doc viewer ----------------------------------------- def test_doc_kind(): assert doc_kind("notes.md") == "markdown" assert doc_kind("a.markdown") == "markdown" assert doc_kind("log.txt") == "text" assert doc_kind("run.log") == "text" assert doc_kind("shot.png") is None assert doc_kind("data.bin") is None def test_render_doc_markdown(): html, is_html = render_doc("# Title\n\n- a\n- b\n", "markdown") assert is_html is True assert "

" in html and "Title" in html assert "
  • " 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 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