ersatztv/scripts/tests/test_pr_changed_files.py

"""Tests for `scripts/pr-changed-files.sh` — the SHARED PR file enumeration (ersatztv#649).

Why this file exists. The enumeration used to be written twice: once in
`.claude/hooks/pretooluse-merge-consent.sh` (ADVISORY — a failure produces a human prompt) and once
inline in `.gitea/workflows/review-verdict.yml` (ENFORCED — it writes the branch-protection-required
`review-verdict/h10` status). They drifted, and in the dangerous direction: four rounds of
ersatztv#643 hardening landed on the advisory copy and never reached the enforced one, so the copy
with real authority ended up strictly weaker than the copy without.

The specific thing this suite pins is the point of ersatztv#649's second Done-when box. A round-4
review traced that the enforced copy's fail-closed behaviour on a garbage response was INCIDENTAL,
not designed: `n` came back empty, `[ "$n" -lt 50 ]` errored to false, the loop ran to MAX_PAGES and
left complete=no. The right answer, reached through a bash arithmetic error that any refactor of the
loop could have silently flipped. Every failure-path test below therefore asserts a NON-ZERO exit
explicitly, so the behaviour is a contract rather than a coincidence.

Observable contract of the script:
  exit 0  -> enumeration complete and bound to the expected head; stdout is the authoritative path set
  exit 1  -> could not enumerate/verify; stdout meaningless, caller MUST withhold any exemption
  exit 2  -> usage error
"""

from __future__ import annotations

import json
import os
import re
import subprocess
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parents[2]
SCRIPT = REPO_ROOT / "scripts" / "pr-changed-files.sh"
HOOK = REPO_ROOT / ".claude" / "hooks" / "pretooluse-merge-consent.sh"
WORKFLOW = REPO_ROOT / ".gitea" / "workflows" / "review-verdict.yml"

SHA = "a9e3e23abf337980ca4c05854f5b1e210099d08b"
OTHER_SHA = "b71c0d4e2f8a91b3c5d7e9f1a3b5c7d9e1f3a5b7"

# Serves paged `pulls/N/files`, plus the PR object the script re-reads to bind the enumeration.
CURL_SHIM = r'''#!/usr/bin/env python3
import json, os, sys, pathlib, urllib.parse

state = pathlib.Path(os.environ["STUB_DIR"])
args = sys.argv[1:]
url = [a for a in args if a.startswith("http")][-1]

if "/pulls/" in url and "/files" in url:
    q = urllib.parse.parse_qs(urllib.parse.urlparse(url).query)
    page = int(q.get("page", ["1"])[0])
    pages = json.loads((state / "pages.json").read_text())
    if page > len(pages):
        print("[]"); sys.exit(0)
    entry = pages[page - 1]
    if entry == "ERROR":          # transport failure on this page
        sys.exit(22)
    if entry == "GARBAGE":        # 200 with a non-array body (proxy/error page)
        print('{"message":"internal error"}'); sys.exit(0)
    print(json.dumps(entry)); sys.exit(0)

if "/pulls/" in url:
    # A second head sha served from the Nth PR-object read onward models a force-push landing
    # between pagination round-trips.
    sha = os.environ["STUB_SHA"]
    alt = state / "pr_sha_after.txt"
    if alt.exists():
        sha = alt.read_text().strip()
    print(json.dumps({"head": {"sha": sha}}))
    sys.exit(0)

print("{}")
'''


def _rows(paths, status="modified"):
    return [{"filename": p, "status": status} for p in paths]


@pytest.fixture
def enumerate_files(tmp_path):
    bindir = tmp_path / "bin"; bindir.mkdir()
    shim = bindir / "curl"; shim.write_text(CURL_SHIM); shim.chmod(0o755)
    state = tmp_path / "state"; state.mkdir()

    env = dict(os.environ)
    env["PATH"] = f"{bindir}{os.pathsep}{env['PATH']}"
    env["STUB_DIR"] = str(state)
    env["STUB_SHA"] = SHA
    env["ETV_GITEA_TOKEN"] = "stub"
    env["ETV_GITEA_URL"] = "http://gitea.example"
    env.pop("ETV_GITEA_BASICAUTH", None)
    env.pop("GITEA_TOKEN", None)

    class Handle:
        def __init__(self, env_, state_):
            self.env = env_
            self.state = state_

        def set_pages(self, *pages):
            (state / "pages.json").write_text(json.dumps(list(pages)))

        def head_moves_to(self, sha):
            (state / "pr_sha_after.txt").write_text(sha)

        def run(self, expected_sha=SHA, args=("timothy", "ersatztv", "42")):
            return subprocess.run(
                ["bash", str(SCRIPT), *args, expected_sha],
                env=self.env, capture_output=True, text=True)

        def paths(self):
            """Assert success and return the enumerated path set."""
            r = self.run()
            assert r.returncode == 0, f"expected success, got {r.returncode}: {r.stderr}"
            return [ln for ln in r.stdout.splitlines() if ln.strip()]

        def fails_closed(self):
            """The whole point: a NON-ZERO exit, asserted, not inferred."""
            r = self.run()
            return r.returncode != 0

    return Handle(env, state)


# --- The happy path, so the failure-path assertions below cannot pass vacuously ----------------

