"""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..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..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'))}"