Files
booth/tests/test_manifest.py
Vuong Hoang 95beede3c3 fix(manifest)!: the size cap opened a service-wide hang; close it
The diff-scoped bug-hunt panel, four arms, artifact-only. Its strongest
finding is one I created two hours earlier while hardening the reader.

`stat` reports size 0 for a FIFO and 0 for a symlink to /dev/zero, so both
sail under the byte cap added for the RecursionError round — and then
`read_text` either blocks in read() with no EOF, so the except never runs,
or allocates until the kernel intervenes. `list_booths` reads every booth
on every GET / and /healthz, so ONE such file stalls the front page for the
whole service, with no error and no recovery short of a restart.
Reproduced before believing it (timeout returned 124). S_ISREG is checked
BEFORE the size in both modules now; verified against the live service with
two FIFOs planted, which answered 200 in 36ms.

The shape worth carrying: st_size answers a different question than "can
this be read", and a bound that trusts it inherits everything it does not
mean. A hardening fix opened a worse hole than the one it closed.

THE UPLOAD PATH WROTE ABOVE ITS OWN CLEANUP GUARD (4/4)

A failed manifest write orphaned a .uploaded half-booth with no files in
it — and because the temp name now carries a random suffix, nothing ever
overwrote the leak, and .booth.json.<hex>.tmp is not a .lock, so
_newest_mtime counted it and kept that empty booth past every sweep. The
uniqueness fix from the previous round is what made the leak permanent.
Both writes moved inside the guard; the temp is removed on every exit path.

DAMAGED BYTES ARE KEPT, NOT REPLACED (4/4, INV-6)

Marks made this explicit in v0.2.1 and this write path contradicted it: a
manifest that failed on ONE field lost the others with it, including a why
the re-announcer may never have kept anywhere. It diverges from marks in
HOW it honours the rule — marks refuse and answer 409 because the
operator's judgment is not restatable; a manifest quarantines and proceeds,
because refusing would fail `booth add` and lose the files it was copying.

ONE OPENNESS PREDICATE, AS U2 SAID (2/4)

`booth answer` spelled out `if m.answer is None` while `booth marks` asked
`open_marks`, so a partially-answered pick read as done to one verb and
open to the other — at the same instant, on the same booth. U2's INV-2 put
openness in one function precisely so they could not drift. The mirror case
is fixed too: a pick that hydrates broken is refused by the web route, so
`answer --wait` polled an hour on a form nothing could ever land.

ALSO

- now_stamp was whole-second while the importer had moved to microseconds,
  and '-' sorts before '.', so a later mark came out ahead of an earlier
  import inside the same second. One format; the previous round's ordering
  fix had opened this one.
- `_broken` was the third of three directory-name fallbacks and the one
  still handing a raw name into a card's sub-line.
- An identical re-announce rewrote the file and reset the TTL. `booth link`
  does this on every post to the standing board.
- The importer's return went through the bare _hydrate, not _hydrate_safe.
- A marks document could be written larger than it can be read back, and
  then read as no marks at all. Refused at the write instead.
- `choice` reached the answer builder raw while `notes` beside it did not.

AND ONE FINDING DELIBERATELY NOT FULLY CLOSED

The mtime-restore race is real. The clean fix — ignore a booth directory's
own mtime whenever the booth holds anything — also silently retires the
documented rule that releasing a kept board resets its clock, which the CLI
header, the README and a deliberately-written test all pin. That is a TTL
doctrine change, not a bug fix, and an existing test caught the attempt.
The concrete half is fixed (a failing os.utime escaped and 500'd the
route); the race is stated in the code where the next reader will meet it.

341 tests. Live service restarted, 24/24 booth pages verified.
2026-09-22 02:27:18 -07:00

720 lines
29 KiB
Python

