"""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" and body["home"] == "/" 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_created_then_id(client): c, data = client b = _pick(data / "b", "zebra") _pick(b, "alpha") # force identical creation stamps: the id is the tie-break, not json order 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) served = c.get(EMBED_SRC).text try: src.write_text("/* POISONED */\n") assert c.get(EMBED_SRC).text == served, "embed.js is being re-read per request" 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 assert out == src + EMBED_SCRIPT_TAG assert out.startswith(src) assert out.lower().lstrip().startswith("bare fragment") assert c.get("/b/b/").text == "

bare fragment

" + EMBED_SCRIPT_TAG 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

"