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"
# (the enforced caller, .gitea/workflows/review-verdict.yml, is wired in the follow-up PR)

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_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_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_moving_during_enumeration_fails_closed(enumerate_files):
    """A force-push between round-trips means page 1 came from head A and page 2 from head B, so
    the assembled list belongs to no single commit."""
    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 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 ----------------------------------

def test_the_hook_uses_the_shared_script_and_does_not_reimplement_it():
    """ersatztv#649's third Done-when box: a test that fails if a copy is re-inlined.

    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.

    SCOPE NOTE — this asserts the HOOK only, deliberately. `.gitea/workflows/review-verdict.yml` is
    the other caller, but its rewiring cannot land in this PR: that workflow will check out the BASE
    ref (never the PR head, so a PR cannot rewrite the gate judging it), and the base is `main`,
    which does not yet contain the scripts this PR ADDS. Wiring it here would make the job exit 127
    on its own PR and block the merge gate via the combined status. The workflow half therefore
    lands in the follow-up PR, once these scripts are on `main`, and that PR extends this test to
    cover both callers.
    """
    text = HOOK.read_text()
    assert "scripts/pr-changed-files.sh" in text, (
        f"{HOOK.relative_to(REPO_ROOT)} no longer calls the shared enumeration")
    # Search CODE only. The hook's comments legitimately discuss this endpoint, and a future comment
    # writing `pulls/$pr/files?limit=100` as an example of what NOT to do would redden this test —
    # which, since `script-tests` red blocks merges via the combined status, would block the repo over
    # a piece of prose.
    code = "\n".join(ln for ln in text.splitlines() if not ln.lstrip().startswith("#"))
    # 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), (
        f"{HOOK.relative_to(REPO_ROOT)} appears to enumerate PR files inline again — "
        "that is the duplication ersatztv#649 removed")