vh
d88f235688
Kept boards had no delete path in the UI at all. The kept lane deliberately omits the wipe control -- a one-click wipe next to the durable stuff is a footgun -- but "deliberate" had been implemented as "impossible": the only routes out were ssh or a hand-written API call. Now it is two deliberate acts. A `release` control on kept cards drops the sentinel, the board moves to the ephemeral lane, and the existing x wipes it from there. Release is reversible -- POST /b/<name>/keep pins it again. POST /b/<name>/unkeep release the pin POST /b/<name>/keep pin it (round-trip, so release is not a one-way door) FOUND WHILE TESTING, and it invalidates the previously-documented workaround: removing the sentinel BUMPS the booth directory's mtime, and booth age is the newest mtime in the tree -- so a released board's clock RESETS from 10,000s to 0s and it survives another full TTL. The old comment said "remove the sentinel first (it rejoins the sweep)", which is true but means the board lives another 24h, not that it gets reaped. Unkeep-and-wait is a delay, not a delete. test_releasing_a_board_RESETS_its_ttl_clock pins that behaviour deliberately so nobody re-derives the workaround. Release is what unlocks the x; the x is what deletes. CLI: `booth rm` already worked on kept boards but said nothing about it. It now announces "(was KEPT -- durable board)" so wiping something durable can never look identical to wiping run output. Not a block -- a CLI user naming a booth is being explicit. 5 new tests (67 pass). Verified live on nh3-dev: release renders on all four kept boards, the ephemeral lane keeps its x, and the links board is untouched with its sentinel intact.
839 lines
29 KiB
Python
839 lines
29 KiB
Python
import os
|
|
import re
|
|
import time
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from booth.app import (
|
|
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 "<h1>Title</h1>" in it["rendered"]
|
|
assert "<strong>bold</strong>" 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 <not html> 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>.
|
|
# Pre-escaping plus template autoescape would double-encode the angle brackets.
|
|
assert it["rendered"] == "plain <not html> 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 <details> disclosure — no navigation
|
|
assert "<details" in html
|
|
assert "<h1>Heading</h1>" 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 <x> & <y>")
|
|
|
|
html = c.get("/b/run1/").text
|
|
|
|
# exactly one level of HTML-escaping (template autoescape inside <pre>),
|
|
# 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 "<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_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("<h1>BRIEF</h1>")
|
|
(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("<h1>BRIEF</h1>")
|
|
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 = "<html><head><title>Brief</title></head><body><h1>REPORT</h1></body></html>"
|
|
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 "<h1>REPORT</h1>" in out # original content preserved
|
|
# favicon lands in the head, chip lands in the body
|
|
assert out.index(FAVICON_LINK) < out.index("</head>")
|
|
assert out.index("booth-nav-home") > out.index("<body>")
|
|
|
|
|
|
def test_wrap_respects_existing_favicon():
|
|
html = '<html><head><link rel="icon" href="data:image/png;base64,AAAA"></head><body>x</body></html>'
|
|
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("<h1>bare fragment</h1>") # no doctype/head/body
|
|
assert 'class="booth-nav-home"' in out
|
|
assert out.rstrip().endswith("</style>") # 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("<body><h1>no head</h1></body>")
|
|
assert 'class="booth-nav-home"' in out
|
|
assert FAVICON_LINK in out # injected even without an explicit <head>
|
|
|
|
|
|
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 = "<!doctype html><meta charset=utf-8><title>T</title><style>body{margin:0}</style><h1>REPORT</h1>"
|
|
out = wrap_verbatim_html(html)
|
|
assert out.lstrip().lower().startswith("<!doctype") # doctype still first -> standards mode
|
|
assert FAVICON_LINK in out
|
|
assert out.index(FAVICON_LINK) < out.index("<h1>") # 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("<h1>REPORT</h1>") # 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("<html><head></head><body><h1>BRIEF</h1></body></html>")
|
|
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/<name>/index.html) still serves the raw bytes — the chip
|
|
# only rides on the booth view (/b/<name>/), so downloads/assets stay verbatim
|
|
c, data = client
|
|
d = data / "brief"
|
|
d.mkdir()
|
|
(d / "index.html").write_text("<html><body><h1>BRIEF</h1></body></html>")
|
|
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 "<h1>" in html and "Title" in html
|
|
assert "<li>" 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 "<h1>" in r.text and "Heading" in r.text # rendered, not raw markdown
|
|
assert "<strong>bold</strong>" 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 "<pre" in r.text and "line one" in r.text
|
|
assert "attachment" not in r.headers.get("content-disposition", "")
|
|
|
|
|
|
def test_gallery_inlines_docs_with_fullpage_and_download_affordances(client):
|
|
# Docs now render INLINE in the gallery (see the inline-doc tests above),
|
|
# not as a link. The full-page viewer stays reachable via the ⤢ affordance,
|
|
# and the raw file via a download link — but the doc content itself is on
|
|
# the page, not behind a click.
|
|
c, data = client
|
|
d = data / "run1"; d.mkdir()
|
|
(d / "readme.md").write_text("# hi")
|
|
(d / "notes.txt").write_text("hello") # loose txt (no media partner) -> own item
|
|
page = c.get("/b/run1/").text
|
|
assert "<h1>hi</h1>" 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
|