booth/tests/test_manifest.py

"""U5 — self-announcing booths.

A booth carries `.booth.json` saying who posted it and why, and the index card
and the booth page render it. Closes job 5 (`Announce`) — the job nobody named,
whose absence is the measured cause of 145 dead link rows.

See docs/contracts/u5_booth_manifest.contract.md.
"""
import ast
import json
import os
import pathlib
import sys

import pytest

from booth.manifest import MANIFEST_FILE, Manifest, read_manifest, write_manifest


# ---- slice 1: the record and its storage ------------------------------------


def test_an_announcement_round_trips(tmp_path):
    b = tmp_path / "r18-ab"
    b.mkdir()
    written = write_manifest(b, "booth-dev", why="pick the winning denoiser")

    assert (b / MANIFEST_FILE).is_file()
    got = read_manifest(b)
    assert got == written
    assert got.handle == "booth-dev"
    assert got.why == "pick the winning denoiser"
    assert got.error is None


def test_the_title_falls_back_to_the_directory_name(tmp_path):
    """A booth always has a display name. `title` is the one the poster chose
    when there is one, and the folder name is a perfectly good one when there
    is not — an empty heading on a card is worse than a plain one."""
    b = tmp_path / "r18-ab"
    b.mkdir()
    assert write_manifest(b, "booth-dev").title == "r18-ab"
    assert write_manifest(b, "booth-dev", title="R18 A/B").title == "R18 A/B"


def test_a_booth_that_never_announced_reads_as_none(tmp_path):
    """The normal case for every booth that predates this unit, and for every
    booth that arrives by rsync — the documented path for any host that is not
    nh3-dev, which never runs the CLI at all."""
    b = tmp_path / "quiet"
    b.mkdir()
    assert read_manifest(b) is None
    assert read_manifest(tmp_path / "does-not-exist") is None


def test_one_line_by_construction_not_by_convention(tmp_path):
    """`why` renders inside a card's sub-line, so a newline in it would break
    the card rather than the field. Truncation and newline-stripping happen at
    the WRITE, so nothing downstream has to remember."""
    b = tmp_path / "b"
    b.mkdir()
    m = write_manifest(b, "booth-dev", why="first line\nsecond line\r\nthird")
    assert "\n" not in m.why and "\r" not in m.why
    assert "first line" in m.why and "second line" in m.why

    long = write_manifest(b, "booth-dev", why="x" * 5000)
    assert len(long.why) <= 200
    assert len(write_manifest(b, "y" * 500).handle) <= 64
    assert len(write_manifest(b, "booth-dev", title="t" * 500).title) <= 120


# ---- slice 2: the read cannot raise (INV-2) ---------------------------------


@pytest.mark.parametrize(
    "payload",
    [
        b"{truncated",                       # not JSON at all
        b"[]",                               # JSON, wrong shape
        b'"a string"',                       # JSON, wronger shape
        b"null",
        b'{"handle": 7}',                    # right shape, wrong type
        b'{"why": "no handle here"}',        # the one required field missing
        b"\xff\xfe not utf-8",
        b"",
    ],
    ids=["truncated", "list", "string", "null", "wrong-type", "no-handle",
         "not-utf8", "empty"],
)
def test_a_damaged_manifest_never_raises(tmp_path, payload):
    """INV-2. `list_booths` calls this once per booth on every index page load,
    so a read that can raise is a service-wide outage wearing a single-booth
    bug's clothes. That is not a hypothetical — a poisoned `.marks.json` did
    exactly that to `/` and `/healthz` across all 25 booths, and the fix shipped
    in v0.2.2. The same reader posture, applied before the same mistake."""
    b = tmp_path / "b"
    b.mkdir()
    (b / MANIFEST_FILE).write_bytes(payload)

    got = read_manifest(b)
    assert isinstance(got, Manifest)
    assert got.error, "a damaged manifest read clean"


def test_damaged_is_not_the_same_as_absent(tmp_path):
    """INV-5. Silently folding "cannot be read" into "never announced" would
    hide the one case somebody has to go and fix."""
    absent = tmp_path / "absent"
    absent.mkdir()
    damaged = tmp_path / "damaged"
    damaged.mkdir()
    (damaged / MANIFEST_FILE).write_text("{oops")

    assert read_manifest(absent) is None
    assert read_manifest(damaged).error


def test_a_manifest_the_module_did_not_write_still_reads(tmp_path):
    """Hand-written is a supported input: the file is plain JSON in a folder the
    operator owns, and half the point is that a booth is just a directory. Only
    `handle` is required; everything else has a default."""
    b = tmp_path / "b"
    b.mkdir()
    (b / MANIFEST_FILE).write_text(json.dumps({"handle": "shutter-dev"}))

    got = read_manifest(b)
    assert got.handle == "shutter-dev" and got.error is None
    assert got.title == "b"
    assert got.why == ""


