"""U6 — benches: a running thing, registered. The contract is docs/contracts/u6_benches.contract.md. Every test here names the invariant it falsifies, and each is written to go RED under the change that defeats that invariant — not merely to assert the outcome the author had in mind. (The U4 round shipped seven falsifiers of which five stayed green under the very change they forbade; see persistent-memory.d/2026-09-22-vacuous-falsifiers.md.) """ from __future__ import annotations import ast import json import os import pathlib import sys import pytest sys.path.insert(0, str(pathlib.Path(__file__).parent.parent)) from booth.benches import ( # noqa: E402 BENCHES_FILE, BENCH_STATES, Bench, normalize_bench_url, order_benches, read_benches, remove_bench, set_bench_state, upsert_bench, ) from booth.links import booth_target # noqa: E402 # ---- INV-6: the identity collapses a re-post and NOTHING else --------------- # # Both directions from ONE fixture. A test that only checked the talk collapse # would pass under origin normalization, which is the measurably wrong rule: # on the live board it merges eight distinct gitea repositories into one row. # Measured on the live board, 2026-09-22. GITEA_EIGHT = [ "https://gitea.phasefinal.com/vh/bifrost/issues/17", "https://gitea.phasefinal.com/vh/brokkr-smithy/src/commit/6adcde6/research/landscape-scans/open-weight-releases-2026-09-15.md", "https://gitea.phasefinal.com/vh/cicada", "https://gitea.phasefinal.com/vh/draupnir", "https://gitea.phasefinal.com/vh/-/packages/pypi/bifrost/1.2.0", "https://gitea.phasefinal.com/vh/-/packages/pypi/bifrost/1.2.1", "https://gitea.phasefinal.com/vh/peedlar", "https://gitea.phasefinal.com/vh/peedlar/releases/tag/v0.3.0", ] TALK_FIVE = ["https://talk.nh3.phasefinal.com:8092/"] * 5 def test_eight_distinct_repos_stay_eight(tmp_path): """INV-6, the direction origin-normalization gets WRONG. Defeating change: normalizing to scheme://host:port. This goes red under it; the collapse test below does not.""" for i, u in enumerate(GITEA_EIGHT): upsert_bench(tmp_path, u, f"repo {i}", "vh") benches, err = read_benches(tmp_path) assert err is None assert len(benches) == 8, [b.id for b in benches] def test_five_reposts_of_one_bench_collapse(tmp_path): """INV-6, the direction the IA doc names. `talk` is on the live board five times; the registry must hold one row, carrying the LAST name.""" for i, u in enumerate(TALK_FIVE): _, created = upsert_bench(tmp_path, u, f"talk v{i}", "nh3-dev") assert created is (i == 0) benches, _ = read_benches(tmp_path) assert len(benches) == 1 assert benches[0].name == "talk v4" def test_two_lrpg_surfaces_on_one_origin_stay_two(tmp_path): """INV-6. The IA doc's OWN example of two real benches shares an origin.""" upsert_bench(tmp_path, "http://10.100.10.50:8321/Authoring%20Studio.dc.html", "authoring", "ldp-dev") upsert_bench(tmp_path, "http://10.100.10.50:8321/GM%20Playback.dc.html", "gm", "ldp-dev") assert len(read_benches(tmp_path)[0]) == 2 def test_query_is_part_of_the_identity(tmp_path): """INV-6. Three ShutterChute rows differ ONLY by `?token=`; they are three links, not one bench posted three times. Defeating change: dropping query.""" base = "http://10.100.10.50:8477/?token=" for tok in ("aaa", "bbb", "ccc"): upsert_bench(tmp_path, base + tok, "shutterchute", "nh3-dev") assert len(read_benches(tmp_path)[0]) == 3 @pytest.mark.parametrize("a,b", [ ("http://x.test/", "http://X.TEST"), # host case + bare-slash path ("http://x.test:80/p", "http://x.test/p"), # default port ("https://x.test:443/p", "https://x.test/p"), ("http://x.test/p#frag", "http://x.test/p"), # fragment dropped (" http://x.test/p ", "http://x.test/p"), # whitespace ]) def test_these_pairs_are_one_bench(a, b): """INV-6. Each pair is the SAME resource reached two ways.""" assert normalize_bench_url(a) == normalize_bench_url(b) @pytest.mark.parametrize("a,b", [ ("http://x.test/p", "http://x.test/p/"), # trailing slash on a REAL path ("http://x.test/p", "http://x.test/P"), # path case ("http://x.test/?a=1&b=2", "http://x.test/?b=2&a=1"), # query order is opaque ("http://x.test:8092/", "https://x.test:8092/"), # scheme ]) def test_these_pairs_are_two_benches(a, b): """INV-6, the other direction. Each pair MAY be two different resources, and the registry must not decide otherwise on the operator's behalf.""" assert normalize_bench_url(a) != normalize_bench_url(b) @pytest.mark.parametrize("bad", [ "", " ", "not a url", "ftp://x.test/f", "file:///etc/passwd", "http://", "https:///path", "//x.test/p", "javascript:alert(1)", ]) def test_refused_urls_raise_with_a_reason(bad): with pytest.raises(ValueError) as e: normalize_bench_url(bad) assert str(e.value).strip(), "a refusal with no reason is a refusal the CLI cannot print" def test_credentials_are_refused_not_stripped(): """Stripping would register a bench whose URL no longer works while telling the poster it succeeded — and put a credential on an unauthenticated LAN surface on the way. Defeating change: `netloc.rpartition('@')[2]`.""" with pytest.raises(ValueError): normalize_bench_url("https://user:hunter2@x.test/p") # ---- INV-7: `url` is what a click goes to; `id` is never the href ----------- def test_the_stored_url_is_the_raw_string(tmp_path): """INV-7. Defeating change: storing the normalized form as `url` because it is 'the clean one'. Every field that differs is asserted, byte for byte.""" raw = " HTTP://X.Test:80/Some%20Path/?b=2&a=1#frag " bench, _ = upsert_bench(tmp_path, raw, "n", "o") assert bench.url == raw.strip() assert bench.id != bench.url assert bench.id == "http://x.test/Some%20Path/?b=2&a=1" assert read_benches(tmp_path)[0][0].url == raw.strip() # ---- INV-4: the rendered order is TOTAL and stated -------------------------- def test_same_name_benches_do_not_swap(tmp_path): """INV-4. Defeating change: dropping the `id` tie-break. Two benches with the SAME name, registered in both orders, must render identically — a test over distinct names passes with no tie-break at all.""" def build(order): root = tmp_path / f"r{order}" root.mkdir() for u in (["http://a.test/", "http://b.test/"] if order else ["http://b.test/", "http://a.test/"]): upsert_bench(root, u, "same name", "o") return [b.id for b in read_benches(root)[0]] assert build(0) == build(1) def test_state_ranks_before_name(tmp_path): """INV-4. live → promoted → retired, THEN name. Defeating change: ordering by name alone, which a fixture of three same-state benches cannot see.""" upsert_bench(tmp_path, "http://a.test/", "aaa", "o") # would sort first by name upsert_bench(tmp_path, "http://z.test/", "zzz", "o") set_bench_state(tmp_path, normalize_bench_url("http://a.test/"), "retired") assert [b.name for b in read_benches(tmp_path)[0]] == ["zzz", "aaa"] def test_order_is_case_insensitive_on_name(tmp_path): upsert_bench(tmp_path, "http://b.test/", "Bravo", "o") upsert_bench(tmp_path, "http://a.test/", "alpha", "o") assert [b.name for b in read_benches(tmp_path)[0]] == ["alpha", "Bravo"] def test_order_benches_is_pure(tmp_path): """INV-4. Defeating change: `order_benches` doing I/O or sorting in place. Called with records belonging to NO root, it must still answer.""" made = [Bench(id=f"http://{c}.test/", url=f"http://{c}.test/", name=c, owner="o", state="live", added="", updated="") for c in "ba"] assert [b.name for b in order_benches(made)] == ["a", "b"] assert [b.name for b in made] == ["b", "a"], "input was mutated" # ---- INV-5: the read cannot raise, and cannot cost the caller unboundedly --- def _write_raw(root: pathlib.Path, payload: str) -> None: (root / BENCHES_FILE).write_text(payload) def test_absent_registry_is_not_an_error(tmp_path): benches, err = read_benches(tmp_path) assert benches == [] and err is None @pytest.mark.parametrize("payload,label", [ ("this is not json", "non-JSON bytes"), ("[]", "valid JSON of the wrong top-level shape"), ('{"benches": []}', "the list shape this unit deliberately does not use"), ('{"http://a/": "a string, not a record"}', "a value of the wrong type"), ('{"http://a/": {"name": [], "owner": "o", "state": "live"}}', "a FIELD of the wrong type"), ('{"http://a/": {"name": "n", "owner": "o", "state": "invented"}}', "an unknown state"), ]) def test_damaged_registries_report_rather_than_raise(tmp_path, payload, label): """INV-5. Defeating change: `json.load` with no guard, or `except: pass` which would report absent. The error must be NON-EMPTY — 'damaged' and 'absent' must not render the same, because only one of them needs a human. The wrong-typed-FIELD row is the shape currently 500ing the gallery elsewhere in this service.""" _write_raw(tmp_path, payload) benches, err = read_benches(tmp_path) assert err, f"{label} reported no error" assert benches == [] def test_oversized_registry_is_refused_by_size_before_parsing(tmp_path): """INV-5. Defeating change: parsing first and checking length after, which costs the caller the whole file. A FIFO has st_size 0, so the guard must bound the READ, not trust the stat — the 2026-09-22 hang lesson.""" from booth.benches import BENCHES_MAX_BYTES _write_raw(tmp_path, '{"http://a/": {"name": "' + "x" * BENCHES_MAX_BYTES + '"}}') benches, err = read_benches(tmp_path) assert err and benches == [] @pytest.mark.skipif(os.geteuid() == 0, reason="root ignores the mode bit") def test_unreadable_registry_reports_rather_than_raises(tmp_path): p = tmp_path / BENCHES_FILE p.write_text("{}") p.chmod(0o000) try: benches, err = read_benches(tmp_path) assert err and benches == [] finally: p.chmod(0o644) # ---- INV-1: one module knows the registry's filename ------------------------ def test_only_benches_py_names_the_registry_file(): """INV-1. Defeating change: a route reading `.benches.json` directly to save an import. Asserting that the panel renders would pass under exactly that.""" root = pathlib.Path(__file__).parent.parent offenders = [] for f in list((root / "booth").rglob("*.py")) + [root / "scripts" / "booth"]: if f.name == "benches.py": continue if ".benches.json" in f.read_text(): offenders.append(str(f.relative_to(root))) assert not offenders, f"the registry filename is hard-coded outside benches.py: {offenders}" # ---- INV-9: stdlib-only, AND sibling-free (seam review SR-1) ---------------- def test_benches_is_stdlib_only_and_imports_no_sibling(): """INV-9. The PARAMETRIZED test in test_marks.py exempts `booth` on purpose, so it cannot catch `from booth.links import booth_target` — which is exactly the import this unit tempts an implementer into. This is the strict copy, mirroring tests/test_manifest.py. Seam review SR-1.""" src = pathlib.Path(__file__).parent.parent / "booth" / "benches.py" tree = ast.parse(src.read_text()) roots = set() for node in ast.walk(tree): if isinstance(node, ast.Import): roots.update(a.name.split(".")[0] for a in node.names) elif isinstance(node, ast.ImportFrom): roots.add("booth" if node.level else (node.module or "").split(".")[0]) outside = {r for r in roots if r and r not in sys.stdlib_module_names} assert not outside, f"booth/benches.py imports outside the stdlib (booth.* included): {sorted(outside)}" # ---- INV-2: ONE predicate decides what a booth URL is ----------------------- # Every row is (url, expected booth name or None). Run against BOTH callers. BOOTH_URL_TABLE = [ ("http://10.100.10.50:8090/b/sindra-bakeoff/", "sindra-bakeoff"), ("http://10.100.10.50:8090/b/sindra-bakeoff", "sindra-bakeoff"), ("http://localhost:8090/b/x/", "x"), ("http://NH3-DEV.nh3.internal:8090/b/x/", "x"), # host-agnostic, any case ("https://10.100.10.50:8090/b/x/", "x"), # scheme-agnostic ("http://10.100.10.50:8090/b/my%20booth/", "my booth"), # SR-7: decoded ("http://10.100.10.50:8090/b/x/zoom/a.png", "x"), # nested path ("http://10.100.10.50:8090/b/x/?q=1", "x"), # query ("http://10.100.10.50:8090/b/x/#frag", "x"), ("http://10.100.10.50:8090/", None), # the Booth root IS a bench ("http://10.100.10.50:8090/b/", None), # no name ("http://10.100.10.50:8090/b//", None), ("http://10.100.10.50:8090/b/.hidden/", None), # resolve_booth's rules ("http://10.100.10.50:8090/b/%2e%2e/", None), # decoded `..` ("http://10.100.10.50:8090/b/a%2Fb/", None), # decoded separator ("https://gitea.phasefinal.com/vh/peedlar", None), ("not a url at all", None), ] @pytest.mark.parametrize("url,expected", BOOTH_URL_TABLE) def test_booth_target_classifies(url, expected): """INV-2. The table is shared with the CLI refusal test and the dead-marker test, so a second implementation in either place goes red here or there.""" assert booth_target(url) == expected def test_booth_target_never_raises(): """A board row is arbitrary operator-editable text; a predicate that raises on one row takes the whole page. Defeating change: `urlsplit` unguarded.""" for junk in ["", " ", "http://[oops", "\x00", "://", "http://]"]: assert booth_target(junk) is None # ---- upsert semantics ------------------------------------------------------- def test_added_survives_reregistration_updated_does_not(tmp_path): first, created = upsert_bench(tmp_path, "http://a.test/", "one", "o1") assert created second, created = upsert_bench(tmp_path, "http://a.test/", "two", "o2") assert not created assert second.added == first.added assert second.name == "two" and second.owner == "o2" def test_state_survives_reregistration(tmp_path): """A promoted bench that re-announces itself is still promoted — otherwise every deploy silently demotes it.""" upsert_bench(tmp_path, "http://a.test/", "one", "o") set_bench_state(tmp_path, normalize_bench_url("http://a.test/"), "promoted") again, _ = upsert_bench(tmp_path, "http://a.test/", "one again", "o") assert again.state == "promoted" def test_a_new_bench_is_live(tmp_path): bench, _ = upsert_bench(tmp_path, "http://a.test/", "one", "o") assert bench.state == "live" and bench.state in BENCH_STATES def test_set_state_refuses_an_unknown_state(tmp_path): upsert_bench(tmp_path, "http://a.test/", "one", "o") with pytest.raises(ValueError): set_bench_state(tmp_path, normalize_bench_url("http://a.test/"), "invented") def test_set_state_and_remove_miss_cleanly(tmp_path): assert set_bench_state(tmp_path, "http://nope/", "live") is None assert remove_bench(tmp_path, "http://nope/") is None def test_remove_returns_the_record_and_drops_it(tmp_path): upsert_bench(tmp_path, "http://a.test/", "one", "o") gone = remove_bench(tmp_path, normalize_bench_url("http://a.test/")) assert gone is not None and gone.name == "one" assert read_benches(tmp_path)[0] == [] def test_fields_are_capped_at_the_write(tmp_path): from booth.benches import NAME_MAX, OWNER_MAX bench, _ = upsert_bench(tmp_path, "http://a.test/", "n" * 500, "o" * 500) assert len(bench.name) == NAME_MAX and len(bench.owner) == OWNER_MAX def test_a_write_over_a_damaged_registry_does_not_destroy_it(tmp_path): """The 2026-09-21 lesson, in this unit's storage: reads are lenient, writes are STRICT. A damaged registry must not be silently replaced by a fresh one carrying only the new row — that is the marks-wipe bug in a new file.""" _write_raw(tmp_path, '{"http://a/": {"name": "real", "owner": "o", "state": "live"}, BROKEN') before = (tmp_path / BENCHES_FILE).read_text() with pytest.raises(ValueError): upsert_bench(tmp_path, "http://b.test/", "new", "o") assert (tmp_path / BENCHES_FILE).read_text() == before def test_the_on_disk_shape_is_an_object_keyed_by_id(tmp_path): """Two rows with one identity are then impossible BY CONSTRUCTION rather than by an upsert remembering to check.""" upsert_bench(tmp_path, "http://a.test/", "one", "o") raw = json.loads((tmp_path / BENCHES_FILE).read_text()) # The key is the NORMALIZED url, so the bare "/" is already gone — which is # the rule `test_these_pairs_are_one_bench` pins independently. assert isinstance(raw, dict) and list(raw) == ["http://a.test"] assert "id" not in raw["http://a.test"], "the key IS the id; storing it twice invites drift" # ---- the rendered surface --------------------------------------------------- # # The benches panel and the dead-row marker both live on the standing board's # page — the one booth carrying a links.md. from fastapi.testclient import TestClient # noqa: E402 from booth.app import create_app # noqa: E402 def _board(root: pathlib.Path, rows: str) -> pathlib.Path: b = root / "links" b.mkdir(parents=True, exist_ok=True) (b / "links.md").write_text(rows) return b def _client(root): return TestClient(create_app(root, ttl_hours=24, start_sweeper=False)) ROW_LIVE = "- [still here](http://10.100.10.50:8090/b/alive/) · x · 2026-09-01 00:00\n" ROW_DEAD = "- [swept](http://10.100.10.50:8090/b/gone/) · x · 2026-09-01 00:00\n" ROW_REF = "- [a repo](https://gitea.phasefinal.com/vh/peedlar) · x · 2026-09-01 00:00\n" def _dead_rows(body: str) -> list[str]: """Board ROWS carrying the dead class. Scoped to `
` on purpose: the class name also appears in base.html's stylesheet, so a whole-document substring test is always true and can never go red — a vacuous falsifier of exactly the shape persistent-memory.d/2026-09-22-vacuous-falsifiers.md describes. """ return [ln for ln in body.splitlines() if 'class="board-row' in ln and "board-dead" in ln] def test_a_dead_row_is_marked_and_a_live_one_is_not(tmp_path): """The marker. Defeating change: marking every `/b/` row dead, or none. Both a live target and a dead one are in ONE fixture, so a marker that is constant in either direction goes red.""" (tmp_path / "alive").mkdir() _board(tmp_path, ROW_LIVE + ROW_DEAD + ROW_REF) r = _client(tmp_path).get("/b/links/") assert r.status_code == 200 rows = _dead_rows(r.text) assert len(rows) == 1, f"exactly one of the three rows is dead, got {rows}" assert "/b/gone/" in r.text and "booth is gone" in r.text def test_the_marker_never_takes_the_page(tmp_path): """Seam review SR-2. `resolve_booth` RAISES HTTPException(404); calling it per row would turn one swept booth into a 404 for the whole board. This is the test that goes red under that exact implementation.""" _board(tmp_path, ROW_DEAD * 5) assert _client(tmp_path).get("/b/links/").status_code == 200 def test_a_percent_encoded_booth_is_not_marked_dead(tmp_path): """Seam review SR-7. `quote(name, safe="")` is how the service emits these, so the marker must decode before it looks on disk. Defeating change: comparing the raw path segment — which marks this row dead forever.""" (tmp_path / "my booth").mkdir() _board(tmp_path, "- [x](http://10.100.10.50:8090/b/my%20booth/) · x · 2026-09-01 00:00\n") assert _dead_rows(_client(tmp_path).get("/b/links/").text) == [] def test_a_reference_row_is_never_marked_dead(tmp_path): _board(tmp_path, ROW_REF) assert _dead_rows(_client(tmp_path).get("/b/links/").text) == [] def test_the_panel_renders_registered_benches(tmp_path): _board(tmp_path, ROW_REF) upsert_bench(tmp_path, "https://talk.nh3.phasefinal.com:8092/", "talk", "tts-dev") body = _client(tmp_path).get("/b/links/").text assert "talk" in body and "tts-dev" in body def test_the_anchor_href_is_the_raw_url_not_the_id(tmp_path): """INV-7. Defeating change: rendering `bench.id` in the href because it is 'the clean one'. The raw URL here normalizes differently in three ways.""" raw = "HTTP://Talk.NH3.test:80/Some%20Path/?b=2&a=1#frag" _board(tmp_path, ROW_REF) upsert_bench(tmp_path, raw, "talk", "o") body = _client(tmp_path).get("/b/links/").text assert 'href="HTTP://Talk.NH3.test:80/Some%20Path/?b=2&a=1#frag"' in body, \ "the href must be the URL as posted, byte for byte" @pytest.mark.parametrize("payload,label", [ (None, "absent"), ("not json", "non-JSON"), ("[]", "wrong top-level shape"), ('{"http://a/": {"name": [], "owner": "o", "state": "live"}}', "wrong-typed field"), ('{"http://a/": {"name": "n", "owner": "o", "state": "invented"}}', "unknown state"), ]) def test_a_damaged_registry_costs_its_panel_and_never_the_page(tmp_path, payload, label): """INV-5, at the render. The v0.2.2 lesson: a poisoned sidecar returned 500 for `/` and `/healthz` across all 25 booths. The wrong-typed-FIELD row is the shape currently 500ing the gallery elsewhere in this service, so it is the one that matters most.""" _board(tmp_path, ROW_REF) if payload is not None: (tmp_path / BENCHES_FILE).write_text(payload) c = _client(tmp_path) assert c.get("/b/links/").status_code == 200, label assert c.get("/").status_code == 200, label assert c.get("/healthz").status_code == 200, label def test_damaged_and_absent_render_different_text(tmp_path): """INV-5. Only ONE of them needs a human. Defeating change: `except: pass` returning ([], None), which renders damaged exactly like absent.""" _board(tmp_path, ROW_REF) absent = _client(tmp_path).get("/b/links/").text (tmp_path / BENCHES_FILE).write_text("not json") damaged = _client(tmp_path).get("/b/links/").text assert absent != damaged def test_the_panel_does_not_render_on_an_ordinary_booth(tmp_path): """A bench registry on every gallery page would be noise, and would cost a read per booth page view for a surface that belongs to exactly one.""" (tmp_path / "ordinary").mkdir() (tmp_path / "ordinary" / "a.png").write_bytes(b"\x89PNG\r\n\x1a\n") upsert_bench(tmp_path, "https://talk.test/", "talk", "o") assert "talk" not in _client(tmp_path).get("/b/ordinary/").text def test_the_benches_routes_round_trip(tmp_path): _board(tmp_path, ROW_REF) c = _client(tmp_path) assert c.post("/b/links/bench-add", data={"url": "https://x.test/", "name": "ex"}, follow_redirects=False).status_code in (302, 303) assert "ex" in c.get("/b/links/").text bid = normalize_bench_url("https://x.test/") c.post("/b/links/bench-state", data={"bench": bid, "state": "retired"}, follow_redirects=False) assert read_benches(tmp_path)[0][0].state == "retired" c.post("/b/links/bench-remove", data={"bench": bid}, follow_redirects=False) assert read_benches(tmp_path)[0] == [] def test_a_bad_url_posted_to_the_route_does_not_500(tmp_path): _board(tmp_path, ROW_REF) c = _client(tmp_path) r = c.post("/b/links/bench-add", data={"url": "ftp://x.test/f", "name": "ex"}, follow_redirects=False) assert r.status_code in (302, 303, 400) assert c.get("/b/links/").status_code == 200