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:
    # A POST can be scripted to fail (`post_fail` holds a URL substring), so the two write paths
    # can be broken independently — the shape ersatztv#792 is about.
    fail_on = (state / "post_fail").read_text().strip() if (state / "post_fail").exists() else ""
    if fail_on and fail_on in url:
        sys.exit(22)
    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",
    }
    # A 2xx body that merely LOST the field, as distinct from an unreachable PR ('GONE' above).
    # This is the shape the `[ -n "$x" ] &&` conjunct used to wave through (ersatztv#778).
    if sha == "NOHEAD":
        body["head"] = {}
    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"  # noqa: S105 - deliberately fake; the real credential comes from the environment
    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 fail_posts_to(self, url_substring):
            """Make POSTs whose URL contains this substring fail, as curl -f does on a 4xx/5xx."""
            (state / "post_fail").write_text(url_substring)

        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"
    # And no comment either, since ersatztv#792. This assertion used to say the opposite — the
    # comment went first, so a refusal left `Review-verdict: MERGEABLE @ <sha>` on the PR with no
    # status behind it, which reads to an operator as consent that was never granted.
    assert gitea.comments() == [], "a refusal must leave no verdict comment standing in for a status"


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"
    )


# --- write-side polarity, the half the #774 rescue initially missed ------------------------------


# Kept deliberately in step with POSITIVE_WORDS/NEGATIVE_WORDS in test_check_review_verdict.py.
# The two lists are NOT compared to each other, and this file does not claim they agree — proving
# that needs one shared vocabulary both scripts read, which is ersatztv#788. What each side proves
# is its OWN polarity: that these established tokens still map the way reviewers rely on.
WRITE_POSITIVE = ["MERGEABLE", "APPROVED", "LGTM"]
WRITE_NEGATIVE = ["BLOCKED", "NOT-MERGEABLE"]


@pytest.mark.parametrize(
    ("word", "expected"),
    [(w, "success") for w in WRITE_POSITIVE] + [(w, "failure") for w in WRITE_NEGATIVE],
)
def test_each_verdict_word_posts_its_established_polarity(word, expected, gitea):
    """The WRITE side of the polarity pair, added after cold review found only the read side.

    ersatztv#774 withdrew a test that compared the two verdict vocabularies by parsing shell, and
    rescued its disjointness half into test_check_review_verdict.py. Review of that rescue found it
    covered only `check-review-verdict.sh`: an in-memory mutation adding `BLOCKED` to the SUCCESS
    arm here produced a write-side overlap the deleted test caught and the rescue did not, because
    the rescue never executes this script. That was a real, undisclosed loss and this closes it.

    `case` takes the FIRST matching arm, so a token listed in both arms is not ambiguous — it
    resolves to whichever comes first, exactly as `is_pos` wins on the read side. The consequence is
    the same and it is the one that matters: a word a reviewer means as BLOCKED silently posting
    `success` writes a green `review-verdict/h10`, which is the required context branch protection
    honours. That flips this assertion.

    What this does NOT prove, stated because the parent test was withdrawn for overclaiming: it is
    a polarity regression over five established tokens, not a universal disjointness property and
    not parity with the read side. A token added to only one script is untested here, not caught.
    """
    result = gitea.run("42", word)
    assert result.returncode == 0, result.stderr
    payload = gitea.statuses()[0]["payload"]
    assert payload["state"] == expected, (
        f"'{word}' posted state {payload['state']!r}, expected {expected!r}. `case` takes the first "
        "matching arm, so a token that has appeared in the other arm resolves there silently — and "
        "a BLOCKED verdict posting `success` writes a green required context."
    )


@pytest.mark.parametrize(
    ("head_seq", "base_seq", "field"),
    [
        (("NOHEAD",), ("main", "main"), "head sha"),
        ((SHA_A,), ("main", "MISSING"), "base ref"),
    ],
)
def test_a_reread_that_LOSES_a_field_refuses_instead_of_posting(head_seq, base_seq, field, gitea):
    """The fail-OPEN one level below the TOCTOU guard, found by cold review (ersatztv#778).

    Both re-read checks were written as `[ -n "$x" ] && [ "$x" != "$want" ]`. That conjunct makes an
    EMPTY value a no-op: a well-formed 2xx response that merely omits `.head.sha` or `.base.ref`
    yields an empty variable, neither comparison runs, and the status is posted having confirmed
    NOTHING about the head or the base — while the script's whole purpose at that point is to refuse
    unless it can confirm. The transport failure one line above was already fatal, which is exactly
    what made this shape easy to miss: the loud case was handled and the quiet one was not.

    Asserted on the OBSERVABLE outcome — no status written — rather than on message text, so it
    still holds if the wording changes.
    """
    gitea.set_head_sequence(SHA_A, *head_seq)
    gitea.set_base_sequence(*base_seq)
    result = gitea.run("42", "MERGEABLE")
    assert result.returncode != 0, (
        f"a re-read missing its {field} was accepted; the script posted a verdict having confirmed nothing about it"
    )
    assert not gitea.statuses(), (
        f"a status was written despite the re-read carrying no {field} — this is the fail-open the -n conjunct created"
    )