# ---- slice 3: re-announcement (INV-3) ---------------------------------------


def test_re_announcing_preserves_created(tmp_path):
    """INV-3. `created` is when the booth APPEARED. Saying something more about
    it later is not a second appearance, and a `booth add` on an existing booth
    is the common case — the poster adds the second batch and sharpens the why."""
    b = tmp_path / "b"
    b.mkdir()
    first = write_manifest(b, "booth-dev", why="first pass")
    second = write_manifest(b, "booth-dev", why="second pass, sharper")

    assert second.created == first.created
    assert second.why == "second pass, sharper"


def test_re_announcing_over_a_damaged_file_does_not_inherit_its_created(tmp_path):
    """A `created` that cannot be read back is replaced rather than guessed at.
    The alternative is a stamp that is silently wrong, which is worse than one
    that is silently new."""
    b = tmp_path / "b"
    b.mkdir()
    (b / MANIFEST_FILE).write_text("{not json")

    m = write_manifest(b, "booth-dev", why="rescued")
    assert m.created and m.error is None
    assert read_manifest(b).why == "rescued"


# ---- slice 4: the write is atomic, and invisible to every listing -----------


def test_the_write_leaves_no_temp_file(tmp_path):
    """Half of the atomic-write promise, and the weaker half — see
    `test_the_write_replaces_rather_than_truncating` for the part that actually
    discriminates. Kept because a leaked `.tmp` is its own small defect: it
    would sit in the booth forever and, unlike the manifest, nothing would ever
    overwrite it."""
    b = tmp_path / "b"
    b.mkdir()
    write_manifest(b, "booth-dev", why="x")
    assert not list(b.glob("*.tmp")), "a temp file survived the write"


def test_a_manifest_is_not_an_item(tmp_path):
    """The whole integration story: it is a DOTFILE, so the existing
    `startswith('.')` skip in `booth_items` already keeps it out of tiles,
    counts and zips. No new exclusion rule anywhere. Asserted rather than
    assumed, because the claim is load-bearing for the contract's scope."""
    from booth.app import zip_booth
    from booth.items import booth_items

    b = tmp_path / "b"
    b.mkdir()
    (b / "a.txt").write_text("real content")
    write_manifest(b, "booth-dev", why="x")

    assert [i.rel for i in booth_items(b)] == ["a.txt"]
    assert MANIFEST_FILE not in zip_booth(b).decode("latin-1")


def test_announcing_is_activity(tmp_path):
    """A manifest is a dotfile but not a `.lock` dotfile, so `_newest_mtime`
    counts it. Creating or re-announcing a booth resets its TTL, which is right:
    both are somebody touching it. The lock exemption added in v0.2.2 is for
    machinery a READ path creates; this is a deliberate write."""
    from booth.app import booth_age_seconds

    b = tmp_path / "b"
    b.mkdir()
    old = 1_000_000_000
    os.utime(b, (old, old))

    write_manifest(b, "booth-dev", why="look at this")
    assert booth_age_seconds(b, now=old + 90_000) < 86_400


def test_stdlib_only():
    """INV-4, and the reason this module exists separately from anything that
    imports a third-party package. `scripts/booth` imports it under the system
    python3 with NO venv, through a `python3 -c` heredoc no AST extractor can
    see. It must also not import `booth.*`: a cross-import between two
    stdlib-only modules is a second way for the invariant to break."""
    src = pathlib.Path(__file__).parent.parent / "booth" / "manifest.py"
    roots = set()
    for node in ast.walk(ast.parse(src.read_text())):
        if isinstance(node, ast.Import):
            roots.update(a.name.split(".")[0] for a in node.names)
        elif isinstance(node, ast.ImportFrom):
            # A RELATIVE import (`from . import marks`) carries no module root
            # and used to pass this walk unseen — which matters more here than
            # in the shared copy, because this module forbids sibling imports
            # outright. Recorded as `booth` so the assertion below catches it.
            roots.add("booth" if node.level else
                      (node.module or "").split(".")[0])
    assert not (roots - set(sys.stdlib_module_names)), (
        f"booth/manifest.py imports outside the stdlib: "
        f"{sorted(roots - set(sys.stdlib_module_names))}"
    )


# ---- slice 5: what the operator actually sees -------------------------------


