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_is_atomic(tmp_path):
    """INV-5 of CLAUDE.md. The CLI writes this in one process while the browser
    reads it in another, so a reader must never see a half-written document."""
    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) and node.level == 0 and node.module:
            roots.add(node.module.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