booth/tests/test_benches.py

"""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
    ("http://[::1]:80/a", "http://[::1]/a"),          # default port, bracketed
    ("http://[::1]/A", "http://[::1]/A"),             # bracket round-trips
])
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
    # A NON-DEFAULT PORT IS PART OF THE IDENTITY. Without this vector, "always
    # omit the port" passes every other row in this file — caught by the cold
    # panel's per-invariant "what would still pass" pass, not by us.
    ("http://x.test:8092/p", "http://x.test/p"),
    ("https://x.test:8443/p", "https://x.test/p"),
])
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():
    """INV-4. Defeating change: dropping the `id` tie-break.

    THIS TEST USED TO GO THROUGH THE REGISTRY AND COULD NOT FAIL. `_write_all`
    serializes with `sort_keys=True`, so whatever order two benches were
    inserted in, they came back off disk already id-sorted — and removing the
    tie-break from `order_benches` left it green. A vacuous falsifier of
    exactly the shape persistent-memory.d/2026-09-22-vacuous-falsifiers.md
    describes: it asserted the outcome the author had in mind rather than the
    discriminator the invariant names. Caught by the cold panel (hulda, solo),
    not by us.

    So it calls `order_benches` DIRECTLY, with records that tie on both prior
    keys, presented in both orders. Nothing upstream can pre-sort them.
    """
    def recs(order):
        pair = [
            Bench(id="http://a.test/", url="http://a.test/", name="same name",
                  owner="o", state="live", added="", updated=""),
            Bench(id="http://b.test/", url="http://b.test/", name="same name",
                  owner="o", state="live", added="", updated=""),
        ]
        return pair if order else list(reversed(pair))
    assert [b.id for b in order_benches(recs(0))] == \
           [b.id for b in order_benches(recs(1))]
    assert [b.id for b in order_benches(recs(1))] == ["http://a.test/", "http://b.test/"]


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."""
    import booth.benches as B
    _write_raw(tmp_path, '{"http://a/": {"name": "' + "x" * B.BENCHES_MAX_BYTES + '"}}')
    benches, err = read_benches(tmp_path)
    assert err and benches == []
    # AND PROVE THE PARSE WAS NEVER REACHED. Asserting only the eventual result
    # passes an implementation that loads the whole document and checks its
    # length afterwards — which costs the caller exactly what the cap exists to
    # save. Booby-trap json.loads: if it runs, the test says so. Cold panel,
    # hulda F11.
    import json as _json
    tripped = []
    real = _json.loads

    def trap(*a, **k):
        tripped.append(True)
        return real(*a, **k)
    B.json.loads = trap
    try:
        benches, err = read_benches(tmp_path)
    finally:
        B.json.loads = real
    assert err and not tripped, "the oversized registry was parsed before it was refused"


@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)}"
    # A STRING IMPORT IS INVISIBLE TO THE WALK ABOVE. `__import__("booth.links")`
    # or `importlib.import_module(...)` inside a function defeats it entirely,
    # and that is the exact shape someone reaches for when a sibling import is
    # refused by review. Caught by the cold panel's per-invariant vacuity pass.
    called = {n.func.id for n in ast.walk(tree)
              if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)}
    assert "__import__" not in called, "benches.py imports by string, defeating the AST walk"
    assert "importlib" not in roots, "benches.py can import anything at runtime via importlib"


# ---- 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),
    # THE ACCEPTED COST, MADE EXPLICIT. The predicate is host-agnostic on
    # purpose — a host allowlist fails OPEN on whichever name somebody reaches
    # this service by next — so a third-party URL with a `/b/<x>` path reads as
    # a booth link and is refused. The contract names this trade-off; the table
    # had no row exercising it, so nothing pinned the behaviour either way.
    # Cold panel, hulda F10.
    ("https://example.com/b/not-ours/", "not-ours"),
]