def test_complete_enumeration_returns_every_path(enumerate_files):
    enumerate_files.set_pages(_rows(["docs/a.md", "ErsatzTV/Program.cs", "README.md"]))
    assert enumerate_files.paths() == ["docs/a.md", "ErsatzTV/Program.cs", "README.md"]


def test_a_rename_contributes_BOTH_sides(enumerate_files):
    """One row, two paths — the `git mv` hole. Reading `.filename` alone hides the source."""
    enumerate_files.set_pages([{"filename": "docs/innocuous-note.md", "status": "renamed",
                                "previous_filename": ".gitea/workflows/renovate.yml"}])
    assert sorted(enumerate_files.paths()) == [".gitea/workflows/renovate.yml",
                                               "docs/innocuous-note.md"]


def test_paths_on_a_later_page_are_included(enumerate_files):
    enumerate_files.set_pages(_rows([f"docs/f{i}.md" for i in range(50)]),
                              _rows(["scripts/decisions_lib.py"]))
    assert "scripts/decisions_lib.py" in enumerate_files.paths()


# --- Fail-closed contract: each of these MUST be non-zero, by design ---------------------------

def test_transport_failure_mid_pagination_fails_closed(enumerate_files):
    """The defect that started all of this: an errored page counted as zero rows and read as
    'end of list', completing the enumeration over a PARTIAL list."""
    enumerate_files.set_pages(_rows([f"docs/f{i}.md" for i in range(50)]), "ERROR",
                              _rows(["docs/tail.md"]))
    assert enumerate_files.fails_closed()


def test_non_array_body_fails_closed(enumerate_files):
    enumerate_files.set_pages("GARBAGE")
    assert enumerate_files.fails_closed()


def test_an_OBJECT_OF_VALID_ROWS_isolates_the_top_level_array_check(enumerate_files):
    """Cold review found `test_non_array_body_fails_closed` passing for the wrong reason, and the
    first attempt to fix it failed for a THIRD reason — worth recording, because both near-misses
    look like coverage.

    `jq`'s `all(.[]; …)` iterates an object's VALUES, so the top-level `type == "array"` check is
    only load-bearing when those values would themselves validate:

      * `{"message":"internal error"}` — values are strings, `.filename` on a string errors. Rejected
        with the clause deleted, so it never isolated it.
      * `{"filename":"docs/a.md","status":"modified"}` — a single row, but its values are still
        strings. Same non-isolation, one level less obvious.
      * `{"0": {"filename":"docs/a.md","status":"modified"}}` — values ARE valid rows. With the clause
        deleted this validates, `length` is 1, the path is collected, and the next page ends the
        enumeration cleanly: a non-array body enumerated as a complete docs-only list. That is the
        fail-open, and only this shape exposes it.
    """
    enumerate_files.set_pages({"0": {"filename": "docs/a.md", "status": "modified"}})
    assert enumerate_files.fails_closed()


def test_rows_without_filename_fail_closed(enumerate_files):
    """`[{}]` is a well-formed array that yields no paths — a partial list wearing a valid shape."""
    enumerate_files.set_pages([{}, {}])
    assert enumerate_files.fails_closed()


def test_a_row_with_a_VALID_status_but_no_filename_isolates_the_filename_check(enumerate_files):
    """Same wrong-reason problem: `[{}]` is also rejected by the closed `.status` allow-list, since
    an absent status is not in it. A row carrying a legitimate `status` and no `filename` removes
    that second reason, leaving only the guard under test."""
    enumerate_files.set_pages([{"status": "modified"}])
    assert enumerate_files.fails_closed()


def test_array_of_scalars_fails_closed(enumerate_files):
    enumerate_files.set_pages(["docs/a.md", "docs/b.md"])
    assert enumerate_files.fails_closed()


@pytest.mark.parametrize("evil", ["safe.md\ndocs/Program.cs", "safe.md\rdocs/Program.cs"])
def test_CRLF_in_filename_fails_closed(enumerate_files, evil):
    """A newline splits one path into two lines that are each allow-list-matched separately, so
    `safe.md\\ndocs/Program.cs` reads as two exempt paths while the real path ends in .cs."""
    enumerate_files.set_pages(_rows([evil]))
    assert enumerate_files.fails_closed()


def test_CRLF_in_previous_filename_on_a_NON_renamed_row_fails_closed(enumerate_files):
    """The hole one predicate wide: `previous_filename` is CONSUMED on every row, so it must be
    VALIDATED on every row — not only where `.status == "renamed"` makes it semantically expected."""
    enumerate_files.set_pages([{"filename": "docs/a.md", "status": "modified",
                                "previous_filename": "safe.md\ndocs/Program.cs"}])
    assert enumerate_files.fails_closed()


def test_dotdot_path_component_fails_closed(enumerate_files):
    """The callers' allow-lists anchor `^docs/`, which `docs/../ErsatzTV/Program.cs` matches."""
    enumerate_files.set_pages(_rows(["docs/../ErsatzTV/Program.cs"]))
    assert enumerate_files.fails_closed()