@pytest.fixture
def client(tmp_path):
    from fastapi.testclient import TestClient

    from booth.app import create_app
    return TestClient(create_app(tmp_path, ttl_hours=24, start_sweeper=False)), tmp_path


def _booth(data, name, *, kept=False):
    b = data / name
    b.mkdir()
    (b / "a.txt").write_text("content")
    if kept:
        (b / ".forever").touch()
    return b


@pytest.mark.parametrize("kept", [False, True], ids=["ephemeral", "kept"])
def test_the_index_card_carries_the_announcement(client, kept):
    """BOTH LANES. Kept boards render first and are a separate block in
    index.html, so patching only the ephemeral lane would leave the 15 kept
    booths — the durable, most-looked-at ones — with exactly the defect this
    unit closes. Same lesson as the `blurtoggle` macro: three branches, one
    definition; here it is two lanes and one rule."""
    c, data = client
    b = _booth(data, "r18-ab", kept=kept)
    write_manifest(b, "booth-dev", why="pick the winning denoiser")

    html = c.get("/").text
    assert "booth-dev" in html
    assert "pick the winning denoiser" in html


@pytest.mark.parametrize("kept", [False, True], ids=["ephemeral", "kept"])
def test_a_booth_that_never_spoke_up_is_marked(client, kept):
    """All 26 live booths are in this state, and rsync keeps making more. The
    marker is what makes the convention adoptable at all: the link board rotted
    to 69% precisely because nothing ever showed which rows were dead.

    ASSERTED ON THE CLASS, not on the word, and the test is named around it.
    `pytest`'s `tmp_path` is derived from the TEST NAME and the index renders
    `data_dir` in its empty-state hint — so a test called
    `test_an_unannounced_booth_says_so` put the literal string "unannounced"
    into the page and passed against a template that did not yet exist. A
    structural hook cannot be spelled by accident — though it has to be the
    rendered ELEMENT and not the bare class, since base.html ships a
    `.prov-none{...}` rule into the very same page."""
    c, data = client
    _booth(data, "quiet", kept=kept)

    html = c.get("/").text
    assert 'class="prov prov-none"' in html
    assert "unannounced" in html


def test_a_damaged_manifest_reads_differently_from_an_absent_one(client):
    """INV-5 on the surface the operator looks at, not just in the reader."""
    c, data = client
    b = _booth(data, "damaged")
    (b / MANIFEST_FILE).write_text("{oops")

    html = c.get("/").text
    assert 'class="prov prov-broken"' in html
    assert "unreadable" in html
    assert c.get("/b/damaged/").status_code == 200


def test_an_announced_booth_with_no_why_shows_only_its_handle(client):
    """`booth new x` with no --why is legal and common. The card shows who made
    it and does not invent a purpose or leave a dangling separator."""
    c, data = client
    b = _booth(data, "scratch")
    write_manifest(b, "booth-dev")

    html = c.get("/").text
    assert "booth-dev" in html
    assert 'class="prov prov-none"' not in html


def test_the_booth_page_header_carries_it_too(client):
    """Deliberate scope, not creep: a booth URL handed to the operator lands
    HERE, never on the index. Job 5 is 'operator, look at this', so the page he
    actually opens is where the answer has to be."""
    c, data = client
    b = _booth(data, "r18-ab")
    write_manifest(b, "booth-dev", why="pick the winning denoiser")

    html = c.get("/b/r18-ab/").text
    assert "booth-dev" in html
    assert "pick the winning denoiser" in html


def test_a_poisoned_manifest_cannot_take_down_the_index(client):
    """The v0.2.2 lesson, asserted for the new reader before it can repeat:
    `list_booths` touches every booth on every page load, so one bad file must
    cost that booth's provenance and nothing else."""
    c, data = client
    _booth(data, "good")
    bad = _booth(data, "bad")
    (bad / MANIFEST_FILE).write_bytes(b"\xff\xfe not utf-8 at all")

    assert c.get("/").status_code == 200
    assert c.get("/healthz").status_code == 200


def test_a_pickup_booth_announces_itself_as_the_booths_own(client):
    """No exemption list. A booth the service made says the service made it,
    which is true — and it keeps the rule to one line: a booth with no manifest
    is unannounced."""
    c, data = client
    r = c.post("/upload", files=[("files", ("a.txt", b"hello", "text/plain"))],
               follow_redirects=False)
    assert r.status_code in (200, 303)

    booth = next(p for p in data.iterdir() if p.is_dir())
    got = read_manifest(booth)
    assert got is not None and got.handle == "booth"
    assert 'class="prov prov-none"' not in c.get("/").text


