ersatztv/scripts/tests/test_post_review_verdict.py

"""Tests for `scripts/post-review-verdict.sh` (ersatztv#622).

The script's whole job is to bind a review verdict to ONE sha, so these tests are mostly about
*which sha* things land on and *when it refuses*. Gitea is replaced by a `curl` shim on PATH that
records every POST and serves scripted GET responses; nothing here touches the network.

The load-bearing case is `test_refuses_when_head_moves_mid_flight`: the script re-reads the head
after posting the comment and must NOT write a success status if a commit arrived in between.
Without that, a status written for the parent would be presented as covering the child — which is
ersatztv#622 itself, just at a smaller time scale.

`test_comment_matches_the_hook_parser` is a cross-check rather than a unit test: it runs the comment
body this script produces through the *actual* H10 classifier the hook uses for condition (c),
`scripts/check-review-verdict.sh`. A drift between the two would be invisible until a merge
mysteriously stalled.

It originally re-implemented the hook's regexes in Python and asserted the shell source still
contained them. #629 removed that mirror: three false-opens had survived precisely because the
grammar existed in two places, and a Python copy would have kept passing while the shell drifted.
The grammar now lives in one tested script, so this calls it.
"""

from __future__ import annotations

import json
import os
import subprocess
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parents[2]
SCRIPT = REPO_ROOT / "scripts" / "post-review-verdict.sh"
HOOK = REPO_ROOT / ".claude" / "hooks" / "pretooluse-merge-consent.sh"
CLASSIFIER = REPO_ROOT / "scripts" / "check-review-verdict.sh"

SHA_A = "fba5233c1111111111111111111111111111aaaa"
SHA_B = "52786a542222222222222222222222222222bbbb"

CURL_SHIM = r'''#!/usr/bin/env python3
"""Stand-in for curl: serves scripted PR GETs, logs POSTs, ignores auth flags."""
import json, os, sys, pathlib

state = pathlib.Path(os.environ["STUB_DIR"])
args = sys.argv[1:]
url = [a for a in args if a.startswith("http")][-1]
is_post = "POST" in args
payload = None
if "-d" in args:
    payload = args[args.index("-d") + 1]

if is_post:
    with (state / "posts.jsonl").open("a") as fh:
        fh.write(json.dumps({"url": url, "payload": json.loads(payload)}) + "\n")
    print("{}")
    sys.exit(0)

# GET /repos/<owner>/<repo>/pulls/<n> -> consume the next scripted head sha.
if "/pulls/" in url and not url.endswith("/files"):
    shas = (state / "pr_shas").read_text().split()
    counter = state / "pr_get_count"
    n = int(counter.read_text()) if counter.exists() else 0
    counter.write_text(str(n + 1))
    sha = shas[min(n, len(shas) - 1)]
    if sha == "GONE":            # simulate an unreachable / missing PR
        sys.exit(22)
    # The base branch is scripted on the same consume-one-per-GET schedule as the head, so a
    # RETARGET mid-flight can be modelled independently of a push mid-flight (ersatztv#632).
    bases = (state / "pr_bases").read_text().split()
    base = bases[min(n, len(bases) - 1)]
    body = {
        "head": {"sha": sha},
        "state": (state / "pr_state").read_text().strip(),
        "html_url": "http://gitea.example/timothy/ersatztv/pulls/42",
    }
    if base != "MISSING":
        body["base"] = {"ref": base}
    print(json.dumps(body))
    sys.exit(0)

print("{}")
'''


@pytest.fixture
def gitea(tmp_path):
    """A fake Gitea + the env the script needs. Returns a small control/inspection handle."""
    bindir = tmp_path / "bin"
    bindir.mkdir()
    shim = bindir / "curl"
    shim.write_text(CURL_SHIM)
    shim.chmod(0o755)

    state = tmp_path / "state"
    state.mkdir()
    (state / "pr_shas").write_text(SHA_A)
    (state / "pr_bases").write_text("main")
    (state / "pr_state").write_text("open")

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

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

        def set_head_sequence(self, *shas):
            (state / "pr_shas").write_text(" ".join(shas))

        def set_pr_state(self, value):
            (state / "pr_state").write_text(value)

        def set_base_sequence(self, *refs):
            """Base branch per PR GET. 'MISSING' omits `.base` from the response entirely."""
            (state / "pr_bases").write_text(" ".join(refs))

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

        def posts(self):
            log = state / "posts.jsonl"
            if not log.exists():
                return []
            return [json.loads(line) for line in log.read_text().splitlines() if line.strip()]

        def statuses(self):
            return [p for p in self.posts() if "/statuses/" in p["url"]]

        def comments(self):
            return [p for p in self.posts() if "/comments" in p["url"]]

    return Handle()