@pytest.mark.parametrize("status", ["Renamed", "moved", ""])
def test_status_outside_the_closed_allow_list_fails_closed(enumerate_files, status):
    """Without a closed set, the `renamed => previous_filename REQUIRED` clause is dodgeable by any
    other value, letting a `git mv` drop its source path and read as docs-only."""
    enumerate_files.set_pages([{"filename": "docs/a.md", "status": status,
                                "previous_filename": "ErsatzTV/Program.cs"}])
    assert enumerate_files.fails_closed()


def test_renamed_row_without_previous_filename_fails_closed(enumerate_files):
    enumerate_files.set_pages([{"filename": "docs/a.md", "status": "renamed"}])
    assert enumerate_files.fails_closed()


def test_head_differing_from_the_expected_sha_fails_closed(enumerate_files):
    """Narrowed to what this actually proves, per cold review.

    The stub serves the alternate sha from the FIRST PR-object read, and the script reads that
    object only once, after paging — so this exercises "the final head does not equal the expected
    sha", not movement *during* enumeration. The distinction matters: an A->B->A force-push round
    trip would restore the expected sha and pass this check while the pages came from two different
    states. That race is inherent to enumerating a mutable list over several round-trips against an
    API with no commit-pinned files endpoint, and is tracked separately rather than papered over
    with a test name that implies it is covered.
    """
    enumerate_files.set_pages(_rows(["docs/a.md"]))
    enumerate_files.head_moves_to(OTHER_SHA)
    assert enumerate_files.fails_closed()


def test_missing_credentials_fails_closed(enumerate_files):
    enumerate_files.set_pages(_rows(["docs/a.md"]))
    enumerate_files.env.pop("ETV_GITEA_TOKEN", None)
    assert enumerate_files.fails_closed()


@pytest.mark.parametrize("args", [("timothy", "ersatztv"), ("timothy", "ersatztv", "")])
def test_usage_errors_exit_2(enumerate_files, args):
    enumerate_files.set_pages(_rows(["docs/a.md"]))
    r = enumerate_files.run(args=args) if len(args) == 3 else subprocess.run(
        ["bash", str(SCRIPT), *args], env=enumerate_files.env, capture_output=True, text=True)
    assert r.returncode == 2, r.stderr


def test_a_SHORT_page_does_not_end_the_enumeration(enumerate_files):
    """Gitea caps `limit` at the server-wide MAX_RESPONSE_ITEMS and may return fewer than asked.
    'Fewer than 50 rows means last page' would complete over a partial list without any transport
    error — so termination requires a validated EMPTY page. A 30-row page followed by code must be
    seen."""
    enumerate_files.set_pages(_rows([f"docs/f{i}.md" for i in range(30)]),
                              _rows(["ErsatzTV/Program.cs"]))
    assert "ErsatzTV/Program.cs" in enumerate_files.paths()


# --- The same guard, under the runner's jq 1.6 ------------------------------------------------
#
# `test_transport_failure_mid_pagination_fails_closed` above does NOT isolate the explicit
# `if [ -z "${raw//[[:space:]]/}" ]` clause — cold review claimed this and mutation confirmed it:
# deleting that clause leaves the whole suite green on a developer Mac, because jq 1.8 rejects empty
# input on its own. jq 1.6 does not, and the runner ships 1.6 — so the one environment where the
# clause is load-bearing was the one environment with no coverage. That is the #643/#647 failure
# class exactly, reproduced in the test suite meant to prevent it.
#
# The shim is IMPORTED, not copied. A second copy of a version-quirk emulator is the same
# two-implementations-drift problem this whole issue is about, one level down.
from scripts.tests.test_merge_consent_exemption import _JQ16_SHIM  # noqa: E402


@pytest.fixture
def enumerate_files_jq16(enumerate_files, tmp_path):
    """The same harness, plus a jq shim reproducing jq 1.6's empty-input `-e` exit status."""
    jq = tmp_path / "bin" / "jq"
    jq.write_text(_JQ16_SHIM)
    jq.chmod(0o755)
    return enumerate_files


def test_jq16_shim_actually_reproduces_the_quirk(enumerate_files_jq16, tmp_path):
    """Verify the verifier. A shim that failed to install would make the test below pass vacuously,
    reporting the guard safe on jq 1.6 without ever exercising the quirk."""
    assert enumerate_files_jq16 is not None      # the fixture is what installs the shim
    jq = str(tmp_path / "bin" / "jq")
    empty = subprocess.run([jq, "-e", "."], input="", capture_output=True, text=True)
    assert empty.returncode == 0, "the shim does not reproduce jq 1.6's empty-input exit 0"
    real = subprocess.run([jq, "-e", ".a"], input='{"a":1}', capture_output=True, text=True)
    assert real.returncode == 0 and real.stdout.strip() == "1", "the shim broke ordinary jq"
    false = subprocess.run([jq, "-e", ".a"], input='{"a":false}', capture_output=True, text=True)
    assert false.returncode == 1, "the shim broke jq's real -e semantics for a false result"


def test_transport_failure_mid_pagination_fails_closed_on_jq16(enumerate_files_jq16):
    """The property, asserted on the interpreter that actually runs it in CI."""
    enumerate_files_jq16.set_pages(_rows([f"docs/f{i}.md" for i in range(50)]), "ERROR",
                                   _rows(["docs/tail.md"]))
    assert enumerate_files_jq16.fails_closed()


