feat(booth): add The Booth — ephemeral media drop board for CC sessions

A standing user-level web server (nh3-dev :8090) that renders drop-folders under ~/booth-data as ephemeral media "booths" so Claude Code sessions can surface A/B renders and smoke results to the operator, then let them self-wipe. - Scan-and-serve model, no database, no upload API — a booth is just a folder. A folder's own index.html is served verbatim; otherwise an auto-gallery of images / webm+mp4 video / audio is rendered, with <file>.txt caption sidecars folded in (labels A/B pairs). - 24h TTL from newest mtime in the tree; background sweeper wipes stale booths. - Path-traversal + symlink-escape guarded; delete via UI button or DELETE API. - FastAPI + Jinja2, runs from the checkout under systemctl --user (booth.service), alongside the other nh3-dev fleet sidecars. 15 tests, all green. - Homepage tile added (Apps -> The Booth, siteMonitor /healthz). - Harden the homepage rsync doc: exclude *.bak* and logs/ so --delete can't wipe the host's dated services.yaml backups (footgun found deploying this).
2026-07-20 10:17:40 -07:00
parent a5dcad8bd3
commit f4a5ba7c31
13 changed files with 909 additions and 0 deletions
@@ -0,0 +1,184 @@
+import os
+import time
+
+import pytest
+from fastapi.testclient import TestClient
+
+from booth.app import (
+    build_gallery,
+    classify,
+    create_app,
+    human_dur,
+    is_expired,
+    sweep_once,
+)
+
+
+# ---- 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
+
+
+# ---- 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 "<img" in r.text
+    assert "shot.png" in r.text
+
+
+def test_booth_serves_own_index_html(client):
+    c, data = client
+    d = data / "custom"
+    d.mkdir()
+    (d / "index.html").write_text("<h1>MY CUSTOM REPORT</h1>")
+    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_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