booth/tests/test_cli.py

"""`scripts/booth` — the surface every fleet session actually calls.

It had no tests at all, which the 2026-09-22 bug-hunt panel found the hard way:
its guard-strength table returned UNVERIFIED for every CLI claim because nothing
in the suite executes the script. Two of that round's findings live in here.

These run the real script under the real system `python3` with no venv, which
also makes them a live check on INV-1 (stdlib-only): a third-party import in
`marks.py` fails here the same way it fails on a fleet host.
"""
import json
import os
import pathlib
import subprocess

import pytest

SCRIPT = pathlib.Path(__file__).parent.parent / "scripts" / "booth"

# Exit codes the verbs promise. 0 is a successful read; a reader that CRASHED
# must never be one of the meaningful codes, or a caller cannot tell "no" from
# "broken" — which is the whole finding.
OK, UNANSWERED, NO_SUCH_PICK, READER_FAILED = 0, 1, 2, 3


def run(data, *args, **kw):
    env = {**os.environ, "BOOTH_DATA_DIR": str(data), "BOOTH_URL": "http://booth.invalid"}
    return subprocess.run([str(SCRIPT), *args], capture_output=True, text=True,
                          env=env, timeout=30, **kw)


@pytest.fixture
def booth(tmp_path):
    b = tmp_path / "b"
    b.mkdir()
    return tmp_path, b


def _declare(booth_dir, mark_id="winner"):
    import sys
    sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
    from booth.marks import declare_pick
    declare_pick(booth_dir, mark_id,
                 {"prompt": "Which one?", "options": ["A", "B"]})


def test_marks_prints_one_json_document(booth):
    """`booth marks <name>` is a read. Its stdout is parsed by the session that
    called it, so it has to be ONE document — and exit 0, because the read
    succeeded. Whether a pick is open is in the payload's `open` list, which is
    where a caller should read it from."""
    data, b = booth
    _declare(b)
    r = run(data, "marks", "b")
    assert r.returncode == OK, r.stderr
    doc = json.loads(r.stdout)
    assert doc["open"] == ["winner"]


def test_marks_wait_prints_once_not_once_per_poll(booth):
    """`--wait` polls every 2 s and printed the whole document on every pass, so
    a capture held several concatenated JSON values and `jq` could not read any
    of them. The wait is a wait; the print is the result."""
    data, b = booth
    _declare(b)
    import sys
    sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
    from booth.marks import answer_pick

    # Answer it after the first poll so --wait genuinely loops at least once.
    r = subprocess.Popen([str(SCRIPT), "marks", "b", "--wait", "20"],
                         stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
                         env={**os.environ, "BOOTH_DATA_DIR": str(data),
                              "BOOTH_URL": "http://booth.invalid"})
    import time
    time.sleep(3)
    answer_pick(b, "winner", "A")
    out, err = r.communicate(timeout=30)
    assert r.returncode == OK, err
    json.loads(out)          # ONE document, or this raises


def test_marks_reports_a_reader_failure_instead_of_printing_garbage(booth):
    """A traceback on stdout with exit 0 is the worst of both: the caller's `jq`
    sees success and gets nothing. A read that could not happen is its own
    answer and gets its own code."""
    data, b = booth
    (b / ".marks.json").write_bytes(b"\xff\xfe not utf-8 at all")
    r = run(data, "marks", "b")
    assert r.returncode == READER_FAILED, f"rc={r.returncode} out={r.stdout!r}"


def test_answer_distinguishes_a_crash_from_an_unanswered_pick(booth):
    """`answer` funnelled a reader crash and "not yet answered" through the same
    exit 1, so `--wait` spun for the full hour on a broken file and then blamed
    the operator for not answering."""
    data, b = booth
    _declare(b)
    r = run(data, "answer", "b", "winner")
    assert r.returncode == UNANSWERED

    (b / ".marks.json").write_bytes(b"\xff\xfe not utf-8 at all")
    r = run(data, "answer", "b", "winner", "--wait", "6")
    assert r.returncode == READER_FAILED, (
        "a crash was read as 'unanswered' and waited out the timeout"
    )


def test_answer_on_a_note_id_says_no_such_pick(booth):
    """`answer` matched on id alone while the web route filters on shape, so a
    note id was reported 'unanswered' and polled forever — a question that could
    never be answered because it was never a question."""
    data, b = booth
    import sys
    sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
    from booth.marks import write_note
    write_note(b, "a.png", "just a note")

    r = run(data, "answer", "b", "note-1")
    assert r.returncode == NO_SUCH_PICK
    assert "no such pick" in r.stderr


# ---- U5: self-announcing booths ---------------------------------------------


def _manifest(booth_dir):
    import sys
    sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
    from booth.manifest import read_manifest
    return read_manifest(booth_dir)