def test_a_docs_only_pr_still_enumerates_cleanly_on_jq16(enumerate_files_jq16):
    """Positive control: the shim must not make everything fail, or the test above proves nothing."""
    enumerate_files_jq16.set_pages(_rows(["docs/a.md", "docs/b.md"]))
    assert enumerate_files_jq16.paths() == ["docs/a.md", "docs/b.md"]


# --- The CALLER contract: a non-zero exit must withhold the exemption, on its own -------------
#
# This is the single line the whole extraction rests on, and it was the one guard nothing pinned. A
# review mutated the hook's `if files=$(...)` into `files=$(...) || true; files_complete=yes` — i.e.
# ignore the exit status entirely — and the ENTIRE suite still passed.
#
# It survives today only by REDUNDANCY: the script writes stdout once, immediately before `exit 0`,
# so every failure path also happens to yield empty stdout, and the hook's independent
# `[ -n "$files" ]` check catches it. That is exactly the shape this PR criticises elsewhere — safe
# by accident rather than by assertion. Any future change that streams pages, or prints a partial
# list before failing, turns it into a live false exemption.
#
# So the stub below FAILS *while emitting a perfectly docs-only list*, which is the one combination
# the redundancy cannot absorb. The positive control immediately after it proves the harness can
# actually observe the difference, rather than reporting "not exempt" for some unrelated reason.

HOOK_STUB_CURL = r'''#!/usr/bin/env python3
import json, sys
url = [a for a in sys.argv[1:] if a.startswith("http")][-1]
if "/pulls/" in url and "/files" not in url:
    print(json.dumps({"head": {"sha": "%s"}, "body": "no linked issue"}))
else:
    print("{}")
''' % SHA


def _mirror_tree(tmp_path, stub_body):
    """Lay out a minimal repo mirror so the hook resolves OUR stub as the shared script.

    The hook finds the script via `${BASH_SOURCE[0]}/../..`, so the mirror must reproduce the real
    `.claude/hooks/` + `scripts/` shape rather than just dropping the stub anywhere.
    """
    hooks = tmp_path / ".claude" / "hooks"
    hooks.mkdir(parents=True)
    (hooks / "pretooluse-merge-consent.sh").write_text(HOOK.read_text())
    scripts = tmp_path / "scripts"
    scripts.mkdir()
    stub = scripts / "pr-changed-files.sh"
    stub.write_text(stub_body)
    stub.chmod(0o755)

    bindir = tmp_path / "bin"; bindir.mkdir()
    curl = bindir / "curl"; curl.write_text(HOOK_STUB_CURL); curl.chmod(0o755)

    env = dict(os.environ)
    env["PATH"] = f"{bindir}{os.pathsep}{env['PATH']}"
    env["ETV_GITEA_TOKEN"] = "stub"
    env["ETV_GITEA_URL"] = "http://gitea.example"
    env.pop("ETV_GITEA_BASICAUTH", None)

    payload = {"tool_input": {"method": "merge", "owner": "timothy",
                              "repo": "ersatztv", "pull_number": 42}}
    r = subprocess.run(["bash", str(hooks / "pretooluse-merge-consent.sh")],
                       input=json.dumps(payload), env=env, capture_output=True, text=True)
    assert r.returncode == 0, r.stderr
    # Exempt == passthrough == exit 0 with no decision JSON on stdout.
    return r.stdout.strip() == ""


DOCS_ONLY_OUTPUT = 'printf "docs/a.md\\ndocs/b.md\\n"\n'


def test_a_FAILING_script_withholds_the_exemption_even_when_stdout_looks_docs_only(tmp_path):
    exempt = _mirror_tree(tmp_path, "#!/usr/bin/env bash\n" + DOCS_ONLY_OUTPUT + "exit 1\n")
    assert exempt is False, (
        "the hook granted a docs-only exemption from the stdout of a script that FAILED — the exit "
        "status is not being checked")


def test_positive_control_the_same_output_with_exit_0_DOES_exempt(tmp_path):
    """Without this, the test above could pass for any unrelated reason and prove nothing."""
    exempt = _mirror_tree(tmp_path, "#!/usr/bin/env bash\n" + DOCS_ONLY_OUTPUT + "exit 0\n")
    assert exempt is True, (
        "the positive control failed, so the negative test above cannot be trusted to be measuring "
        "the exit status at all")


def test_a_script_that_crashes_also_withholds_the_exemption(tmp_path):
    """Not every failure is a clean `exit 1` — a crash must not read as success either."""
    exempt = _mirror_tree(
        tmp_path, "#!/usr/bin/env bash\n" + DOCS_ONLY_OUTPUT + "kill -TERM $$\n")
    assert exempt is False


def test_a_MISSING_script_withholds_the_exemption(tmp_path):
    """The transition window: the shared checkout has no such script until this lands."""
    hooks = tmp_path / ".claude" / "hooks"
    hooks.mkdir(parents=True)
    (hooks / "pretooluse-merge-consent.sh").write_text(HOOK.read_text())
    (tmp_path / "scripts").mkdir()
    bindir = tmp_path / "bin"; bindir.mkdir()
    curl = bindir / "curl"; curl.write_text(HOOK_STUB_CURL); curl.chmod(0o755)
    env = dict(os.environ)
    env["PATH"] = f"{bindir}{os.pathsep}{env['PATH']}"
    env["ETV_GITEA_TOKEN"] = "stub"
    env["ETV_GITEA_URL"] = "http://gitea.example"
    payload = {"tool_input": {"method": "merge", "owner": "timothy",
                              "repo": "ersatztv", "pull_number": 42}}
    r = subprocess.run(["bash", str(hooks / "pretooluse-merge-consent.sh")],
                       input=json.dumps(payload), env=env, capture_output=True, text=True)
    assert r.returncode == 0, r.stderr
    assert r.stdout.strip() != "", "a missing shared script must not grant an exemption"


