"""U3 — the declared embed seam, server side. The Booth used to reach into a verbatim report with ten regular expressions: six to find somewhere to hang a favicon and a chip, four to substitute rendered ask markup into the author's own tags. This unit replaces all of it with a seam the page declares: What is tested here is the SERVER half — the payload that crosses the seam, the one static asset, and the single conditional append that is now the only thing the Booth does to author HTML. The half that mounts fragments into a live DOM lives in tests/test_embed_browser.py, because no amount of string assertion can see whether a form actually submits. Contract: docs/contracts/u3_declared_embed_seam.contract.md """ import json import os import time import pytest from fastapi.testclient import TestClient from booth.app import EMBED_SCRIPT_TAG, EMBED_SRC, create_app from booth.marks import answer_pick, declare_pick, set_flag, write_note DECLARED = f'rhi' @pytest.fixture def client(tmp_path): app = create_app(tmp_path, ttl_hours=24, start_sweeper=False) return TestClient(app), tmp_path def _pick(booth, stem="winner", **kw): doc = {"prompt": "Which render wins?", "options": ["A — baseline", "B — async"]} doc.update(kw) booth.mkdir(parents=True, exist_ok=True) declare_pick(booth, stem, doc) return booth def _multi(booth, stem="batch"): booth.mkdir(parents=True, exist_ok=True) declare_pick(booth, stem, {"title": "Round one", "questions": [ {"key": "r1", "prompt": "First?", "options": ["a", "b"]}, {"key": "r2", "prompt": "Second?", "options": ["a", "b"]}, ]}) return booth # ---- slice 1: the payload ---------------------------------------------------- def test_embed_payload_carries_a_fragment_for_every_shape(client): c, data = client _multi(data / "b") body = c.get("/b/b/embed.json").json() assert body["booth"] == "b" assert "home" not in body, "a value nothing reads is a second copy waiting to drift" assert body["favicon"].startswith("data:image/svg+xml,") (m,) = body["marks"] assert m["id"] == "batch" and m["error"] is None assert "First?" in m["whole"] and "Second?" in m["whole"] assert 'action="/b/b/answer"' in m["submit"] assert [q["key"] for q in m["questions"]] == ["r1", "r2"] # declaration order assert "First?" in m["questions"][0]["html"] assert 'type="radio"' in m["questions"][0]["html"] def test_a_single_question_pick_has_one_question_with_a_null_key(client): """SR-2. `normalize_ask` gives a single-question ask `key: None`, so the payload cannot key questions by name — JSON would write that as "null" and invent a name. Every one-question ask in the fleet hits this.""" c, data = client _pick(data / "b") (m,) = c.get("/b/b/embed.json").json()["marks"] assert [q["key"] for q in m["questions"]] == [None] assert "Which render wins?" in m["questions"][0]["html"] def test_marks_are_ordered_by_creation_before_id(client): """`created` LEADS. The fixture makes creation order and alphabetical order disagree, because a fixture where they agree cannot tell the stated rule from a plain id sort — which is what the first version of this test did, and what a panel caught by reading the fixture rather than the assertion.""" c, data = client b = _pick(data / "b", "zebra") _pick(b, "alpha") raw = json.loads((b / ".marks.json").read_text()) stamps = {"zebra": "2026-09-22T10:00:00.000000-07:00", # first "alpha": "2026-09-22T11:00:00.000000-07:00"} # second for e in raw["marks"]: e["created"] = stamps[e["id"]] (b / ".marks.json").write_text(json.dumps(raw)) assert [m["id"] for m in c.get("/b/b/embed.json").json()["marks"]] == ["zebra", "alpha"] def test_the_id_is_only_the_tie_break(client): """And with `created` equal, the id decides — so two marks written in the same second cannot swap between renders.""" c, data = client b = _pick(data / "b", "zebra") _pick(b, "alpha") raw = json.loads((b / ".marks.json").read_text()) for e in raw["marks"]: e["created"] = "2026-09-22T10:00:00.000000-07:00" (b / ".marks.json").write_text(json.dumps(raw)) assert [m["id"] for m in c.get("/b/b/embed.json").json()["marks"]] == ["alpha", "zebra"] def test_open_is_computed_by_the_server_not_the_page(client): """INV-4. The chip count follows `open_marks`, which is the ONE openness predicate — a half-answered multi-question pick is still open.""" c, data = client b = _multi(data / "b") assert c.get("/b/b/embed.json").json()["open"] == ["batch"] answer_pick(b, "batch", {"r1": "a"}) assert c.get("/b/b/embed.json").json()["open"] == ["batch"] # partial is OPEN answer_pick(b, "batch", {"r1": "a", "r2": "b"}) assert c.get("/b/b/embed.json").json()["open"] == [] def test_the_payload_carries_picks_only(client): """SR-4. Notes and flags never reach it, which is also what keeps a flag's `flag:` id — the one mark id containing the anchor separator — out of a payload whose specs split on the first colon.""" c, data = client b = _pick(data / "b") (b / "shot.png").write_bytes(b"x") write_note(b, None, "a remark") set_flag(b, "shot.png", True) assert [m["id"] for m in c.get("/b/b/embed.json").json()["marks"]] == ["winner"] def test_a_damaged_marks_file_does_not_500_the_report(client): c, data = client b = _pick(data / "b") (b / ".marks.json").write_text("{not json") r = c.get("/b/b/embed.json") assert r.status_code == 200 assert r.json()["marks"] == [] and r.json()["error"] def test_a_broken_pick_offers_whole_and_nothing_else(client): c, data = client b = _pick(data / "b") raw = json.loads((b / ".marks.json").read_text()) raw["marks"][0]["declaration"] = {"prompt": "p"} # no options -> AskError (b / ".marks.json").write_text(json.dumps(raw)) (m,) = c.get("/b/b/embed.json").json()["marks"] assert m["error"] and m["questions"] == [] and m["submit"] == "" assert "broken ask" in m["whole"] def test_the_payload_does_not_record_a_view(client): """`booth_view` already recorded the look, above both of its early returns. A script's fetch of the page it is already on must not count a second time or reset the TTL on machinery instead of on the operator.""" c, data = client b = _pick(data / "b") (b / "index.html").write_text(DECLARED) c.get("/b/b/") before = os.stat(b / ".viewed").st_mtime_ns time.sleep(0.01) c.get("/b/b/embed.json") assert os.stat(b / ".viewed").st_mtime_ns == before def test_embed_payload_404s_for_an_unknown_booth(client): c, _ = client assert c.get("/b/nope/embed.json").status_code == 404 # ---- slice 2: the one static asset ------------------------------------------- def test_embed_js_is_served_as_javascript(client): c, _ = client r = c.get(EMBED_SRC) assert r.status_code == 200 assert r.headers["content-type"].startswith("text/javascript") assert "data-booth-mark" in r.text def test_embed_js_does_not_hot_reload_from_disk(tmp_path): """INV-5, and the 2026-09-21 lesson restated. A live asset editable under a running process is how 19 of 25 booths hit 500 with the Python from 22:03 and the templates from 23:40. One rule in this repo: nothing takes effect until you restart.""" import pathlib import booth.app as app_mod src = pathlib.Path(app_mod.__file__).parent / "static" / "embed.js" original = src.read_text() app = create_app(tmp_path, ttl_hours=24, start_sweeper=False) c = TestClient(app) try: # Poisoned BEFORE the first request, not between two of them. The # earlier shape passed for a route that read the file lazily and cached # on first use — which is not "read once at startup", and is exactly the # staleness this invariant exists to forbid. src.write_text("/* POISONED */\n") first = c.get(EMBED_SRC).text assert "POISONED" not in first, "embed.js is read at request time, not at startup" assert first == original assert c.get(EMBED_SRC).text == first finally: src.write_text(original) # ---- slice 3: what the Booth does to author HTML ----------------------------- def test_declaring_page_is_served_untouched(client): """INV-1. Whole-body equality, not a substring absence: the promise is that NOTHING is added, and an absence assertion cannot tell a clean page from one carrying something nobody thought to look for.""" c, data = client b = _pick(data / "b") (b / "index.html").write_text(DECLARED) assert c.get("/b/b/").text == DECLARED def test_undeclared_page_gains_only_the_tag(client): """INV-2. Appended, so the source is a strict prefix — nothing is inserted, nothing is prepended, and neither the doctype nor the charset window moves.""" c, data = client b = _pick(data / "b") src = "r