def test_new_announces_the_booth(tmp_path):
    """`$ALTHING_HANDLE` is the whole provenance story: the session already has
    it, so the booth can say who made it without anybody typing a name."""
    env = {**os.environ, "ALTHING_HANDLE": "shutter-dev"}
    r = subprocess.run([str(SCRIPT), "new", "r18-ab", "--why", "pick the winner"],
                       capture_output=True, text=True, timeout=30,
                       env={**env, "BOOTH_DATA_DIR": str(tmp_path),
                            "BOOTH_URL": "http://booth.invalid"})
    assert r.returncode == 0, r.stderr
    m = _manifest(tmp_path / "r18-ab")
    assert m.handle == "shutter-dev"
    assert m.why == "pick the winner"


def test_new_without_a_why_is_still_legal(tmp_path):
    """The flags are optional and existing call sites keep working. A booth
    that says only who made it is still a booth that said something."""
    r = subprocess.run([str(SCRIPT), "new", "scratch"], capture_output=True,
                       text=True, timeout=30,
                       env={**os.environ, "ALTHING_HANDLE": "booth-dev",
                            "BOOTH_DATA_DIR": str(tmp_path),
                            "BOOTH_URL": "http://booth.invalid"})
    assert r.returncode == 0, r.stderr
    m = _manifest(tmp_path / "scratch")
    assert m.handle == "booth-dev" and m.why == ""


def test_add_announces_and_still_copies_the_files(tmp_path):
    """`add` is the verb most sessions actually use — it creates the booth AND
    fills it — so the why has to ride on it or it rides nowhere."""
    src = tmp_path / "src"
    src.mkdir()
    (src / "a.txt").write_text("content")
    r = subprocess.run([str(SCRIPT), "add", "r18-ab", str(src / "a.txt"),
                        "--why", "second pass", "--title", "R18 A/B"],
                       capture_output=True, text=True, timeout=30,
                       env={**os.environ, "ALTHING_HANDLE": "booth-dev",
                            "BOOTH_DATA_DIR": str(tmp_path),
                            "BOOTH_URL": "http://booth.invalid"})
    assert r.returncode == 0, r.stderr
    assert (tmp_path / "r18-ab" / "a.txt").read_text() == "content"
    m = _manifest(tmp_path / "r18-ab")
    assert m.why == "second pass" and m.title == "R18 A/B"


def test_add_re_announcing_keeps_the_original_created(tmp_path):
    """The common shape: `new` opens the booth, `add` drops the second batch and
    sharpens the why. The booth appeared once."""
    env = {**os.environ, "ALTHING_HANDLE": "booth-dev",
           "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"}
    src = tmp_path / "a.txt"
    src.write_text("x")
    subprocess.run([str(SCRIPT), "new", "b", "--why", "first"], check=True,
                   capture_output=True, timeout=30, env=env)
    first = _manifest(tmp_path / "b").created
    subprocess.run([str(SCRIPT), "add", "b", str(src), "--why", "sharper"],
                   check=True, capture_output=True, timeout=30, env=env)

    after = _manifest(tmp_path / "b")
    assert after.created == first
    assert after.why == "sharper"


def test_the_link_board_announces_itself_as_the_booths_own(tmp_path):
    """No exemption list. The standing board is made by the service and posted
    to by seventeen handles, so no single agent owns it — `booth` is the
    truthful answer, and it keeps the rule to one line."""
    r = subprocess.run([str(SCRIPT), "link", "http://example.invalid", "a thing"],
                       capture_output=True, text=True, timeout=30,
                       env={**os.environ, "ALTHING_HANDLE": "booth-dev",
                            "BOOTH_DATA_DIR": str(tmp_path),
                            "BOOTH_URL": "http://booth.invalid"})
    assert r.returncode == 0, r.stderr
    m = _manifest(tmp_path / "links")
    assert m is not None and m.handle == "booth"
    assert m.why


def test_the_flags_can_sit_on_either_side_of_the_files(tmp_path):
    """`booth add b *.png --why "..."` and `booth add b --why "..." *.png` both
    work. A glob is usually last and a flag usually after it, but nothing
    enforces that and a session should not have to remember which."""
    src = tmp_path / "a.png"
    src.write_bytes(b"x")
    env = {**os.environ, "ALTHING_HANDLE": "booth-dev",
           "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"}
    for name, args in (("after", ["add", "after", str(src), "--why", "w"]),
                       ("before", ["add", "before", "--why", "w", str(src)])):
        r = subprocess.run([str(SCRIPT), *args], capture_output=True, text=True,
                           timeout=30, env=env)
        assert r.returncode == 0, r.stderr
        assert _manifest(tmp_path / name).why == "w"
        assert (tmp_path / name / "a.png").exists(), "the files stopped being copied"


def test_a_why_survives_quotes_and_non_ascii_and_is_flattened(tmp_path):
    """The reason this goes through manifest.py instead of printf-ing JSON from
    the shell: a why containing a quote, a backslash or a newline is not an edge
    case, it is a sentence somebody wrote. Newlines flatten because the field
    renders inside a card's sub-line."""
    r = subprocess.run(
        [str(SCRIPT), "new", "b", "--why", 'he said "pick v3" — line1\nline2 · ünï'],
        capture_output=True, text=True, timeout=30,
        env={**os.environ, "ALTHING_HANDLE": "booth-dev",
             "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"})
    assert r.returncode == 0, r.stderr
    why = _manifest(tmp_path / "b").why
    assert why == 'he said "pick v3" — line1 line2 · ünï'