# ---- findings from the cross-frontier CODE-REVIEW panel, 2026-09-22 ----------
#
# Heid panel (thread 01M341E9XAPZEFBSPK9HPGAM0S). Four arms, artifact-only.
# The round found ZERO drift in the strict sense and landed its weight one layer
# down, in test strength: five of the ten adopted findings are tests of mine
# that pass on the regression they exist to catch.


def test_the_read_survives_a_document_no_one_can_parse(tmp_path):
    """INV-2 said "never raises" and named a 4 GB file as a tested case. It was
    not tested, and it did not hold: `except ValueError` catches a truncated
    document, but `json.loads` on deeply nested input raises RecursionError,
    which is not a ValueError and is not an OSError either.

    `list_booths` calls this once per booth on every index load, so the one
    file costs the whole front page — the exact outage shape the invariant
    cites as its reason for existing. Three of four arms reached it
    independently; the eight-payload parametrize above has no size or depth
    case, so the hole stayed green.
    """
    b = tmp_path / "b"
    b.mkdir()
    (b / MANIFEST_FILE).write_text("[" * 200_000 + "]" * 200_000)

    got = read_manifest(b)
    assert isinstance(got, Manifest) and got.error


def test_the_read_refuses_a_document_too_large_to_be_a_manifest(tmp_path):
    """The other half of INV-2's named case. A manifest is four short fields;
    anything approaching a megabyte is not one, and reading it into memory to
    discover that is the wrong order of operations. Bounded BEFORE the read, so
    the size is checked by `stat` rather than survived."""
    from booth.manifest import MANIFEST_MAX_BYTES

    b = tmp_path / "b"
    b.mkdir()
    (b / MANIFEST_FILE).write_text('{"handle": "x", "why": "' +
                                   "y" * (MANIFEST_MAX_BYTES + 100) + '"}')

    got = read_manifest(b)
    assert isinstance(got, Manifest) and got.error
    assert "too large" in got.error


def test_a_hostile_directory_name_does_not_reach_the_record_raw(tmp_path):
    """`_one_line(title, TITLE_MAX) or booth.name` — the FALLBACK skips the
    normalization the explicit value gets. A directory name may legally carry a
    newline on POSIX and may be 255 bytes, and either lands in a card's
    sub-line. Same shape on the read path's fallback."""
    # 200-odd bytes, under the filesystem's own 255 limit but well over
    # TITLE_MAX — and a newline, which POSIX permits in a filename.
    name = "we" + "i" * 200 + "rd\nname"
    b = tmp_path / name
    b.mkdir()

    m = write_manifest(b, "booth-dev")
    assert "\n" not in m.title and len(m.title) <= 120
    assert "\n" not in read_manifest(b).title


def test_the_write_replaces_rather_than_truncating(tmp_path):
    """The previous version of this test asserted only that no `*.tmp` file
    survived — which a plain `write_text` passes, since it leaves no temp file
    either. All four arms said so, and they were right.

    THE INODE IS THE DISCRIMINATOR. `os.replace` publishes a different file over
    the old name, so the inode changes; truncate-and-rewrite keeps it. That is
    also exactly why the promise holds for a concurrent reader: it either has
    the old inode, intact, or opens the new one, complete. A test of the
    mechanism rather than of its litter.

    (An earlier draft spied on `os.open` to prove the published path was never
    opened for writing. It passed — vacuously. `Path.write_text` reaches the
    syscall through `io.open` in C and never touches the Python-level
    `os.open`, so the spy could not have fired either way. Recorded because
    writing a second vacuous test while fixing the first is the failure mode
    this whole round is about.)
    """
    b = tmp_path / "b"
    b.mkdir()
    published = b / MANIFEST_FILE

    write_manifest(b, "booth-dev", why="first")
    first_inode = published.stat().st_ino
    write_manifest(b, "booth-dev", why="second")

    assert published.stat().st_ino != first_inode, (
        "the manifest was rewritten in place, not replaced"
    )
    assert read_manifest(b).why == "second"


def test_the_temp_file_is_not_a_name_two_writers_share(tmp_path):
    """Every writer derived the same `.booth.json.tmp`. Two `booth add` calls on
    one booth could then interleave through a stale descriptor into the
    published path — the atomic-write promise is that READERS never see a
    partial file, and it says nothing about two writers sharing a scratch name.
    Marks are protected from this by their flock; the manifest has none."""
    b = tmp_path / "b"
    b.mkdir()
    seen = set()
    for i in range(5):
        write_manifest(b, "booth-dev", why=f"pass {i}")
        seen.update(p.name for p in b.iterdir() if p.name != MANIFEST_FILE)
    assert not seen, f"left temp files behind: {sorted(seen)}"

    from booth.manifest import _temp_path
    names = {_temp_path(b).name for _ in range(20)}
    assert len(names) > 1, "every writer derives the same temp name"