REPORT

" (b / "index.html").write_text(src) out = c.get("/b/b/").text # The literal, not just the constant: `out == src + EMBED_SCRIPT_TAG` also # holds when EMBED_SCRIPT_TAG is the empty string, which is a mutation this # test exists to catch. A panel found it by reading the assertion, not the # code. assert out == src + '' assert out == src + EMBED_SCRIPT_TAG assert len(out) > len(src) assert out.lower().lstrip().startswith("bare fragment") assert c.get("/b/b/").text == "

bare fragment

" + EMBED_SCRIPT_TAG def test_a_declaring_page_is_served_BYTE_for_byte(client): """INV-1, at the level the promise is actually made. The first version read the file with `read_text()`, which opens in universal-newline mode: a report written with CRLF came back with LF, and `errors="replace"` turned any non-UTF-8 byte into U+FFFD. A declaring page was NOT served as its author wrote it — the headline promise — and the original test could not see it, because its fixture was LF-only ASCII. Found by a cross-frontier bug-hunt panel. """ c, data = client b = _pick(data / "b") src = (b'\r\nr\r\ncaf\xe9 \xff\r\n' b'\r\n') (b / "index.html").write_bytes(src) r = c.get("/b/b/") assert r.content == src, "the operator's document was edited on the way out" assert b"\r\n" in r.content and b"\xff" in r.content def test_an_undeclared_page_keeps_every_byte_and_gains_the_tag(client): """INV-2, same level: the source is a BYTE-exact prefix of the response.""" c, data = client b = _pick(data / "b") src = b'\r\nr\r\ncaf\xe9 \xff\r\n' (b / "index.html").write_bytes(src) r = c.get("/b/b/") assert r.content == src + EMBED_SCRIPT_TAG.encode("utf-8") assert r.content.startswith(src) def test_a_wrongly_shaped_answer_costs_its_pick_not_the_report(client): """`marks_for` hydrates `{"answer": {"answers": []}}` with no error — the JSON is well formed, the SHAPE is not — and the template then asks a list for `.get`. This endpoint renders every pick on every load of the operator's report, so an unguarded raise here is the whole seam gone while `hold_read` calls the file perfectly readable. Verified reachable, not assumed.""" c, data = client b = data / "b" b.mkdir(parents=True, exist_ok=True) declare_pick(b, "batch", {"title": "T", "questions": [ {"key": "r1", "prompt": "A?", "options": ["x", "y"]}, {"key": "r2", "prompt": "B?", "options": ["x", "y"]}]}) _pick(b, "healthy") raw = json.loads((b / ".marks.json").read_text()) for e in raw["marks"]: if e["id"] == "batch": e["answer"] = {"answers": [], "notes": ""} (b / ".marks.json").write_text(json.dumps(raw)) (b / "index.html").write_text(DECLARED) r = c.get("/b/b/embed.json") assert r.status_code == 200 by = {m["id"]: m for m in r.json()["marks"]} # THE PROMISE, NOT THE LAYER. This used to pin the string # `_safe_fragments` produces ("could not be rendered"), which made the test # an assertion about WHICH guard fired. As of the `_hydrate` answer-shape # check, this input is caught one layer earlier and never reaches # `_pick_fragments` at all — the endpoint's promise is unchanged and the # error is better (it names what is wrong with the stored answer instead of # reporting a render failure), so the assertion moved to the promise. # `_safe_fragments` is still the backstop and is still falsified, by # `test_safe_fragments_still_catches_what_hydration_cannot` below. assert by["batch"]["error"], "a wrong-shaped answer reported no error" assert "broken ask" in by["batch"]["whole"] # and the booth's other pick is untouched — one bad entry costs one entry assert by["healthy"]["error"] is None assert "Which render wins?" in by["healthy"]["whole"] assert c.get("/b/b/").status_code == 200 # ⚠ THE GALLERY AND MARKS PAGES USED TO 500 ON THIS ENTRY, and that was NOT # U3's doing — measured at 42ea67f, the commit before that unit. CLOSED # 2026-09-22 at the hydration boundary rather than by a third copy of this # guard: see tests/test_marks.py # ::test_a_wrong_shaped_answer_is_an_error_at_hydration_not_a_500 and # persistent-memory.d/2026-09-22-a-wrong-shaped-answer-500s-the-gallery.md def test_safe_fragments_still_catches_what_hydration_cannot(client): """U3's `_safe_fragments` guard, kept falsifiable after `_hydrate` took its natural trigger away. The answer-shape check in `_hydrate` now catches every wrong answer shape reachable from a `.marks.json` — probed 2026-09-22: `answers` as a list, a string or null all become hydration errors, and a wrong-typed VALUE inside `answers` renders without raising, because Jinja absorbs attribute access on a non-mapping. **No natural input reaches `_safe_fragments` by this route any more**, and a test that kept pretending one did would assert nothing — which is the failure this suite has now paid for twice. So the trigger is synthetic and says so: the shared `_ask_inline` macro module is made to raise. `_pick_fragments` resolves `whole` off that object per call, and `create_app` stashes the environment on `app.state`, so this reaches the very object the closure captured. What it pins is the guard itself — one raising pick costs that pick, never the report. Defeating change: removing the try/except in `_safe_fragments`, under which this returns 500. """ c, data = client b = data / "b" b.mkdir(parents=True, exist_ok=True) declare_pick(b, "batch", {"prompt": "Which?", "options": ["x", "y"]}) (b / "index.html").write_text(DECLARED) frag = c.app.state.templates.env.get_template("_ask_inline.html").module real_submit = frag.submit def explode(*a, **k): raise RuntimeError("synthetic render failure") object.__setattr__(frag, "submit", explode) try: assert frag.submit is explode, "the patch did not take; this test is vacuous" r = c.get("/b/b/embed.json") assert r.status_code == 200, "a raising fragment renderer took the whole report" by = {m["id"]: m for m in r.json()["marks"]} assert by["batch"]["error"] and "could not be rendered" in by["batch"]["error"] assert by["batch"]["whole"], "the fallback rendered nothing at all" finally: object.__setattr__(frag, "submit", real_submit) # and the guard is not sticky — with the macro restored, the pick is fine assert c.get("/b/b/embed.json").json()["marks"][0]["error"] is None def test_the_handler_survives_the_failure_it_is_handling(client): """`_safe_fragments` caught a raising `_pick_fragments` and then rebuilt the broken-ask box THROUGH THE SAME MACRO MODULE that had just raised. So when `whole` itself was the broken thing, the handler re-raised and took the whole report — a guard that only worked when the failure was somewhere else. Found by accident: the first draft of the falsifier above patched `whole`, and the guard failed rather than caught. Defeating change: removing the inner try/except, under which this returns 500.""" c, data = client b = data / "b" b.mkdir(parents=True, exist_ok=True) declare_pick(b, "batch", {"prompt": "Which?", "options": ["x", "y"]}) (b / "index.html").write_text(DECLARED) frag = c.app.state.templates.env.get_template("_ask_inline.html").module real_whole = frag.whole def explode(*a, **k): raise RuntimeError("even the fallback macro is broken") object.__setattr__(frag, "whole", explode) try: r = c.get("/b/b/embed.json") assert r.status_code == 200, "the handler re-raised through the broken macro" assert r.json()["marks"][0]["error"] finally: object.__setattr__(frag, "whole", real_whole) def test_no_regex_touches_author_html(): """INV-3, and the version that actually falsifies it. The first draft of this test name-matched the six deleted patterns. A cold panel pointed out — correctly, and on the contract's CENTRAL promise — that reintroducing the same regex under a new name (`_TAIL_RE`, applied in the verbatim branch) would leave it green. A test that guards names does not guard behaviour, and this repo's own vacuity pass missed it because the mutation it tried was the named one. So: `booth/app.py` is allowed EXACTLY ONE regex operation, and it is `ask_form_id`'s `re.sub` over a mark id — not over a page. Any other regex anywhere in the module fails here, whatever it is called. If a future change genuinely needs one, the failure is the conversation: say which string it reads and why it is not author HTML. """ import ast import pathlib import booth.app as app_mod root = pathlib.Path(app_mod.__file__).parent assert not (root / "inline.py").exists(), "booth/inline.py survived U3" tree = ast.parse((root / "app.py").read_text()) # every node -> the function it sits in, so a finding names its site site = {} for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): for child in ast.walk(node): site.setdefault(child, node.name) # `re` reaches this module ONE way: a plain module-level `import re`. An # alias (`import re as _r`) or a direct name import (`from re import sub`) # would route around the call check below under a name it does not know — # found by re-running the vacuity pass against the FIXED test, which is the # only reason it is here and is the argument for running that pass on a fix # and not only on a draft. for node in ast.walk(tree): if isinstance(node, ast.Import): for a in node.names: assert not (a.name == "re" and a.asname), f"`re` aliased as {a.asname}" elif isinstance(node, ast.ImportFrom): assert node.module != "re", f"names imported from re: {[a.name for a in node.names]}" METHODS = {"search", "sub", "subn", "match", "fullmatch", "finditer", "findall", "split", "compile", "escape"} found = [] for node in ast.walk(tree): if not (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)): continue f = node.func on_re = isinstance(f.value, ast.Name) and f.value.id == "re" on_pattern = (isinstance(f.value, ast.Name) and f.value.id.endswith("_RE") and f.attr in METHODS) if on_re or on_pattern: found.append((site.get(node, ""), f.attr)) assert found == [("ask_form_id", "sub")], ( f"booth/app.py performs regex operations outside ask_form_id: {found}" ) # the six named patterns and the chips are gone, and stay gone assigned = { t.id for node in ast.walk(tree) if isinstance(node, ast.Assign) for t in node.targets if isinstance(t, ast.Name) } gone = {"_ICON_RE", "_HEAD_CLOSE_RE", "_HTML_OPEN_RE", "_DOCTYPE_RE", "_BODY_CLOSE_RE", "_HTML_CLOSE_RE", "_BACK_CHIP", "FAVICON_LINK"} assert not (assigned & gone), f"deleted names are back: {sorted(assigned & gone)}" funcs = {n.name for n in ast.walk(tree) if isinstance(n, ast.FunctionDef)} assert not ({"wrap_verbatim_html", "asks_chip", "inject_asks", "_insert_before", "_insert_after"} & funcs) def test_a_page_that_only_mentions_the_path_is_not_declaring_it(client): """A report that QUOTES the seam — a code sample, a comment, a sentence about this very feature — is not declaring it, and the Booth's own design reports are the pages most likely to do that. Read as declared, such a page would be served untouched and show no chrome at all, silently. The detection therefore fails the other way: an unrecognised spelling gets a duplicate tag, and embed.js mounts once regardless. """ c, data = client b = _pick(data / "b") for body in ( "

add /_booth/embed.js to your report

", "", '', ): (b / "index.html").write_text(body) assert c.get("/b/b/").text == body + EMBED_SCRIPT_TAG, body # and the real declaration, in either quote style, is honoured for decl in (f'', f""): body = f"hi{decl}" (b / "index.html").write_text(body) assert c.get("/b/b/").text == body def test_an_oversize_verbatim_page_is_served_raw(client, monkeypatch): """WRAP_MAX_BYTES survives: a pathological file is still not pulled into memory, and it loses its chrome exactly as it does today.""" import booth.app as app_mod c, data = client b = _pick(data / "b") (b / "index.html").write_text("

huge

") monkeypatch.setattr(app_mod, "WRAP_MAX_BYTES", 4) assert c.get("/b/b/").text == "

huge

"