ersatztv/scripts/tests/test_merge_consent_head_change.py

"""Tests for the head re-read in `.claude/hooks/pretooluse-merge-consent.sh` (ersatztv#803).

`$sha` is captured once from the PR snapshot at the top of the hook, and every later check — the CI
combined status, the `review-verdict/h10` status, and the verdict-comment classification — is
addressed by it. Between the capture and those checks sits the docs-only enumeration, up to forty
round trips. A push landing in that gap was therefore checked against the commit it had just
replaced, and the hook would report "a positive Review-verdict references the current head" about a
head that was no longer current.

The fix is deliberately the SAME one the base got in #778, at the SAME hoist and off the SAME
response, so the two axes cannot describe two different instants. These tests pin that it fires, that
it names both shas, that it does not fire on a quiet run, and that an unreadable head asks rather than
denying.

WHAT IS NOT CLAIMED HERE. The hook cannot detect an ABA — a push away and back leaves `.head.sha`
equal at both of its reads, exactly as it does inside `scripts/pr-changed-files.sh`. That case belongs
to the monotonic `pull_push` count in `.gitea/workflows/review-verdict.yml`
(`ci.verdict-write-retarget-fence`), and is covered in `test_pr_changed_files.py`. No test here
implies the hook closes it.

Observable contract: the hook exits 0 with EMPTY stdout when it has no opinion (passthrough to normal
permissioning), and emits a JSON `permissionDecision` otherwise.
"""

from __future__ import annotations

import json
import os
import subprocess
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parents[2]
HOOK = REPO_ROOT / ".claude" / "hooks" / "pretooluse-merge-consent.sh"

SHA = "a9e3e23abf337980ca4c05854f5b1e210099d08b"
MOVED = "b7c4d19ffe2210873bb9a0417c6e3f2280551ac4"

# NOT docs-only, on purpose: the docs-only exemption short-circuits the gate long before the hoist,
# so a docs PR would never reach the head check and every test here would pass without exercising it.
#
# THE MOVE IS SERVED FROM THE SECOND `/pulls/{n}` READ ONWARD, not at a hardcoded read index. The
# hook's own two reads are not adjacent — `scripts/pr-changed-files.sh` makes its own in between, in a
# separate process sharing this shim — so counting to a specific read would pin an implementation
# detail that any refactor of the enumerator would silently break. "Moved once, early, and stayed
# moved" is both the realistic shape and the one that needs no such knowledge.
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])
    if page == 1:
        print(json.dumps([{"filename": "ErsatzTV/Program.cs", "status": "modified"}]))
    else:
        print("[]")
    sys.exit(0)

if "/status" in url:
    print(json.dumps({"state": "success", "statuses": [
        {"context": "review-verdict/h10", "status": "success",
         "description": "Review-verdict: MERGEABLE @ a9e3e23 (base: main)"}]}))
    sys.exit(0)

if "/pulls/" in url:
    ctr = state / "pr_reads"
    seen = int(ctr.read_text()) if ctr.exists() else 0
    ctr.write_text(str(seen + 1))
    mode = (state / "head_mode").read_text().strip()
    sha = os.environ["STUB_SHA"]
    if seen >= 1:
        if mode == "moved":
            sha = os.environ["STUB_MOVED"]
        elif mode == "unreadable":
            sha = ""
    body = {"base": {"ref": "main"}, "body": "fixes #1"}
    if sha:
        body["head"] = {"sha": sha}
    print(json.dumps(body))
    sys.exit(0)