# --- Drift guard: the reason this file is worth having at all ----------------------------------
#
# Scope is BOTH callers. Until the enforced workflow was rewired (the follow-up half of ersatztv#649)
# this guard could only assert the hook, which left the copy with real authority — the one that
# writes the branch-protection-required `review-verdict/h10` status — unpinned. That asymmetry was
# the entire subject of the issue, so a guard that covered only the advisory side would have been
# the same mistake one level up.

CALLERS = pytest.mark.parametrize(
    "caller", [HOOK, WORKFLOW], ids=["advisory-hook", "enforced-workflow"])


def _code_lines(path: Path) -> str:
    """Strip comment lines.

    Both callers' comments legitimately discuss the `pulls/N/files` endpoint, and a future comment
    writing `pulls/$pr/files?limit=100` as an example of what NOT to do would redden these tests —
    which, since `script-tests` red blocks merges via the combined status, would block the repo over
    a piece of prose. `#`-prefixed works for both files: YAML comments and the shell comments inside
    the workflow's `run:` block share the marker.
    """
    return "\n".join(
        ln for ln in path.read_text().splitlines() if not ln.lstrip().startswith("#"))


@CALLERS
def test_the_caller_uses_the_shared_script(caller):
    assert "scripts/pr-changed-files.sh" in _code_lines(caller), (
        f"{caller.relative_to(REPO_ROOT)} no longer calls the shared enumeration")


@CALLERS
def test_the_caller_does_not_reimplement_the_enumeration(caller):
    """Structural rather than behavioural on purpose. Behavioural equivalence tests would still pass
    if someone pasted the loop back inline and kept it correct *that day* — which is exactly how the
    drift happened the first time. What must be prevented is a SECOND implementation existing.
    """
    # An inline `pulls/<n>/files?` fetch is the signature of a re-inlined copy. Match on the endpoint
    # alone, NOT on `?limit=` — an earlier version anchored the query string, so a copy written as
    # `files?page=1&limit=50` would have walked straight past a guard that exists to stop exactly
    # that. Still evadable by a copy that builds the URL without a literal `?`, so this narrows the
    # gap rather than closing it.
    assert not re.search(r"pulls/\$?\{?\w+\}?/files\?", _code_lines(caller)), (
        f"{caller.relative_to(REPO_ROOT)} appears to enumerate PR files inline again — "
        "that is the duplication ersatztv#649 removed")


# --- The ENFORCED caller's own preconditions ---------------------------------------------------

def _workflow_steps():
    import yaml
    wf = yaml.safe_load(WORKFLOW.read_text())
    return wf["jobs"]["set-verdict-status"]["steps"]


def _classify_step():
    for step in _workflow_steps():
        if "review-verdict/h10" in (step.get("run") or ""):
            return step
    raise AssertionError("no step in review-verdict.yml posts review-verdict/h10")


def test_the_workflow_checks_out_the_BASE_ref_never_the_head():
    """The self-rewrite hole. This workflow judges the PR, so the PR must not supply the code that
    judges it: checking out `head.sha` would let a PR rewrite `scripts/pr-changed-files.sh` to
    return an empty list and exempt itself.

    Asserted through a YAML parse rather than a substring search, because the dangerous edit is
    subtle — `head.sha` for `base.sha` is a nine-character diff in a `with:` block — and a
    text-level `"base.sha" in text` check would still pass if a second checkout step took the head
    afterwards and won.
    """
    checkouts = [s for s in _workflow_steps() if "actions/checkout" in (s.get("uses") or "")]
    assert len(checkouts) == 1, (
        f"expected exactly one checkout step, found {len(checkouts)} — a second checkout can "
        "silently overwrite the base ref with the PR head")
    with_ = checkouts[0].get("with") or {}
    ref = str(with_.get("ref", ""))
    assert "pull_request.base.sha" in ref, (
        f"the checkout ref is {ref!r}; it must be the PR's BASE sha, so a PR cannot rewrite the "
        "gate that judges it")
    assert "head" not in ref, f"the checkout ref {ref!r} references the PR head"
    assert with_.get("persist-credentials") is False, (
        "persist-credentials must be false — nothing here pushes, and a token left in .git/config "
        "is handed to every script the job runs")