@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"
    assert second.updated >= first.updated


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/) <sub>· x · 2026-09-01 00:00</sub>\n"
ROW_DEAD = "- [swept](http://10.100.10.50:8090/b/gone/) <sub>· x · 2026-09-01 00:00</sub>\n"
ROW_REF = "- [a repo](https://gitea.phasefinal.com/vh/peedlar) <sub>· x · 2026-09-01 00:00</sub>\n"


def _dead_rows(body: str) -> list[str]:
    """Board ROWS carrying the dead class.

    Scoped to `<div class="board-row ...">` 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/) <sub>· x · 2026-09-01 00:00</sub>\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&amp;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"),
    ("OVERSIZED", "over the size cap"),
    ("UNREADABLE", "chmod 000"),
    ("FIFO", "a named pipe"),
])
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)
    reg = tmp_path / BENCHES_FILE
    if payload == "OVERSIZED":
        from booth.benches import BENCHES_MAX_BYTES
        reg.write_text('{"http://a/": {"name": "' + "x" * BENCHES_MAX_BYTES + '"}}')
    elif payload == "UNREADABLE":
        if os.geteuid() == 0:
            pytest.skip("root ignores the mode bit")
        reg.write_text("{}")
        reg.chmod(0o000)
    elif payload == "FIFO":
        os.mkfifo(reg)
    elif payload is not None:
        reg.write_text(payload)
    try:
        c = _client(tmp_path)
        body = c.get("/b/links/")
        assert body.status_code == 200, label
        assert c.get("/").status_code == 200, label
        assert c.get("/healthz").status_code == 200, label
        # AND THE ERROR IS VISIBLE. Asserting only 200 was the gap: a render
        # that swallowed the failure and drew an empty panel passed every case
        # here while telling the operator nothing needed fixing. Absent is the
        # one case that must NOT show an error.
        shown = "the bench registry could not be read" in body.text
        assert shown is (payload is not None), label
    finally:
        if payload == "UNREADABLE":
            reg.chmod(0o644)


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


# ---- found by the in-session adversarial pass, after the cold panels shipped -


def test_a_fifo_at_the_registry_path_cannot_hang_the_render(tmp_path):
    """A NAMED PIPE IS NOT A REGULAR FILE, AND open() BLOCKS ON IT.

    This is the 2026-09-22 lesson recurring in a new file: a size cap that
    bounds the READ does not help, because the hang is in `open()` — a FIFO
    with no writer blocks there forever, before a single byte is bounded.
    `read_benches` runs on the board page's render path, so one FIFO would hang
    that request and, with enough hits, the threadpool behind every route.

    The guard is a REGULAR-FILE check before the open, which is what marks.py
    already does (`stat.S_ISREG`). Defeating change: reverting to `path.open()`
    guarded only by a byte cap — which is what this unit shipped first, while
    its docstring claimed the cap closed exactly this hole.
    """
    os.mkfifo(tmp_path / BENCHES_FILE)
    benches, err = read_benches(tmp_path)      # must RETURN, not block
    assert benches == [] and err


def test_a_directory_at_the_registry_path_is_an_error_not_a_crash(tmp_path):
    (tmp_path / BENCHES_FILE).mkdir()
    benches, err = read_benches(tmp_path)
    assert benches == [] and err


@pytest.mark.parametrize("encoded", ["%00", "%0a", "%0d", "%09", "%1b"])
def test_a_control_character_is_not_an_addressable_booth(encoded):
    """`unquote` happily produces a NUL or a newline, and neither can name a
    real directory. Left unfiltered they reach `is_dir()` (which raises
    ValueError on an embedded NUL on some paths), the refusal message the CLI
    prints, and the marker the board renders. Defeating change: dropping the
    control-character clause — the `%2e%2e` and `%2f` rows above stay green
    under it, so this needs its own."""
    assert booth_target(f"http://h:8090/b/{encoded}/") is None


def test_normalization_is_idempotent(tmp_path):
    """LOAD-BEARING for `bench state <id|url>` and `bench rm <id|url>`: both
    normalize whatever they are handed, so an id must normalize to itself or
    addressing a bench by the id the registry stores would miss it. Defeating
    change: any rule that rewrites an already-normalized form."""
    for u in (GITEA_EIGHT + TALK_FIVE + [
        "http://x.test/", "http://x.test:8080/p/", "https://x.test/?a=1",
        "HTTP://X.Test:80/Some%20Path/?b=2&a=1#frag",
    ]):
        once = normalize_bench_url(u)
        assert normalize_bench_url(once) == once, u


def test_a_failed_write_leaves_no_scratch_file(tmp_path, monkeypatch):
    """The temp file is named per-pid so two writers cannot share it, but a
    write that dies between create and replace would strand it beside the
    registry forever. Defeating change: dropping the cleanup."""
    import booth.benches as B
    upsert_bench(tmp_path, "http://a.test/", "one", "o")
    real = B.os.replace

    def boom(src, dst):
        raise OSError("disk full")
    monkeypatch.setattr(B.os, "replace", boom)
    with pytest.raises(OSError):
        upsert_bench(tmp_path, "http://b.test/", "two", "o")
    monkeypatch.setattr(B.os, "replace", real)
    strays = [p.name for p in tmp_path.iterdir() if ".tmp" in p.name]
    assert not strays, strays
    # and the prior registry is intact — a failed write destroys nothing
    assert [b.name for b in read_benches(tmp_path)[0]] == ["one"]


def test_an_ipv6_literal_keeps_its_brackets():
    """`urlsplit().hostname` strips them, and a netloc rebuilt from it is not
    another spelling of the URL — it is a broken one, so a re-post never
    matches the row the operator means to update. Defeating change: rebuilding
    netloc from `hostname` with no re-wrap, which is what this shipped as."""
    assert normalize_bench_url("http://[::1]:8080/a") == "http://[::1]:8080/a"
    assert normalize_bench_url("http://[2001:DB8::1]/p") == "http://[2001:db8::1]/p"
    assert normalize_bench_url("HTTP://[::1]:80/p") == "http://[::1]/p"
    # An UNBRACKETED IPv6 netloc is refused with a reason, not repaired:
    # `urlsplit(...).port` raises on `::1:8080` because it cannot tell the
    # address from the port — which is precisely why the brackets exist. The
    # refusal is the honest answer; guessing where the address ends would be
    # inventing an identity out of an ambiguous string.
    with pytest.raises(ValueError):
        normalize_bench_url("http://::1:8080/a")


def test_deeply_nested_json_does_not_escape_the_read(tmp_path):
    """RecursionError is neither ValueError nor OSError, so it went straight
    past `read_benches`'s except pair and 500'd the page the function exists to
    protect. The byte cap does not help: 200k open brackets is 200 KB, well
    inside it. Defeating change: dropping the RecursionError arm."""
    (tmp_path / BENCHES_FILE).write_text("[" * 200_000)
    benches, err = read_benches(tmp_path)
    assert benches == [] and err


def test_the_panel_renders_on_an_EMPTY_board(tmp_path):
    """The panel is gated on PAGE IDENTITY, not page content. Gating on
    `board or benches` hid the panel and its registration form exactly when the
    board was empty and the registry absent — the state a new deployment starts
    in, and the one where "no benches registered yet" is most worth saying.
    Defeating change: any content-derived gate."""
    _board(tmp_path, "")
    body = _client(tmp_path).get("/b/links/").text
    assert "no benches registered yet" in body
    assert "bench-add" in body, "the registration form vanished with the panel"


def test_the_panel_shows_when_a_bench_was_added(tmp_path):
    """INV-N/What renders: the contract says the panel shows the date it was
    added; `b.added` appeared nowhere in the template and no test asked. All
    four cold arms found this independently."""
    _board(tmp_path, ROW_REF)
    upsert_bench(tmp_path, "https://talk.test/", "talk", "o")
    added = read_benches(tmp_path)[0][0].added[:10]
    assert added in _client(tmp_path).get("/b/links/").text


@pytest.mark.parametrize("url,expected", BOOTH_URL_TABLE)
def test_the_dead_marker_classifies_the_SAME_table(tmp_path, url, expected):
    """INV-2 names RENDER-LEVEL agreement, and the marker tests never ran the
    table — three hand-written rows with no query between them, so a marker
    that stopped calling `booth_target` and treated `?q=1` as "not a booth"
    stayed green while disagreeing with the CLI. Caught by the cold panel.

    Every row whose target does not exist on disk must be marked dead; every
    non-booth row must not be."""
    _board(tmp_path, f"- [r]({url}) <sub>· x · 2026-09-01 00:00</sub>\n")
    marked = bool(_dead_rows(_client(tmp_path).get("/b/links/").text))
    assert marked is (expected is not None), (url, expected)


def test_updated_is_replaced_and_added_is_not(tmp_path, monkeypatch):
    """The other half of `test_added_survives_reregistration_updated_does_not`,
    which asserted only the half in the first clause of its own name.

    The stamp has SECOND resolution, so a fast test cannot tell a replaced
    `updated` from a frozen one by comparing real clocks — `>=` passes either
    way, which is a falsifier that cannot fail. The clock is driven instead, so
    "was it rewritten" is answerable. Cold panel, hulda F12.

    Defeating change: carrying `updated` forward from the prior record the way
    `added` is carried, which every real-clock assertion in this file survives.
    """
    import booth.benches as B
    ticks = iter(["2026-01-01T00:00:00+00:00",
                  "2026-06-06T06:06:06+00:00",
                  "2026-12-31T23:59:59+00:00"])
    monkeypatch.setattr(B, "_now", lambda: next(ticks))

    first, _ = upsert_bench(tmp_path, "http://a.test/", "one", "o")
    assert first.added == first.updated == "2026-01-01T00:00:00+00:00"

    second, _ = upsert_bench(tmp_path, "http://a.test/", "two", "o")
    assert second.added == "2026-01-01T00:00:00+00:00", "added must survive an upsert"
    assert second.updated == "2026-06-06T06:06:06+00:00", "updated must be replaced"

    # A STATE CHANGE IS A MUTATION and bumps it too — this is what the contract
    # was amended to say, after the panel read "most recent upsert" literally.
    third = set_bench_state(tmp_path, normalize_bench_url("http://a.test/"), "retired")
    assert third.added == "2026-01-01T00:00:00+00:00"
    assert third.updated == "2026-12-31T23:59:59+00:00"


def test_a_registration_cannot_make_the_registry_unreadable(tmp_path):
    """Cold contract panel, hulda solo: the write path permitted a file the
    reader then refuses on size — so the LAST bench somebody added would be the
    one that made every other bench invisible, and the write that did it
    reported success.

    Defeating change: dropping the size check from `_write_all`. The reader is
    lenient about damage and deliberately NOT lenient about size; a writer
    ignoring a limit its own reader enforces manufactures exactly the state
    that leniency exists to survive."""
    from booth.benches import BENCHES_MAX_BYTES, NAME_MAX
    n = 0
    while True:
        n += 1
        try:
            upsert_bench(tmp_path, f"http://h{n}.test/{'p' * 1800}", "x" * NAME_MAX, "o")
        except ValueError as exc:
            assert "past" in str(exc) and str(BENCHES_MAX_BYTES) in str(exc)
            break
        assert n < 500, "never hit the cap; widen the fixture"
    # THE REGISTRY IS STILL READABLE, and still holds everything that fit.
    benches, err = read_benches(tmp_path)
    assert err is None, err
    assert len(benches) == n - 1


def test_an_over_long_stored_url_is_damage_not_a_silent_clip(tmp_path):
    """Cold contract panel, 4-of-4 on cap semantics: "applied at the read" did
    not say TRUNCATE or REFUSE, and the code had picked truncate for every
    field. For `name` and `owner` that is right — they are display budgets and
    clipping costs a few characters in a panel row. For `url` it is wrong:
    INV-7 promises the click goes to the posted address byte for byte, and a
    clipped URL keeps that promise in the type system while breaking it in the
    browser. Defeating change: routing `url` back through `_cap`."""
    from booth.benches import URL_MAX
    long_url = "http://a/" + "p" * (URL_MAX + 10)
    _write_raw(tmp_path, json.dumps({"http://a/": {
        "url": long_url, "name": "n", "owner": "o", "state": "live"}}))
    benches, err = read_benches(tmp_path)
    assert err and benches == [], "an over-long url was clipped into a dead anchor"


def test_name_and_owner_ARE_clipped_at_the_read(tmp_path):
    """The other half of the same rule, so the asymmetry is pinned in both
    directions rather than asserted in one."""
    from booth.benches import NAME_MAX, OWNER_MAX
    _write_raw(tmp_path, json.dumps({"http://a/": {
        "url": "http://a/", "name": "n" * 500, "owner": "o" * 500, "state": "live"}}))
    benches, err = read_benches(tmp_path)
    assert err is None
    assert len(benches[0].name) == NAME_MAX and len(benches[0].owner) == OWNER_MAX


def test_the_benches_panel_is_not_nested_inside_a_span(tmp_path):
    """Cold bug-hunt panel, 3-of-4, seat-confirmed by byte offset in the live
    document: the panel `<div>` had landed INSIDE the booth header's
    `<span class="sub">`, because the insertion matched the first
    `{% if board %}` in the template rather than the block-level one.

    A `<div>` inside a `<span>` is invalid HTML — the parser closes the span
    implicitly and hoists the div out, orphaning the rest of the sub-line. It
    renders "fine" in the sense that nothing 500s, which is exactly why no
    other test in this file could see it.

    Checked the way the seat checked it: by offset. Defeating change: moving
    the panel back above the sub-span's close."""
    _board(tmp_path, ROW_REF)
    upsert_bench(tmp_path, "https://talk.test/", "talk", "o")
    body = _client(tmp_path).get("/b/links/").text
    sub_open = body.index('<span class="sub">')
    sub_close = body.index("</span>", body.index("· ", sub_open))
    panel = body.index('<div class="benches">')
    assert not (sub_open < panel < sub_close), (
        f"the benches div (offset {panel}) sits inside the sub span "
        f"({sub_open}..{sub_close})")


def test_a_symlinked_booth_is_dead_to_the_marker_as_it_is_to_the_page(tmp_path):
    """Cold bug-hunt panel, 3-of-4: `_booth_exists` used a bare `is_dir()`
    while `resolve_booth` resolves and requires the parent to BE the data root.
    They disagreed on a symlink — the marker called a booth pointing outside
    the root alive while the page 404s it, so the row rendered healthy and the
    link was dead. The worst of both, and invisible.

    Defeating change: dropping the containment check from `_booth_exists`."""
    outside = tmp_path.parent / f"outside-{tmp_path.name}"
    outside.mkdir()
    try:
        (tmp_path / "escapee").symlink_to(outside, target_is_directory=True)
    except OSError:
        pytest.skip("no symlink support here")
    _board(tmp_path, "- [x](http://h:8090/b/escapee/) <sub>· a · 2026-09-01 00:00</sub>\n")
    c = _client(tmp_path)
    body = c.get("/b/links/")
    assert body.status_code == 200
    # the page's own verdict on that name, which the marker must agree with
    assert c.get("/b/escapee/").status_code == 404
    assert _dead_rows(body.text), "the marker called a booth alive that the page 404s"