vh
6770ba26d6
Agent sessions hand the operator URLs and they drown in terminal scrollback. The Booth is the right home for them — it already has the one property that decides adoption, which is that a session can publish with mkdir and cp, no API key, no schema, no deploy — but everything in it dies in 24h. So: a booth containing `.forever` is never swept, and renders in its own Kept lane at the top of the index. Opt-in per booth, so the ephemeral default is untouched and nobody inherits a cleanup chore. `rm` the sentinel and the board rejoins the sweep; the CLI verbs are sugar over exactly that, which keeps the filesystem-is-the-state model honest. The pin is deliberately NOT wired into is_expired(). That stays a pure age question feeding the `expires_in` countdown; only sweep_once() honours the sentinel. Keeping expiry arithmetic and reaper policy apart means they cannot drift into each other. Kept cards are visually separated per Australis: a 2px top edge in aurora blue, the one accent border the system sanctions. They show "kept" instead of a countdown, and they deliberately lose the one-click wipe button — a × next to the durable stuff is a footgun, so removing a kept board is a two-step act. `booth link <url> [description]` appends to the standing `links` board, creating and keeping it on first use. Entries carry provenance (handle or hostname, plus a timestamp) because a bare URL is unreadable three days later. The append is one printf of one line to an O_APPEND fd — atomic under PIPE_BUF on POSIX — which matters because many agents post to one board and interleaved half-lines would be the obvious failure mode. Seven tests cover the sentinel: detection, survival of a sweep that wipes its neighbour, the deliberate is_expired/sweep_once split, the listing flag, the sentinel not inflating item counts, and both lane-rendering directions. Two of them originally asserted on the bare strings "Kept" and "kept-grid", which passed for the wrong reason — those also appear in the inlined stylesheet served on every page — so they now assert the full class attribute. 55 pass. Also corrects the Homepage card's description, which advertised a flat 24h TTL that is no longer the whole story.
653 lines
22 KiB
Python
653 lines
22 KiB
Python
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
|
|
|
|
|
|
# ---- 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_links_docs_to_view(client):
|
|
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/")
|
|
assert "view?f=readme.md" in page.text # md -> viewer
|
|
assert "view?f=notes.txt" in page.text # txt -> viewer
|
|
assert 'href="readme.md" download' not in page.text # not a forced download
|
|
|
|
|
|
# ---- 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
|