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_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()


# --- Drift guard: the reason this file is worth having at all ----------------------------------

def test_both_callers_use_the_shared_script_and_neither_reimplements_it():
    """ersatztv#649's third Done-when box: a test that fails if the two copies diverge again.

    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.
    """
    for caller in (HOOK, WORKFLOW):
        text = caller.read_text()
        assert "scripts/pr-changed-files.sh" in text, (
            f"{caller.relative_to(REPO_ROOT)} no longer calls the shared enumeration")
        # An inline `pulls/<n>/files?limit=` fetch is the signature of a re-inlined copy.
        assert not re.search(r"pulls/\$?\{?\w+\}?/files\?limit=", text), (
            f"{caller.relative_to(REPO_ROOT)} appears to enumerate PR files inline again — "
            "that is the duplication ersatztv#649 removed")