# --- ersatztv#792: no path may write no status and report success ------------------------------
#
# The issue was filed on an observed "printed the refusal AND exited 0". Re-measured on the tree
# that fixed the re-read fence: every refusal already exits NON-ZERO, and the exit-0 came from the
# caller's pipeline, not from the script. That is worth an executed contract rather than a second
# reading of the source — `die` is one line away from being edited into a `return`, and this file is
# where that would be caught. The parametrisation covers each refusal REASON, not one
# representative, because those paths were added at four different times and only the shared helper
# makes them agree today.
#
# SCOPE, since "every path" would overclaim: these are the eight refusal MODES reachable through the
# real entry point. The source also exits non-zero for a usage error (2), a failing `jq` (its own
# status, under `set -e` + `pipefail`), and a signal (128+n); none of those is a refusal DECISION,
# and only the non-zero-ness is common to all of them.


def _drive(gitea, mode):
    if mode == "head-moved":
        gitea.set_head_sequence(SHA_A, SHA_B)
    elif mode == "reread-failed":
        gitea.set_head_sequence(SHA_A, "GONE")
    elif mode == "reread-lost-head":
        gitea.set_head_sequence(SHA_A, "NOHEAD")
    elif mode == "base-retargeted":
        gitea.set_base_sequence("main", "some-feature-branch")
    elif mode == "reread-lost-base":
        gitea.set_base_sequence("main", "MISSING")
    elif mode == "first-read-failed":
        gitea.set_head_sequence("GONE")
    elif mode == "pr-closed":
        gitea.set_pr_state("closed")
    elif mode == "status-post-failed":
        gitea.fail_posts_to("/statuses/")
    else:  # pragma: no cover - a typo in the parametrisation must not pass silently
        raise AssertionError(f"unknown mode {mode}")
    return gitea.run("42", "MERGEABLE")


@pytest.mark.parametrize(
    "mode",
    [
        "head-moved",
        "reread-failed",
        "reread-lost-head",
        "base-retargeted",
        "reread-lost-base",
        "first-read-failed",
        "pr-closed",
        "status-post-failed",
    ],
)
def test_every_REFUSAL_MODE_that_writes_no_status_exits_non_zero(gitea, mode):
    result = _drive(gitea, mode)
    assert gitea.statuses() == [], f"{mode} wrote a status it had no business writing"
    assert result.returncode != 0, (
        f"{mode} wrote no status and reported SUCCESS — anything checking $? concludes the verdict "
        f"posted. stdout={result.stdout!r} stderr={result.stderr!r}"
    )


@pytest.mark.parametrize(
    "mode",
    [
        "head-moved",
        "reread-failed",
        "reread-lost-head",
        "base-retargeted",
        "reread-lost-base",
        "status-post-failed",
    ],
)
def test_no_refusal_leaves_a_VERDICT_COMMENT_standing_in_for_the_status(gitea, mode):
    """The half-state, which is the part of #792 that was really broken.

    The comment is not the gate — `review-verdict/h10` is — but `Review-verdict: MERGEABLE @ <head>`
    sitting on a PR reads exactly like consent. Every mode here is one where the status is refused
    after the head has been resolved, i.e. every mode that could once have left that comment behind.
    """
    _drive(gitea, mode)
    assert gitea.comments() == [], f"{mode} left an orphaned verdict comment: {gitea.comments()}"


def test_the_status_is_written_BEFORE_the_comment(gitea):
    """Ordering is the mechanism, so it is asserted rather than described.

    Status-then-comment makes the only reachable half-state the safe one: a status with no comment
    leaves the merge hook's condition (c) with nothing to classify, which is an `ask`. The reverse
    order manufactures the appearance of a granted verdict.
    """
    assert gitea.run("42", "MERGEABLE").returncode == 0
    urls = [p["url"] for p in gitea.posts()]
    assert len(urls) == 2, urls
    assert "/statuses/" in urls[0], f"the status must be written first, got {urls}"
    assert "/comments" in urls[1], f"the comment must be written second, got {urls}"


def test_a_failed_COMMENT_after_a_written_status_is_still_an_error(gitea):
    """The surviving half-state is safe, not silent: the operator is told to re-run."""
    gitea.fail_posts_to("/comments")
    result = gitea.run("42", "MERGEABLE")
    assert result.returncode != 0
    assert len(gitea.statuses()) == 1, "the status was already written and must not be rolled back"
    assert gitea.comments() == []
    assert "COMMENT could not be posted" in result.stderr
    assert "ask" in result.stderr.lower(), "it must say what the gate will do, not just that a call failed"