def test_the_workflow_runs_the_jq_preflight_in_FLOOR_mode_only():
    """`--expect` pins an exact jq version and fails when it drifts. That is right for the advisory
    `script-tests` job and catastrophic here: this workflow writes `review-verdict/h10`, a REQUIRED
    check on `main`, so a pin would turn any jq upgrade on the runner into a repo-wide merge
    deadlock — a required gate failing because an upstream package manager did its job.
    """
    # CODE only, for the reason `_code_lines` documents: the first draft of this assertion read the
    # raw text and went red on the workflow's own comment explaining why `--expect` is banned here.
    code = _code_lines(WORKFLOW)
    # A bare `"jq-preflight.sh" in code` is NOT enough, and cold review was right to say so: the
    # path also appears in the `if [ -x ./scripts/jq-preflight.sh ]` presence guard, so deleting the
    # actual invocation would leave that substring behind and the assertion green. Require a line
    # that INVOKES it.
    steps = [s for s in _workflow_steps() if "jq-preflight.sh" in (s.get("run") or "")]
    assert steps, (
        "review-verdict.yml no longer runs the jq preflight, so the version its shell gates run "
        "under is unobservable again (ersatztv#648)")
    invocations = [ln.strip() for ln in steps[0]["run"].splitlines()
                   if re.match(r"^\s*(\./)?scripts/jq-preflight\.sh(\s|$)", ln)]
    assert invocations, "the jq preflight is referenced but never actually invoked"
    assert "--expect" not in code, (
        "review-verdict.yml must run jq-preflight.sh in floor-only mode; --expect here deadlocks "
        "every merge on `main` the day the runner's jq changes")
    assert all("--expect" not in ln for ln in invocations)


# --- The ENFORCED caller's contract, EXECUTED ---------------------------------------------------
#
# The structural guards above prove the workflow *calls* the shared script. They cannot prove it
# reacts correctly when the script FAILS — and that is precisely the mutation that survived the last
# round on the hook side: making the caller ignore the exit status left the entire suite green,
# because every failure path also happened to produce empty stdout. So the stub below FAILS while
# emitting a perfectly docs-only list, the one combination that redundancy cannot absorb.
#
# The step's `run:` block is extracted from the YAML and executed directly. That is a real
# behavioural test of the shipped text — not a paraphrase of it — at the cost of not exercising the
# runner's step wiring, which no local test can reach anyway.

WORKFLOW_STUB_CURL = r'''#!/usr/bin/env python3
import json, os, pathlib, sys

args = sys.argv[1:]
url = [a for a in args if a.startswith("http")][-1]
out = pathlib.Path(os.environ["STUB_DIR"])

if "-X" in args and args[args.index("-X") + 1] == "POST":
    payload = args[args.index("-d") + 1]
    (out / "posted.json").write_text(payload)
    (out / "posted_url.txt").write_text(url)
    print("{}")
    sys.exit(0)

if "/status" in url:
    # Configurable. Hardcoding "no verdict yet" left the ersatztv#647 emptiness guard and the
    # never-overwrite short-circuit unreachable: neither could be made to fire, so mutations
    # deleting them survived the whole suite.
    mode = os.environ.get("STUB_STATUS_MODE", "none")
    if mode == "transport-error":
        # Real `gh()` is `curl -sf`: an HTTP error exits 22 with EMPTY stdout.
        sys.exit(22)
    if mode == "garbage":
        print("<html>502 Bad Gateway</html>")
        sys.exit(0)
    if mode.startswith("existing:"):
        print(json.dumps({"statuses": [
            {"context": "review-verdict/h10", "status": mode.split(":", 1)[1]}]}))
        sys.exit(0)
    print(json.dumps({"statuses": []}))
    sys.exit(0)

print("{}")
'''


def _run_classify(tmp_path, enum_stub: str | None, author: str = "timothy",
                  status_mode: str = "none"):
    """Execute the workflow's classify `run:` block with a stubbed enumeration script.

    Returns the status payload the job POSTed, or None if it posted nothing.
    """
    bindir = tmp_path / "bin"; bindir.mkdir()
    curl = bindir / "curl"; curl.write_text(WORKFLOW_STUB_CURL); curl.chmod(0o755)
    scripts = tmp_path / "scripts"; scripts.mkdir()
    if enum_stub is not None:
        enum = scripts / "pr-changed-files.sh"
        enum.write_text(enum_stub)
        enum.chmod(0o755)

    env = dict(os.environ)
    env["PATH"] = f"{bindir}{os.pathsep}{env['PATH']}"
    env["STUB_DIR"] = str(tmp_path)
    env["STUB_STATUS_MODE"] = status_mode
    env.update({
        "GITEA_TOKEN": "stub",
        "BASE_URL": "http://gitea.example/api/v1",
        "GITEA_BASE_URL": "http://gitea.example/api/v1",
        "REPO": "timothy/ersatztv",
        "PR": "42",
        "SHA": SHA,
        "BASE_SHA": OTHER_SHA,
        "AUTHOR": author,
        "PR_URL": "http://gitea.example/timothy/ersatztv/pulls/42",
    })

    script = tmp_path / "step.sh"
    script.write_text(_classify_step()["run"])
    r = subprocess.run(["bash", str(script)], cwd=tmp_path, env=env,
                       capture_output=True, text=True)
    # Assert the WIRING, not only the classification. The stub accepts every POST, so a status aimed
    # at the wrong endpoint, sha, host or repo would otherwise leave these tests green while the real
    # required check was never written.
    #
    # Two ways this check could disable itself, both found by cold review of an earlier draft:
    #   * it was guarded by `if url_file.exists()`, so deleting the recorder in the stub turned it
    #     into a no-op and every test stayed green — a verifier that silently opts out;
    #   * it compared only the URL SUFFIX, so a POST to the right path on the WRONG HOST OR REPO
    #     passed. Compare the whole URL against the env this job was given.
    posted = tmp_path / "posted.json"
    url_file = tmp_path / "posted_url.txt"
    if posted.exists():
        assert url_file.exists(), (
            "a status was POSTed but its URL was not recorded — the wiring assertion below would "
            "have silently skipped")
        expected = f"{env['BASE_URL']}/repos/{env['REPO']}/statuses/{SHA}"
        assert url_file.read_text().strip() == expected, (
            f"the status was POSTed to {url_file.read_text().strip()!r}, expected {expected!r}")
    return (json.loads(posted.read_text()) if posted.exists() else None), r