def test_a_bare_re_announce_does_not_wipe_the_why(tmp_path):
    """THE WORKFLOW IS `new --why` THEN `add`. Omitted flags meant empty
    strings, and empty strings overwrote — so the second command silently
    erased the sentence the first one existed to record, on the single most
    common sequence this feature has.

    Two arms of the paraphrase panel predicted it from the contract's wording
    alone ("gains a manifest with no why" does not distinguish a first write
    from a re-announce with the flags omitted). Every test I wrote passed
    `--why` on both calls, so none of them could see it.

    Omitted now means UNCHANGED; only a value that was actually supplied
    overwrites, and an explicit empty string still clears.
    """
    b = tmp_path / "b"
    b.mkdir()
    write_manifest(b, "booth-dev", title="R18 A/B", why="pick the denoiser")

    write_manifest(b, "booth-dev")                       # a bare `booth add`
    kept = read_manifest(b)
    assert kept.why == "pick the denoiser", "a bare re-announce wiped the why"
    assert kept.title == "R18 A/B"

    write_manifest(b, "booth-dev", why="sharper")        # supplied: overwrites
    assert read_manifest(b).why == "sharper"

    write_manifest(b, "booth-dev", why="")               # explicit: clears
    assert read_manifest(b).why == ""


def test_re_announcing_preserves_a_created_from_before_this_second(tmp_path):
    """`_now()` is whole-second resolution, so two `write_manifest` calls in a
    row share a timestamp and the old preservation test passed even against an
    implementation that regenerated `created` every time. Three of four arms
    caught it. Seed a stamp that could not have come from now()."""
    b = tmp_path / "b"
    b.mkdir()
    (b / MANIFEST_FILE).write_text(json.dumps({
        "handle": "booth-dev", "title": "b", "why": "first",
        "created": "2019-03-04T11:22:33-08:00",
    }))

    assert write_manifest(b, "booth-dev", why="second").created == \
        "2019-03-04T11:22:33-08:00"


def test_only_the_manifest_module_opens_the_manifest(tmp_path):
    """INV-1, which had no guard anywhere. One resolver is only one resolver
    while nothing else learns the filename."""
    root = pathlib.Path(__file__).parent.parent
    offenders = []
    for src in sorted((root / "booth").glob("*.py")):
        if src.name == "manifest.py":
            continue
        if ".booth.json" in src.read_text():
            offenders.append(src.name)
    assert not offenders, f"{offenders} name the manifest file directly"


def test_announcing_is_activity_via_the_manifest_file_itself(tmp_path):
    """The previous version could not fail. Writing the manifest creates a
    directory entry, which bumps the DIRECTORY's mtime, so the booth read as
    fresh whether or not `_newest_mtime` counted the manifest at all — a test
    of the side effect rather than of the thing.

    Put the directory's clock back afterwards, leaving the manifest's own mtime
    as the only thing that can keep the booth alive."""
    import os

    from booth.app import booth_age_seconds

    b = tmp_path / "b"
    b.mkdir()
    old = 1_000_000_000
    os.utime(b, (old, old))
    write_manifest(b, "booth-dev", why="look at this")
    os.utime(b, (old, old))                  # only the file can save it now

    assert booth_age_seconds(b, now=old + 90_000) < 86_400


def test_the_booth_header_marks_an_unannounced_booth_too(client):
    """The negative states were asserted on `/` only, so a header that rendered
    provenance for clean manifests and nothing for the other two would have
    passed the whole suite."""
    c, data = client
    _booth(data, "quiet")
    damaged = _booth(data, "damaged")
    (damaged / MANIFEST_FILE).write_text("{oops")

    assert 'class="prov prov-none"' in c.get("/b/quiet/").text
    assert 'class="prov prov-broken"' in c.get("/b/damaged/").text


def test_the_title_reaches_a_surface(client):
    """`--title` promised a display name and nothing rendered it — 4/4 on the
    paraphrase panel, independently the top-ranked flag of that round. It lands
    on the booth page heading, where there is room for it; the INDEX card keeps
    the directory name, because that is the identity the operator navigates and
    refers to positionally."""
    c, data = client
    b = _booth(data, "r18-ab")
    write_manifest(b, "booth-dev", title="R18 A/B — denoiser bakeoff", why="w")

    page = c.get("/b/r18-ab/").text
    assert "R18 A/B — denoiser bakeoff" in page
    assert "r18-ab" in page, "the directory name stopped being visible"