print("{}")
"""


@pytest.fixture
def hook(tmp_path):
    bindir = tmp_path / "bin"
    bindir.mkdir()
    curl = bindir / "curl"
    curl.write_text(CURL_SHIM)
    curl.chmod(0o755)
    state = tmp_path / "state"
    state.mkdir()
    (state / "head_mode").write_text("stable")

    env = dict(os.environ)
    env["PATH"] = f"{bindir}{os.pathsep}{env['PATH']}"
    env["STUB_DIR"] = str(state)
    env["STUB_SHA"] = SHA
    env["STUB_MOVED"] = MOVED
    env["ETV_GITEA_TOKEN"] = "stub"  # noqa: S105 - deliberately fake; the real credential comes from the environment
    env["ETV_GITEA_URL"] = "http://gitea.example"
    env.pop("ETV_GITEA_BASICAUTH", None)

    class Handle:
        def set_head_mode(self, mode):
            """'stable' | 'moved' | 'unreadable' — applied from the SECOND /pulls read onward."""
            (state / "head_mode").write_text(mode)

        def decision(self, mwcs=False):
            payload = {"tool_input": {"method": "merge", "owner": "timothy", "repo": "ersatztv", "pull_number": 42}}
            if mwcs:
                payload["tool_input"]["merge_when_checks_succeed"] = True
            r = subprocess.run(["bash", str(HOOK)], input=json.dumps(payload), env=env, capture_output=True, text=True)
            assert r.returncode == 0, r.stderr
            return json.loads(r.stdout) if r.stdout.strip() else None

        def reason(self, mwcs=False):
            d = self.decision(mwcs=mwcs)
            return "" if d is None else json.dumps(d)

    return Handle()


def test_a_head_that_MOVED_mid_run_denies(hook):
    hook.set_head_mode("moved")
    reason = hook.reason()
    assert "deny" in reason, "the hook kept its opinion about a head that was replaced while it was evaluating"
    assert SHA[:7] in reason and MOVED[:7] in reason, (
        "the deny must name BOTH shas; 'the head moved' is not something a reader can act on"
    )
    assert "803" in reason, "the deny should cite the issue that explains the window"


def test_positive_control_a_QUIET_head_does_not_trigger_the_head_deny(hook):
    """Without this, the test above passes against a hook that denies on every path.

    It very nearly would: this PR is deliberately non-docs, so the gate runs to completion, and the
    assertion is not "no decision" but "not a decision caused by the HEAD arm".
    """
    hook.set_head_mode("stable")
    reason = hook.reason()
    assert "head moved from" not in reason, f"the head arm fired on a run whose head never moved: {reason}"
    # Positively witness that the run got PAST the head arm rather than never reaching it — an
    # absence assertion alone is satisfied by a hook that exited before the check ever ran. This
    # fixture stops at the issue-body read, which is downstream of the hoist.
    assert "could not fetch linked issue" in reason, (
        "the run did not reach the checks downstream of the head re-read, so this control does not "
        f"show the head arm was exercised and stayed silent: {reason}"
    )


def test_an_UNREADABLE_head_on_re_read_ASKS_rather_than_denying(hook):
    """The two cases are not the same and must not collapse into one decision.

    A head that MOVED is a state we positively established — the verdict covers an older commit,
    which is what the `stale` verdict class denies for. A head we could not READ is the absence of
    evidence, and this gate's standing rule is that uncertainty asks. Collapsing them would block
    merges on a transport blip; collapsing them the other way would grant on one.
    """
    hook.set_head_mode("unreadable")
    reason = hook.reason()
    assert "deny" not in reason, f"an unreadable head was treated as a moved one: {reason}"
    # ASSERT WHICH ask, not merely that SOME ask happened. This fixture's issue endpoint returns
    # `{}`, so the hook asks later anyway for a completely unrelated reason — meaning a mutation
    # deleting this arm outright (empty `sha_now` -> fall through and keep using `$sha`) left the
    # test green. Naming the arm's own message is what makes the test measure the arm.
    assert "reports no head commit (.head.sha) on re-read" in reason, (
        f"the hook asked, but NOT via the unreadable-head arm: {reason}"
    )


@pytest.mark.parametrize("mwcs", [False, True], ids=["immediate", "scheduled"])
def test_the_head_deny_covers_BOTH_merge_paths(hook, mwcs):
    """The twin-miss shape, tested because this exact hook has already been bitten by it.

    The base re-read first landed INSIDE the scheduled-auto-merge branch only, and the
    consequence showed up on this repo's own fixture: scheduled+retarget denied while
    immediate+retarget auto-GRANTED. The head check is placed at the same hoist precisely so it
    sits ABOVE the point where the two paths diverge — but "it is above the split" is a claim about
    the source, and the thing worth pinning is the OUTCOME on both paths.
    """
    hook.set_head_mode("moved")
    reason = hook.reason(mwcs=mwcs)
    assert "deny" in reason, (
        f"a head that moved mid-run was not denied on the {'scheduled' if mwcs else 'immediate'} merge path: {reason}"
    )
    assert MOVED[:7] in reason, f"the deny does not name the new head: {reason}"