DOCS_ONLY = 'printf "docs/a.md\\ndocs/b.md\\n"\n'


def test_a_FAILING_enumeration_withholds_the_exemption_even_when_stdout_looks_docs_only(tmp_path):
    """The mutation that previously survived: ignore the exit status, trust stdout."""
    posted, r = _run_classify(tmp_path, "#!/usr/bin/env bash\n" + DOCS_ONLY + "exit 1\n")
    assert posted is not None, f"the job posted no status at all: {r.stderr}"
    assert posted["state"] == "pending", (
        f"a docs-only exemption was granted from the stdout of a script that FAILED "
        f"(state={posted['state']}, desc={posted['description']!r}) — the exit status is not being "
        "checked")


def test_workflow_positive_control_the_same_output_with_exit_0_DOES_exempt(tmp_path):
    """Without this, the test above could pass for any unrelated reason and prove nothing."""
    posted, r = _run_classify(tmp_path, "#!/usr/bin/env bash\n" + DOCS_ONLY + "exit 0\n")
    assert posted is not None, f"the job posted no status at all: {r.stderr}"
    assert posted["state"] == "success", (
        "the positive control failed, so the negative test above cannot be trusted to be measuring "
        f"the exit status at all (state={posted['state']}, desc={posted['description']!r})")


def test_a_crashing_enumeration_also_withholds_the_exemption(tmp_path):
    posted, _ = _run_classify(tmp_path, "#!/usr/bin/env bash\n" + DOCS_ONLY + "kill -TERM $$\n")
    assert posted is not None
    assert posted["state"] == "pending"


def test_a_MISSING_enumeration_script_withholds_the_exemption(tmp_path):
    """A PR whose BASE predates the script's introduction. It must post an actionable `pending`
    rather than dying with no status — an absent required check blocks the merge either way, but a
    stalled PR with no explanation is how a gate acquires a reputation for being flaky.
    """
    posted, r = _run_classify(tmp_path, None)
    assert posted is not None, f"the job posted no status at all: {r.stderr}"
    assert posted["state"] == "pending"


def _emitting(*paths):
    body = "".join(f'printf "%s\\n" "{p}"\n' for p in paths)
    return "#!/usr/bin/env bash\n" + body + "exit 0\n"


def test_a_code_file_defeats_the_docs_only_exemption(tmp_path):
    """The workflow's own classification — deliberately NOT shared with the hook, whose allow-list
    is wider because a match there falls through to a human prompt rather than a green status.
    """
    posted, _ = _run_classify(tmp_path, _emitting("docs/a.md", "ErsatzTV/Program.cs"))
    assert posted is not None
    assert posted["state"] == "pending"


def test_a_BOT_pr_touching_a_protected_path_is_NOT_exempt(tmp_path):
    """`PROTECTED` guards the BOT exemption specifically, and mutation testing is how that got
    stated correctly. The first version of this test used a docs-only+protected file list and
    passed even with the `PROTECTED` clause deleted — `PROTECTED` (`.claude/ .gitea/ .husky/
    scripts/ docker/ci/`) and `DOCS_ONLY` (`docs/`, root `*.md`) are DISJOINT, so on the docs-only
    path that clause can never fire and the `DOCS_ONLY` check was doing all the work. The test
    looked like it covered the self-exemption hole and covered nothing.

    Renovate lands patch bumps unattended via Gitea's own auto-merge, so a bot PR that edits the
    gate, CI, the hooks, or the scripts they call is the one path where an unreviewed change to the
    merge gate could actually merge itself.
    """
    posted, _ = _run_classify(
        tmp_path, _emitting("ErsatzTV/Program.cs", "scripts/pr-changed-files.sh"),
        author="renovate")
    assert posted is not None
    assert posted["state"] == "pending", (
        "a bot PR editing the shared enumeration was auto-exempted — a PR that weakens the merge "
        "gate must never be able to exempt itself from the merge gate")


def test_bot_positive_control_a_plain_bot_pr_IS_exempt(tmp_path):
    """Proves the test above measures `PROTECTED` and not merely 'bot PRs are never exempt'."""
    posted, _ = _run_classify(
        tmp_path, _emitting("Directory.Packages.props"), author="renovate")
    assert posted is not None
    assert posted["state"] == "success", (
        "the bot exemption never fires at all, so the protected-path test above proves nothing")


# --- The STATUS READ: the ersatztv#647 guard and the never-overwrite short-circuit -------------
#
# These were unreachable until the stub's status response became configurable. Four mutations
# survived the full suite without them, including re-introducing the literal ersatztv#647 fail-open.