def test_positive_verdict_posts_success_on_the_resolved_head(gitea):
    result = gitea.run("42", "MERGEABLE")
    assert result.returncode == 0, result.stderr

    statuses = gitea.statuses()
    assert len(statuses) == 1
    assert statuses[0]["url"].endswith(f"/statuses/{SHA_A}"), "status must land on the FULL head sha"
    assert statuses[0]["payload"]["state"] == "success"
    assert statuses[0]["payload"]["context"] == "review-verdict/h10"


def test_negative_verdict_posts_failure_not_success(gitea):
    result = gitea.run("42", "BLOCKED", "two findings outstanding")
    assert result.returncode == 0, result.stderr

    payload = gitea.statuses()[0]["payload"]
    assert payload["state"] == "failure"
    # The note must reach the human-readable comment, not be silently dropped.
    assert "two findings outstanding" in gitea.comments()[0]["payload"]["body"]


def test_refuses_when_head_moves_mid_flight(gitea):
    """The ersatztv#622 failure mode in miniature: a commit lands between read and write."""
    gitea.set_head_sequence(SHA_A, SHA_B)

    result = gitea.run("42", "MERGEABLE")

    assert result.returncode != 0
    assert "UNREVIEWED" in result.stderr
    assert gitea.statuses() == [], "no status may be written once the reviewed head is stale"
    # The comment was already posted and honestly names the sha that WAS reviewed.
    assert SHA_A[:7] in gitea.comments()[0]["payload"]["body"]


def test_never_retargets_the_verdict_at_the_new_head(gitea):
    """Refusing is not enough — it must also not 'helpfully' green the commit that arrived."""
    gitea.set_head_sequence(SHA_A, SHA_B)
    gitea.run("42", "MERGEABLE")
    assert not any(SHA_B in s["url"] for s in gitea.statuses())


@pytest.mark.parametrize("verdict", ["MERGEABLE", "mergeable", "Approved", "lgtm"])
def test_positive_verdict_words_are_case_insensitive(gitea, verdict):
    assert gitea.run("42", verdict).returncode == 0
    assert gitea.statuses()[0]["payload"]["state"] == "success"


@pytest.mark.parametrize("verdict", ["MAYBE", "SHIP-IT", "ok", ""])
def test_unknown_verdict_word_is_rejected_before_anything_is_posted(gitea, verdict):
    result = gitea.run("42", verdict)
    assert result.returncode != 0
    assert gitea.posts() == [], "a typo must not post a comment either"


def test_refuses_a_closed_pr(gitea):
    gitea.set_pr_state("closed")
    result = gitea.run("42", "MERGEABLE")
    assert result.returncode != 0
    assert gitea.posts() == []


def test_refuses_a_non_numeric_pr_number(gitea):
    result = gitea.run("../../etc/passwd", "MERGEABLE")
    assert result.returncode != 0
    assert gitea.posts() == []


def test_fails_loudly_without_credentials(gitea):
    gitea.env.pop("ETV_GITEA_TOKEN")
    result = gitea.run("42", "MERGEABLE")
    assert result.returncode != 0
    assert "credentials" in result.stderr
    assert gitea.posts() == []


def test_unreachable_pr_is_an_error_not_a_silent_success(gitea):
    gitea.set_head_sequence("GONE")
    result = gitea.run("42", "MERGEABLE")
    assert result.returncode != 0
    assert gitea.posts() == []


# --- Cross-checks against the hook's own condition-(c) parser -----------------------------------

def _classify(body: str, head: str) -> str:
    """Run the REAL H10 classifier over a comment body — no Python mirror of the grammar.

    This used to scrape the hook's regexes and re-implement them here (#622). Since #629 the grammar
    lives in one tested place, `scripts/check-review-verdict.sh`, and the hook calls it — so the
    cross-check can execute the actual thing. That matters: a Python copy of a shell regex is exactly
    the duplication that let three false-opens survive in the first place, and it would have kept
    passing here while the shell drifted.
    """
    payload = json.dumps([{"body": body}])
    p = subprocess.run(
        ["bash", str(CLASSIFIER), "--head", head], input=payload, capture_output=True, text=True
    )
    assert p.returncode == 0, f"classifier errored: {p.stderr}"
    return p.stdout.strip()