def test_a_flag_with_no_value_does_not_eat_the_booth_name(tmp_path):
    """`booth new b --why` with nothing after it must not consume `b` as the
    value and then create a booth called nothing. Usage, and no directory."""
    r = subprocess.run([str(SCRIPT), "new", "b", "--why"], capture_output=True,
                       text=True, timeout=30,
                       env={**os.environ, "BOOTH_DATA_DIR": str(tmp_path),
                            "BOOTH_URL": "http://booth.invalid"})
    assert r.returncode == 2
    assert "usage:" in r.stderr
    assert not (tmp_path / "b").exists()


def test_a_bare_add_does_not_wipe_the_why_the_new_set(tmp_path):
    """`booth new x --why "..."` then `booth add x out/*.png` is THE sequence,
    and the second call must not erase the first one's sentence. The module
    distinguishes omitted from empty; the shell has to carry that distinction
    across, which means an UNSET variable, not an empty one."""
    env = {**os.environ, "ALTHING_HANDLE": "booth-dev",
           "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"}
    src = tmp_path / "a.png"
    src.write_bytes(b"x")

    subprocess.run([str(SCRIPT), "new", "b", "--why", "pick the denoiser",
                    "--title", "R18 A/B"],
                   check=True, capture_output=True, timeout=30, env=env)
    subprocess.run([str(SCRIPT), "add", "b", str(src)],
                   check=True, capture_output=True, timeout=30, env=env)

    m = _manifest(tmp_path / "b")
    assert m.why == "pick the denoiser", "a bare `booth add` wiped the why"
    assert m.title == "R18 A/B"


def test_an_explicitly_empty_why_still_clears_it(tmp_path):
    """Omitted means unchanged; supplied-and-empty means the poster meant to
    take it back. Both have to be reachable from the shell."""
    env = {**os.environ, "ALTHING_HANDLE": "booth-dev",
           "BOOTH_DATA_DIR": str(tmp_path), "BOOTH_URL": "http://booth.invalid"}
    subprocess.run([str(SCRIPT), "new", "b", "--why", "wrong"], check=True,
                   capture_output=True, timeout=30, env=env)
    subprocess.run([str(SCRIPT), "new", "b", "--why", ""], check=True,
                   capture_output=True, timeout=30, env=env)

    assert _manifest(tmp_path / "b").why == ""


def test_answer_and_marks_agree_about_what_open_means(tmp_path):
    """U2 made `_is_open` THE openness predicate — "nothing else may spell this
    out" — and `booth answer`'s reader spelled it out anyway, as
    `if m.answer is None`. So a PARTIALLY answered pick read as done to
    `answer` and still-open to `marks --wait`: one verb returns the half-filled
    form and the other blocks on the same booth at the same instant.

    Found 2/4. The two verbs are the session's whole view of the loop, and a
    session that asks both gets two answers.
    """
    import sys
    sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
    from booth.marks import answer_pick, declare_pick

    b = tmp_path / "b"
    b.mkdir()
    declare_pick(b, "batch", {
        "title": "R18",
        "questions": [
            {"key": "q1", "prompt": "One?", "options": ["keep", "drop"]},
            {"key": "q2", "prompt": "Two?", "options": ["keep", "drop"]},
        ],
    })
    answer_pick(b, "batch", {"q1": "keep", "q2": None})      # partial

    env = {**os.environ, "BOOTH_DATA_DIR": str(tmp_path),
           "BOOTH_URL": "http://booth.invalid"}
    marks = subprocess.run([str(SCRIPT), "marks", "b"], capture_output=True,
                           text=True, timeout=30, env=env)
    answer = subprocess.run([str(SCRIPT), "answer", "b", "batch"],
                            capture_output=True, text=True, timeout=30, env=env)

    still_open = "batch" in json.loads(marks.stdout)["open"]
    assert still_open, "a partial answer stopped counting as open"
    assert answer.returncode == UNANSWERED, (
        "`answer` called a partially-answered pick done while `marks` called it open"
    )


def test_answer_does_not_poll_forever_on_a_pick_that_cannot_be_answered(tmp_path):
    """The mirror failure. A pick whose declaration went bad hydrates with
    `error` set, which makes it NOT open — so `marks --wait` returns at once
    while `answer --wait` polled the full hour against a form the web route
    refuses with a 400. Nothing was ever going to land."""
    b = tmp_path / "b"
    b.mkdir()
    (b / ".marks.json").write_text(json.dumps({
        "version": 1,
        "marks": [{"id": "broken", "shape": "pick", "declaration": {},
                   "error": "pick has no declaration",
                   "created": "2026-09-21T00:00:00.000000+00:00"}],
    }))

    r = subprocess.run([str(SCRIPT), "answer", "b", "broken", "--wait", "8"],
                       capture_output=True, text=True, timeout=40,
                       env={**os.environ, "BOOTH_DATA_DIR": str(tmp_path),
                            "BOOTH_URL": "http://booth.invalid"})
    assert r.returncode != 0
    assert "broken" in r.stderr.lower() or "cannot" in r.stderr.lower()