def test_a_transport_failure_on_the_STATUS_READ_posts_NOTHING(tmp_path):
    """`gh()` is `curl -sf`, so an HTTP error yields exit 22 and EMPTY stdout. Reading that as "no
    verdict exists" would let the job post over a real human verdict. It must fail WITHOUT posting.
    """
    posted, r = _run_classify(tmp_path, _emitting("docs/a.md"), status_mode="transport-error")
    assert r.returncode != 0, "an unreadable status read must fail the job"
    assert posted is None, "nothing may be posted when the existing verdict state is unknown"


def test_a_GARBAGE_status_response_posts_NOTHING(tmp_path):
    """A proxy error page is a 200 with a non-JSON body — not an absent verdict."""
    posted, r = _run_classify(tmp_path, _emitting("docs/a.md"), status_mode="garbage")
    assert r.returncode != 0
    assert posted is None


@pytest.mark.parametrize("existing", ["success", "failure"])
def test_an_existing_verdict_on_this_head_is_NEVER_overwritten(tmp_path, existing):
    """A human verdict for this exact head may already exist — the reviewer ran
    post-review-verdict.sh before this job finished, or the job re-ran. Re-posting would un-approve
    a reviewed head, or (worse) approve one a human marked BLOCKED.

    `failure` is the sharp case: that is a human saying NO, and an exemption posted over it would
    turn a rejection into a merge.
    """
    posted, r = _run_classify(tmp_path, _emitting("ErsatzTV/Program.cs"),
                              status_mode=f"existing:{existing}")
    assert r.returncode == 0, r.stderr
    assert posted is None, f"overwrote an existing '{existing}' verdict on this head"


# --- The `count -eq 0` guard, on the path where it is the ONLY guard ---------------------------

def test_an_EMPTY_enumeration_is_not_exempt_even_for_a_BOT(tmp_path):
    """The author must be a BOT for this to test anything.

    With a non-bot author the blank line an empty list produces already fails `DOCS_ONLY`, so the
    `[ "${count:-0}" -eq 0 ]` guard never decides the outcome — the same short-circuit that made an
    earlier `PROTECTED` test vacuous. On the bot path that guard is the ONLY thing between an
    unreadable-but-successful enumeration and an unattended `success`.

    Verified by mutation: changing `grep -c .` to `grep -c ''` (counting the blank line, so
    `count=1`) grants a bot PR `success` here while every other test stays green.
    """
    posted, r = _run_classify(tmp_path, "#!/usr/bin/env bash\nexit 0\n", author="renovate")
    assert posted is not None, f"the job posted no status at all: {r.stderr}"
    assert posted["state"] == "pending", (
        "an empty file list was treated as a bot exemption — the enumeration returning nothing is "
        "not evidence that nothing was changed")


# --- Anchors in the classifier predicates ------------------------------------------------------

def test_the_BOT_match_is_whole_line_not_substring(tmp_path):
    """`grep -qxF` is anchored; plain `grep -qF` would exempt any author whose name CONTAINS a bot
    name. `ova` is a substring of `renovate`."""
    posted, _ = _run_classify(tmp_path, _emitting("ErsatzTV/Program.cs"), author="ova")
    assert posted is not None
    assert posted["state"] == "pending", "a substring of a bot name was granted the bot exemption"


def test_DOCS_ONLY_anchors_the_markdown_extension(tmp_path):
    r"""`[^/]*\.md$` must match only a top-level file ENDING in .md. Losing the `$` exempts
    `evil.mdx`, and the docs-only exemption posts a green status with nobody in the loop."""
    posted, _ = _run_classify(tmp_path, _emitting("evil.mdx"))
    assert posted is not None
    assert posted["state"] == "pending", "a .mdx file was accepted as docs-only"


def test_the_status_read_checks_EMPTINESS_in_SHELL_before_invoking_jq():
    """Structural, and deliberately so — no behavioural test can catch this on a dev machine.

    Deleting the `[ -z "${statusjson//[[:space:]]/}" ]` half of the guard leaves the empty case to
    `jq -e`'s exit status, which is 4 on jq >= 1.7 (guard still fires, and
    `test_a_transport_failure_on_the_STATUS_READ_posts_NOTHING` still passes on a Mac with jq 1.8)
    but **0 on jq 1.6, which is what the runner ships**. There the job sails past the guard and
    posts over a possibly-existing human verdict. That is literally ersatztv#647, and the reason
    `ci.jq-version-contract` exists.

    So the behavioural test catches this mutation only when the suite runs under 1.6 — in CI, not
    locally. This closes the local gap by pinning the construct.

    COMMENTS ARE STRIPPED FIRST. A raw substring search over the step text is satisfiable by moving
    the guard into a comment while deleting the real one — the same "a comment mentioning it is not
    a call" vacuity this file already fixed for the enumeration drift guard.
    """
    body = _classify_step()["run"]
    code = "\n".join(ln for ln in body.splitlines() if not ln.lstrip().startswith("#"))
    assert '[ -z "${statusjson//[[:space:]]/}" ]' in code, (
        "the status read must check emptiness in SHELL before jq; leaving it to `jq -e`'s exit "
        "status is fail-open on the runner's jq 1.6 (ersatztv#647)")