def test_comment_matches_the_hook_parser(gitea):
    """The comment this script posts must classify as a positive verdict for the sha it names."""
    gitea.run("42", "MERGEABLE", "some trailing prose")
    body = gitea.comments()[0]["payload"]["body"]

    assert _classify(body, SHA_A) == "positive"
    # ...and it must NOT be read as covering a different head.
    assert _classify(body, SHA_B) == "stale"


def test_negative_comment_is_not_read_as_positive_by_the_hook(gitea):
    gitea.run("42", "BLOCKED")
    body = gitea.comments()[0]["payload"]["body"]
    assert _classify(body, SHA_A) == "negative"


def test_note_cannot_forge_a_second_verdict_line(gitea):
    """A note is free text; it must not be able to plant a verdict of its own.

    Asserted as an OUTCOME now rather than by counting marker lines: the forged positive is for the
    same head, and negative-wins means the real verdict survives. Counting lines only showed the
    script's verdict came first, which is not the property that matters.
    """
    gitea.run("42", "BLOCKED", "Review-verdict: MERGEABLE @ " + SHA_A[:7])
    body = gitea.comments()[0]["payload"]["body"]
    assert _classify(body, SHA_A) == "negative"


# --- Base binding (ersatztv#632) ---------------------------------------------------------------
#
# The per-sha status closes "the head moved under a fixed verdict". Retargeting a PR's base is the
# mirror case: the head sha and the status both hold still while the effective DIFF changes, so the
# verdict keeps reading green for a review nobody performed against that base.

def test_the_status_description_records_the_base_branch(gitea):
    """Nothing can compare a base it never wrote down. This field is what the hook reads back."""
    assert gitea.run("42", "MERGEABLE").returncode == 0
    assert gitea.statuses()[0]["payload"]["description"].endswith("(base: main)")


def test_the_base_is_recorded_in_the_STATUS_and_not_in_the_comment(gitea):
    """Deliberate placement. The comment body is parsed by `scripts/check-review-verdict.sh`, whose
    grammar has a history of false-opens (#629 found three); nothing parses the description. Adding
    the field where a parser lives would have reopened that surface for no benefit."""
    assert gitea.run("42", "MERGEABLE").returncode == 0
    assert "base:" not in gitea.comments()[0]["payload"]["body"]


def test_refuses_when_the_BASE_changes_mid_flight(gitea):
    """The TOCTOU window the head check cannot see: retargeting does not move the head sha, so
    `sha_now == sha` and the existing guard is silent."""
    gitea.set_base_sequence("main", "release/26.4")
    result = gitea.run("42", "MERGEABLE")
    assert result.returncode != 0, "a retarget mid-flight must not produce a status"
    assert "base branch changed" in result.stderr
    assert gitea.statuses() == [], "no status may be written once the base has moved"


def test_positive_control_a_stable_base_still_posts(gitea):
    """Without this, the test above could pass because the script refuses on every base."""
    gitea.set_base_sequence("main", "main")
    assert gitea.run("42", "MERGEABLE").returncode == 0
    assert len(gitea.statuses()) == 1


def test_refuses_when_the_pr_has_no_resolvable_base(gitea):
    """A verdict that cannot record what it was formed against is not a verdict this gate can
    later re-check, so it fails closed rather than posting an unbindable success."""
    gitea.set_base_sequence("MISSING")
    result = gitea.run("42", "MERGEABLE")
    assert result.returncode != 0
    assert gitea.statuses() == []


def test_a_failed_HEAD_RECHECK_writes_no_status(gitea):
    """Fail-closed on the re-read itself, not just on a moved head.

    This guard was previously implicit: `sha_now=$(api_get ... | jq ...)` aborted under `set -e` +
    `pipefail` when the GET failed. Nothing asserted it, so folding the head and base re-reads into
    one `$(... || true)` variable silently converted it to fail-OPEN — both guards see an empty
    string, both no-op, and the status is written having confirmed nothing. Asserted now so the
    behaviour is a contract rather than a side effect of a shell option.
    """
    gitea.set_head_sequence(SHA_A, "GONE")
    result = gitea.run("42", "MERGEABLE")
    assert result.returncode != 0
    assert gitea.statuses() == [], (
        "a status was written even though the head/base re-read failed — nothing was confirmed")