"""U5 — self-announcing booths.
A booth carries `.booth.json` saying who posted it and why, and the index card
and the booth page render it. Closes job 5 (`Announce`) — the job nobody named,
whose absence is the measured cause of 145 dead link rows.
See docs/contracts/u5_booth_manifest.contract.md.
"""
import ast
import json
import os
import pathlib
import sys
import pytest
from booth.manifest import MANIFEST_FILE, Manifest, read_manifest, write_manifest
# ---- slice 1: the record and its storage ------------------------------------
def test_an_announcement_round_trips(tmp_path):
b = tmp_path / "r18-ab"
b.mkdir()
written = write_manifest(b, "booth-dev", why="pick the winning denoiser")
assert (b / MANIFEST_FILE).is_file()
got = read_manifest(b)
assert got == written
assert got.handle == "booth-dev"
assert got.why == "pick the winning denoiser"
assert got.error is None
def test_the_title_falls_back_to_the_directory_name(tmp_path):
"""A booth always has a display name. `title` is the one the poster chose
when there is one, and the folder name is a perfectly good one when there
is not — an empty heading on a card is worse than a plain one."""
b = tmp_path / "r18-ab"
b.mkdir()
assert write_manifest(b, "booth-dev").title == "r18-ab"
assert write_manifest(b, "booth-dev", title="R18 A/B").title == "R18 A/B"
def test_a_booth_that_never_announced_reads_as_none(tmp_path):
"""The normal case for every booth that predates this unit, and for every
booth that arrives by rsync — the documented path for any host that is not
nh3-dev, which never runs the CLI at all."""
b = tmp_path / "quiet"
b.mkdir()
assert read_manifest(b) is None
assert read_manifest(tmp_path / "does-not-exist") is None
def test_one_line_by_construction_not_by_convention(tmp_path):
"""`why` renders inside a card's sub-line, so a newline in it would break
the card rather than the field. Truncation and newline-stripping happen at
the WRITE, so nothing downstream has to remember."""
b = tmp_path / "b"
b.mkdir()
m = write_manifest(b, "booth-dev", why="first line\nsecond line\r\nthird")
assert "\n" not in m.why and "\r" not in m.why
assert "first line" in m.why and "second line" in m.why
long = write_manifest(b, "booth-dev", why="x" * 5000)
assert len(long.why) <= 200
assert len(write_manifest(b, "y" * 500).handle) <= 64
assert len(write_manifest(b, "booth-dev", title="t" * 500).title) <= 120
# ---- slice 2: the read cannot raise (INV-2) ---------------------------------
@pytest.mark.parametrize(
"payload",
[
b"{truncated", # not JSON at all
b"[]", # JSON, wrong shape
b'"a string"', # JSON, wronger shape
b"null",
b'{"handle": 7}', # right shape, wrong type
b'{"why": "no handle here"}', # the one required field missing
b"\xff\xfe not utf-8",
b"",
],
ids=["truncated", "list", "string", "null", "wrong-type", "no-handle",
"not-utf8", "empty"],
)
def test_a_damaged_manifest_never_raises(tmp_path, payload):
"""INV-2. `list_booths` calls this once per booth on every index page load,
so a read that can raise is a service-wide outage wearing a single-booth
bug's clothes. That is not a hypothetical — a poisoned `.marks.json` did
exactly that to `/` and `/healthz` across all 25 booths, and the fix shipped
in v0.2.2. The same reader posture, applied before the same mistake."""
b = tmp_path / "b"
b.mkdir()
(b / MANIFEST_FILE).write_bytes(payload)
got = read_manifest(b)
assert isinstance(got, Manifest)
assert got.error, "a damaged manifest read clean"
def test_damaged_is_not_the_same_as_absent(tmp_path):
"""INV-5. Silently folding "cannot be read" into "never announced" would
hide the one case somebody has to go and fix."""
absent = tmp_path / "absent"
absent.mkdir()
damaged = tmp_path / "damaged"
damaged.mkdir()
(damaged / MANIFEST_FILE).write_text("{oops")
assert read_manifest(absent) is None
assert read_manifest(damaged).error
def test_a_manifest_the_module_did_not_write_still_reads(tmp_path):
"""Hand-written is a supported input: the file is plain JSON in a folder the
operator owns, and half the point is that a booth is just a directory. Only
`handle` is required; everything else has a default."""
b = tmp_path / "b"
b.mkdir()
(b / MANIFEST_FILE).write_text(json.dumps({"handle": "shutter-dev"}))
got = read_manifest(b)
assert got.handle == "shutter-dev" and got.error is None
assert got.title == "b"
assert got.why == ""
# ---- slice 3: re-announcement (INV-3) ---------------------------------------
def test_re_announcing_preserves_created(tmp_path):
"""INV-3. `created` is when the booth APPEARED. Saying something more about
it later is not a second appearance, and a `booth add` on an existing booth
is the common case — the poster adds the second batch and sharpens the why."""
b = tmp_path / "b"
b.mkdir()
first = write_manifest(b, "booth-dev", why="first pass")
second = write_manifest(b, "booth-dev", why="second pass, sharper")
assert second.created == first.created
assert second.why == "second pass, sharper"
def test_re_announcing_over_a_damaged_file_does_not_inherit_its_created(tmp_path):
"""A `created` that cannot be read back is replaced rather than guessed at.
The alternative is a stamp that is silently wrong, which is worse than one
that is silently new."""
b = tmp_path / "b"
b.mkdir()
(b / MANIFEST_FILE).write_text("{not json")
m = write_manifest(b, "booth-dev", why="rescued")
assert m.created and m.error is None
assert read_manifest(b).why == "rescued"
# ---- slice 4: the write is atomic, and invisible to every listing -----------
def test_the_write_leaves_no_temp_file(tmp_path):
"""Half of the atomic-write promise, and the weaker half — see
`test_the_write_replaces_rather_than_truncating` for the part that actually
discriminates. Kept because a leaked `.tmp` is its own small defect: it
would sit in the booth forever and, unlike the manifest, nothing would ever
overwrite it."""
b = tmp_path / "b"
b.mkdir()
write_manifest(b, "booth-dev", why="x")
assert not list(b.glob("*.tmp")), "a temp file survived the write"
def test_a_manifest_is_not_an_item(tmp_path):
"""The whole integration story: it is a DOTFILE, so the existing
`startswith('.')` skip in `booth_items` already keeps it out of tiles,
counts and zips. No new exclusion rule anywhere. Asserted rather than
assumed, because the claim is load-bearing for the contract's scope."""
from booth.app import zip_booth
from booth.items import booth_items
b = tmp_path / "b"
b.mkdir()
(b / "a.txt").write_text("real content")
write_manifest(b, "booth-dev", why="x")
assert [i.rel for i in booth_items(b)] == ["a.txt"]
assert MANIFEST_FILE not in zip_booth(b).decode("latin-1")
def test_announcing_is_activity(tmp_path):
"""A manifest is a dotfile but not a `.lock` dotfile, so `_newest_mtime`
counts it. Creating or re-announcing a booth resets its TTL, which is right:
both are somebody touching it. The lock exemption added in v0.2.2 is for
machinery a READ path creates; this is a deliberate write."""
from booth.app import booth_age_seconds
b = tmp_path / "b"
b.mkdir()
old = 1_000_000_000
os.utime(b, (old, old))
write_manifest(b, "booth-dev", why="look at this")
assert booth_age_seconds(b, now=old + 90_000) < 86_400
def test_stdlib_only():
"""INV-4, and the reason this module exists separately from anything that
imports a third-party package. `scripts/booth` imports it under the system
python3 with NO venv, through a `python3 -c` heredoc no AST extractor can
see. It must also not import `booth.*`: a cross-import between two
stdlib-only modules is a second way for the invariant to break."""
src = pathlib.Path(__file__).parent.parent / "booth" / "manifest.py"
roots = set()
for node in ast.walk(ast.parse(src.read_text())):
if isinstance(node, ast.Import):
roots.update(a.name.split(".")[0] for a in node.names)
elif isinstance(node, ast.ImportFrom):
# A RELATIVE import (`from . import marks`) carries no module root
# and used to pass this walk unseen — which matters more here than
# in the shared copy, because this module forbids sibling imports
# outright. Recorded as `booth` so the assertion below catches it.
roots.add("booth" if node.level else
(node.module or "").split(".")[0])
assert not (roots - set(sys.stdlib_module_names)), (
f"booth/manifest.py imports outside the stdlib: "
f"{sorted(roots - set(sys.stdlib_module_names))}"
)
# ---- slice 5: what the operator actually sees -------------------------------
@pytest.fixture
def client(tmp_path):
from fastapi.testclient import TestClient
from booth.app import create_app
return TestClient(create_app(tmp_path, ttl_hours=24, start_sweeper=False)), tmp_path
def _booth(data, name, *, kept=False):
b = data / name
b.mkdir()
(b / "a.txt").write_text("content")
if kept:
(b / ".forever").touch()
return b
@pytest.mark.parametrize("kept", [False, True], ids=["ephemeral", "kept"])
def test_the_index_card_carries_the_announcement(client, kept):
"""BOTH LANES. Kept boards render first and are a separate block in
index.html, so patching only the ephemeral lane would leave the 15 kept
booths — the durable, most-looked-at ones — with exactly the defect this
unit closes. Same lesson as the `blurtoggle` macro: three branches, one
definition; here it is two lanes and one rule."""
c, data = client
b = _booth(data, "r18-ab", kept=kept)
write_manifest(b, "booth-dev", why="pick the winning denoiser")
html = c.get("/").text
assert "booth-dev" in html
assert "pick the winning denoiser" in html
@pytest.mark.parametrize("kept", [False, True], ids=["ephemeral", "kept"])
def test_a_booth_that_never_spoke_up_is_marked(client, kept):
"""All 26 live booths are in this state, and rsync keeps making more. The
marker is what makes the convention adoptable at all: the link board rotted
to 69% precisely because nothing ever showed which rows were dead.
ASSERTED ON THE CLASS, not on the word, and the test is named around it.
`pytest`'s `tmp_path` is derived from the TEST NAME and the index renders
`data_dir` in its empty-state hint — so a test called
`test_an_unannounced_booth_says_so` put the literal string "unannounced"
into the page and passed against a template that did not yet exist. A
structural hook cannot be spelled by accident — though it has to be the
rendered ELEMENT and not the bare class, since base.html ships a
`.prov-none{...}` rule into the very same page."""
c, data = client
_booth(data, "quiet", kept=kept)
html = c.get("/").text
assert 'class="prov prov-none"' in html
assert "unannounced" in html
def test_a_damaged_manifest_reads_differently_from_an_absent_one(client):
"""INV-5 on the surface the operator looks at, not just in the reader."""
c, data = client
b = _booth(data, "damaged")
(b / MANIFEST_FILE).write_text("{oops")
html = c.get("/").text
assert 'class="prov prov-broken"' in html
assert "unreadable" in html
assert c.get("/b/damaged/").status_code == 200
def test_an_announced_booth_with_no_why_shows_only_its_handle(client):
"""`booth new x` with no --why is legal and common. The card shows who made
it and does not invent a purpose or leave a dangling separator."""
c, data = client
b = _booth(data, "scratch")
write_manifest(b, "booth-dev")
html = c.get("/").text
assert "booth-dev" in html
assert 'class="prov prov-none"' not in html
def test_the_booth_page_header_carries_it_too(client):
"""Deliberate scope, not creep: a booth URL handed to the operator lands
HERE, never on the index. Job 5 is 'operator, look at this', so the page he
actually opens is where the answer has to be."""
c, data = client
b = _booth(data, "r18-ab")
write_manifest(b, "booth-dev", why="pick the winning denoiser")
html = c.get("/b/r18-ab/").text
assert "booth-dev" in html
assert "pick the winning denoiser" in html
def test_a_poisoned_manifest_cannot_take_down_the_index(client):
"""The v0.2.2 lesson, asserted for the new reader before it can repeat:
`list_booths` touches every booth on every page load, so one bad file must
cost that booth's provenance and nothing else."""
c, data = client
_booth(data, "good")
bad = _booth(data, "bad")
(bad / MANIFEST_FILE).write_bytes(b"\xff\xfe not utf-8 at all")
assert c.get("/").status_code == 200
assert c.get("/healthz").status_code == 200
def test_a_pickup_booth_announces_itself_as_the_booths_own(client):
"""No exemption list. A booth the service made says the service made it,
which is true — and it keeps the rule to one line: a booth with no manifest
is unannounced."""
c, data = client
r = c.post("/upload", files=[("files", ("a.txt", b"hello", "text/plain"))],
follow_redirects=False)
assert r.status_code in (200, 303)
booth = next(p for p in data.iterdir() if p.is_dir())
got = read_manifest(booth)
assert got is not None and got.handle == "booth"
assert 'class="prov prov-none"' not in c.get("/").text
# ---- findings from the cross-frontier CODE-REVIEW panel, 2026-09-22 ----------
#
# Heid panel (thread 01M341E9XAPZEFBSPK9HPGAM0S). Four arms, artifact-only.
# The round found ZERO drift in the strict sense and landed its weight one layer
# down, in test strength: five of the ten adopted findings are tests of mine
# that pass on the regression they exist to catch.
def test_the_read_survives_a_document_no_one_can_parse(tmp_path):
"""INV-2 said "never raises" and named a 4 GB file as a tested case. It was
not tested, and it did not hold: `except ValueError` catches a truncated
document, but `json.loads` on deeply nested input raises RecursionError,
which is not a ValueError and is not an OSError either.
`list_booths` calls this once per booth on every index load, so the one
file costs the whole front page — the exact outage shape the invariant
cites as its reason for existing. Three of four arms reached it
independently; the eight-payload parametrize above has no size or depth
case, so the hole stayed green.
"""
b = tmp_path / "b"
b.mkdir()
(b / MANIFEST_FILE).write_text("[" * 200_000 + "]" * 200_000)
got = read_manifest(b)
assert isinstance(got, Manifest) and got.error
def test_the_read_refuses_a_document_too_large_to_be_a_manifest(tmp_path):
"""The other half of INV-2's named case. A manifest is four short fields;
anything approaching a megabyte is not one, and reading it into memory to
discover that is the wrong order of operations. Bounded BEFORE the read, so
the size is checked by `stat` rather than survived."""
from booth.manifest import MANIFEST_MAX_BYTES
b = tmp_path / "b"
b.mkdir()
(b / MANIFEST_FILE).write_text('{"handle": "x", "why": "' +
"y" * (MANIFEST_MAX_BYTES + 100) + '"}')
got = read_manifest(b)
assert isinstance(got, Manifest) and got.error
assert "too large" in got.error
def test_a_hostile_directory_name_does_not_reach_the_record_raw(tmp_path):
"""`_one_line(title, TITLE_MAX) or booth.name` — the FALLBACK skips the
normalization the explicit value gets. A directory name may legally carry a
newline on POSIX and may be 255 bytes, and either lands in a card's
sub-line. Same shape on the read path's fallback."""
# 200-odd bytes, under the filesystem's own 255 limit but well over
# TITLE_MAX — and a newline, which POSIX permits in a filename.
name = "we" + "i" * 200 + "rd\nname"
b = tmp_path / name
b.mkdir()
m = write_manifest(b, "booth-dev")
assert "\n" not in m.title and len(m.title) <= 120
assert "\n" not in read_manifest(b).title
def test_the_write_replaces_rather_than_truncating(tmp_path):
"""The previous version of this test asserted only that no `*.tmp` file
survived — which a plain `write_text` passes, since it leaves no temp file
either. All four arms said so, and they were right.
THE INODE IS THE DISCRIMINATOR. `os.replace` publishes a different file over
the old name, so the inode changes; truncate-and-rewrite keeps it. That is
also exactly why the promise holds for a concurrent reader: it either has
the old inode, intact, or opens the new one, complete. A test of the
mechanism rather than of its litter.
(An earlier draft spied on `os.open` to prove the published path was never
opened for writing. It passed — vacuously. `Path.write_text` reaches the
syscall through `io.open` in C and never touches the Python-level
`os.open`, so the spy could not have fired either way. Recorded because
writing a second vacuous test while fixing the first is the failure mode
this whole round is about.)
"""
b = tmp_path / "b"
b.mkdir()
published = b / MANIFEST_FILE
write_manifest(b, "booth-dev", why="first")
first_inode = published.stat().st_ino
write_manifest(b, "booth-dev", why="second")
assert published.stat().st_ino != first_inode, (
"the manifest was rewritten in place, not replaced"
)
assert read_manifest(b).why == "second"
def test_the_temp_file_is_not_a_name_two_writers_share(tmp_path):
"""Every writer derived the same `.booth.json.tmp`. Two `booth add` calls on
one booth could then interleave through a stale descriptor into the
published path — the atomic-write promise is that READERS never see a
partial file, and it says nothing about two writers sharing a scratch name.
Marks are protected from this by their flock; the manifest has none."""
b = tmp_path / "b"
b.mkdir()
seen = set()
for i in range(5):
write_manifest(b, "booth-dev", why=f"pass {i}")
seen.update(p.name for p in b.iterdir() if p.name != MANIFEST_FILE)
assert not seen, f"left temp files behind: {sorted(seen)}"
from booth.manifest import _temp_path
names = {_temp_path(b).name for _ in range(20)}
assert len(names) > 1, "every writer derives the same temp name"
def test_a_bare_re_announce_does_not_wipe_the_why(tmp_path):
"""THE WORKFLOW IS `new --why` THEN `add`. Omitted flags meant empty
strings, and empty strings overwrote — so the second command silently
erased the sentence the first one existed to record, on the single most
common sequence this feature has.
Two arms of the paraphrase panel predicted it from the contract's wording
alone ("gains a manifest with no why" does not distinguish a first write
from a re-announce with the flags omitted). Every test I wrote passed
`--why` on both calls, so none of them could see it.
Omitted now means UNCHANGED; only a value that was actually supplied
overwrites, and an explicit empty string still clears.
"""
b = tmp_path / "b"
b.mkdir()
write_manifest(b, "booth-dev", title="R18 A/B", why="pick the denoiser")
write_manifest(b, "booth-dev") # a bare `booth add`
kept = read_manifest(b)
assert kept.why == "pick the denoiser", "a bare re-announce wiped the why"
assert kept.title == "R18 A/B"
write_manifest(b, "booth-dev", why="sharper") # supplied: overwrites
assert read_manifest(b).why == "sharper"
write_manifest(b, "booth-dev", why="") # explicit: clears
assert read_manifest(b).why == ""
def test_re_announcing_preserves_a_created_from_before_this_second(tmp_path):
"""`_now()` is whole-second resolution, so two `write_manifest` calls in a
row share a timestamp and the old preservation test passed even against an
implementation that regenerated `created` every time. Three of four arms
caught it. Seed a stamp that could not have come from now()."""
b = tmp_path / "b"
b.mkdir()
(b / MANIFEST_FILE).write_text(json.dumps({
"handle": "booth-dev", "title": "b", "why": "first",
"created": "2019-03-04T11:22:33-08:00",
}))
assert write_manifest(b, "booth-dev", why="second").created == \
"2019-03-04T11:22:33-08:00"
def test_only_the_manifest_module_opens_the_manifest(tmp_path):
"""INV-1, which had no guard anywhere. One resolver is only one resolver
while nothing else learns the filename."""
root = pathlib.Path(__file__).parent.parent
offenders = []
for src in sorted((root / "booth").glob("*.py")):
if src.name == "manifest.py":
continue
tree = ast.parse(src.read_text())
# STRING CONSTANTS, not raw text. A comment naming the file is prose
# about the design and harms nothing — the first version of this test
# scanned the whole source and went red on a comment explaining why a
# leaked `.booth.json.<hex>.tmp` keeps a booth alive. The invariant is
# about code that knows the filename, so ask the code.
docstrings = set()
for node in ast.walk(tree):
if isinstance(node, (ast.Module, ast.ClassDef,
ast.FunctionDef, ast.AsyncFunctionDef)):
body = getattr(node, "body", None)
if body and isinstance(body[0], ast.Expr) and \
isinstance(body[0].value, ast.Constant):
docstrings.add(id(body[0].value))
for node in ast.walk(tree):
if (isinstance(node, ast.Constant) and isinstance(node.value, str)
and id(node) not in docstrings and ".booth.json" in node.value):
offenders.append(f"{src.name}:{node.lineno}")
assert not offenders, f"{offenders} name the manifest file in code"
def test_announcing_is_activity_via_the_manifest_file_itself(tmp_path):
"""The previous version could not fail. Writing the manifest creates a
directory entry, which bumps the DIRECTORY's mtime, so the booth read as
fresh whether or not `_newest_mtime` counted the manifest at all — a test
of the side effect rather than of the thing.
Put the directory's clock back afterwards, leaving the manifest's own mtime
as the only thing that can keep the booth alive."""
import os
from booth.app import booth_age_seconds
b = tmp_path / "b"
b.mkdir()
old = 1_000_000_000
os.utime(b, (old, old))
write_manifest(b, "booth-dev", why="look at this")
os.utime(b, (old, old)) # only the file can save it now
assert booth_age_seconds(b, now=old + 90_000) < 86_400
def test_the_booth_header_marks_an_unannounced_booth_too(client):
"""The negative states were asserted on `/` only, so a header that rendered
provenance for clean manifests and nothing for the other two would have
passed the whole suite."""
c, data = client
_booth(data, "quiet")
damaged = _booth(data, "damaged")
(damaged / MANIFEST_FILE).write_text("{oops")
assert 'class="prov prov-none"' in c.get("/b/quiet/").text
assert 'class="prov prov-broken"' in c.get("/b/damaged/").text
def test_the_title_reaches_a_surface(client):
"""`--title` promised a display name and nothing rendered it — 4/4 on the
paraphrase panel, independently the top-ranked flag of that round. It lands
on the booth page heading, where there is room for it; the INDEX card keeps
the directory name, because that is the identity the operator navigates and
refers to positionally."""
c, data = client
b = _booth(data, "r18-ab")
write_manifest(b, "booth-dev", title="R18 A/B — denoiser bakeoff", why="w")
page = c.get("/b/r18-ab/").text
assert "R18 A/B — denoiser bakeoff" in page
assert "r18-ab" in page, "the directory name stopped being visible"
# ---- findings from the cross-frontier BUG-HUNT panel, 2026-09-22 -------------
#
# Heid panel (thread 01M343SXX27Z47C3STXXRC7M42). Four arms, artifact-only,
# diff-scoped. The strongest finding is one the SIZE CAP ITSELF opened.
def test_a_reader_never_blocks_on_a_file_that_is_not_a_file(tmp_path):
"""`stat` reports size 0 for a FIFO, so it sails under the byte cap — and
then `read_text` blocks in `read` with no EOF, so the `except` never runs
and the call never returns. `list_booths` reads every booth on every `GET /`
and `/healthz`, so ONE such file stalls the front page for the whole service,
with no error and no recovery short of a restart.
A symlink to `/dev/zero` is the same hole with unbounded allocation instead
of a hang: `st_size` is 0 there too.
Two of four arms reached it independently. The bound added an hour earlier
is what made it reachable — `st_size` answers a different question than
"can this be read", and a cap that trusts it inherits the difference.
"""
import os
import signal
b = tmp_path / "b"
b.mkdir()
os.mkfifo(b / MANIFEST_FILE)
# ⚠ ALARMED. Without this the RED state of this test does not fail, it HANGS
# — which is the defect itself, and is also useless as a signal: a suite that
# stops is indistinguishable from a suite that is slow. Five seconds is a
# thousand times the budget a read of a four-field file should need.
def _timeout(signum, frame):
raise AssertionError("read_manifest blocked on a FIFO and never returned")
old_handler = signal.signal(signal.SIGALRM, _timeout)
signal.alarm(5)
try:
got = read_manifest(b)
finally:
signal.alarm(0)
signal.signal(signal.SIGALRM, old_handler)
assert isinstance(got, Manifest) and got.error
assert "regular file" in got.error
def test_a_damaged_manifest_is_kept_when_it_is_replaced(tmp_path):
"""4/4, and it contradicted this repo's own doctrine. Marks made the rule
explicit in v0.2.1 — reads stay lenient, writes go strict, damaged bytes
STAY ON DISK — and the manifest's write replaced them outright.
The sharpest leg: a file that fails on ONE field still holds the others.
`{"handle": 7, "why": "the thing I wanted you to look at"}` reads as broken
and used to be destroyed whole, taking a `why` the re-announcer may not have
kept anywhere.
Quarantined rather than refused: refusing would fail `booth add` and lose
the files it was copying, which is the worse trade. One fixed-name
quarantine, so this cannot accumulate.
"""
from booth.manifest import QUARANTINE_FILE
b = tmp_path / "b"
b.mkdir()
damaged = json.dumps({"handle": 7, "why": "the thing I wanted you to see"})
(b / MANIFEST_FILE).write_text(damaged)
write_manifest(b, "booth-dev", why="rescued")
assert read_manifest(b).why == "rescued"
assert (b / QUARANTINE_FILE).read_text() == damaged, "the damaged bytes were destroyed"
def test_a_broken_record_normalizes_the_directory_name_too(tmp_path):
"""The third fallback. `write_manifest`'s and `read_manifest`'s were fixed
in the previous round and `_broken`'s was missed — same raw `booth.name`,
same card sub-line, same newline."""
b = tmp_path / ("wei" + "i" * 200 + "rd\nname")
b.mkdir()
(b / MANIFEST_FILE).write_text("{oops")
got = read_manifest(b)
assert got.error and "\n" not in got.title and len(got.title) <= 120
def test_an_identical_re_announce_does_not_touch_the_booth(tmp_path):
"""Marks learned this in v0.2.0: a write that changes nothing is not
activity and must not reset a booth's TTL. The manifest wrote
unconditionally, so `booth add` on an unchanged booth kept a dead one alive
— and `booth link` does it on every single post to the standing board."""
import os
b = tmp_path / "b"
b.mkdir()
write_manifest(b, "booth-dev", why="x")
path = b / MANIFEST_FILE
os.utime(path, (1_000_000_000, 1_000_000_000))
os.utime(b, (1_000_000_000, 1_000_000_000))
before = path.stat().st_mtime
write_manifest(b, "booth-dev", why="x") # identical
assert path.stat().st_mtime == before, "an identical re-announce rewrote the file"
def test_a_failed_write_leaves_no_temp_file_behind(tmp_path):
"""The unique temp name fixed a cross-writer hazard and created a litter
one: a fixed name is overwritten by the next writer, a random one is not.
And `.booth.json.<hex>.tmp` is NOT a `.lock`, so `_newest_mtime` counts it —
an orphaned temp would keep a dead booth alive forever."""
import os
b = tmp_path / "b"
b.mkdir()
real_replace = os.replace
def boom(src, dst, *a, **kw):
raise OSError("no space left on device")
os.replace = boom
try:
with pytest.raises(OSError):
write_manifest(b, "booth-dev", why="x")
finally:
os.replace = real_replace
assert not list(b.glob("*.tmp")), f"orphaned temp: {list(b.glob('*.tmp'))}"