Files
vh e702be4e1a fix: a wrong-shaped answer no longer 500s the gallery and the marks page
Pre-existing, measured at 42ea67f, so it predates U3. `_hydrate` checked only
that `answer` was a dict and never that `answer["answers"]` was one, so
`marks_for` and `hold_read` both reported the mark healthy with no read error
-- and `_ask_inline.html` then asked a list for `.get`. The v0.2.2 lesson was
half-implemented: that outage was a file that could not be PARSED and the
reader was made lenient, while this one parses perfectly and breaks one layer
further in, at render, where no leniency existed.

Closed at the hydration boundary rather than by a third copy of the guard --
one predicate, one place, every surface inherits it. Only the multi case is
checked, because only the multi case indexes; requiring `answers`
unconditionally would break every single-question pick, and that direction has
its own test. Measured before and after: gallery and marks pages 500 -> 200,
the error visible on the page, the booth's other healthy pick untouched.

The placement was the one open operator question of the session. It was
surfaced three times without a ruling, so it is taken under a stated assumption
and is cheap to move: the whole fix is one condition in one function.

Two things fell out of it worth more than the fix.

`_safe_fragments` no longer has a reachable natural trigger. Probed every wrong
answer shape a .marks.json can carry: `answers` as a list, a string or null all
become hydration errors now, and a wrong-typed value INSIDE `answers` renders
without raising, because Jinja absorbs attribute access on a non-mapping. U3's
guard is a pure backstop, and its test now says so and trips it synthetically
through the shared macro module rather than asserting a path nothing reaches.
A guard tested by an unreachable input is an untested guard.

And that guard's handler could not survive the failure it was handling: it
caught a raising `_pick_fragments` and rebuilt the broken-ask box through the
SAME macro module that had just raised, so whenever `whole` was the broken
thing it re-raised and took the whole report. Found by accident while building
the falsifier. Fixed, with its own test.

Both new falsifiers were verified RED against their defeating change rather
than assumed.

607 -> 611 tests.
2026-09-22 14:34:28 -07:00

528 lines
23 KiB
Python

"""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:
<script src="/_booth/embed.js" defer></script>
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'<!doctype html><title>r</title><body>hi<script src="{EMBED_SRC}" defer></script></body>'
@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:<target>` 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 = "<!doctype html><meta charset=utf-8><title>r</title><h1>REPORT</h1>"
(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 + '<script src="/_booth/embed.js" defer></script>'
assert out == src + EMBED_SCRIPT_TAG
assert len(out) > len(src)
assert out.lower().lstrip().startswith("<!doctype")
assert out.index("charset") < 1024
def test_a_booth_with_no_marks_still_gets_the_seam(client):
"""The seam carries the way home and the icon too, so it is not conditional
on there being an ask — the old chip was not either."""
c, data = client
(data / "b").mkdir()
(data / "b" / "index.html").write_text("<h1>bare fragment</h1>")
assert c.get("/b/b/").text == "<h1>bare fragment</h1>" + 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'<!doctype html>\r\n<title>r</title>\r\n<body>caf\xe9 \xff\r\n'
b'<script src="/_booth/embed.js" defer></script>\r\n</body>')
(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'<!doctype html>\r\n<title>r</title>\r\n<body>caf\xe9 \xff\r\n</body>'
(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, "<module level>"), 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 (
"<!doctype html><body><p>add <code>/_booth/embed.js</code> to your report</p></body>",
"<!doctype html><body><!-- src=/_booth/embed.js --></body>",
'<!doctype html><body><script src="/_booth/embed.js?v=2"></script></body>',
):
(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'<script src="{EMBED_SRC}" defer></script>',
f"<script src='{EMBED_SRC}' defer></script>"):
body = f"<!doctype html><body>hi{decl}</body>"
(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("<h1>huge</h1>")
monkeypatch.setattr(app_mod, "WRAP_MAX_BYTES", 4)
assert c.get("/b/b/").text == "<h1>huge</h1>"