Review gate: automated PR reviewer (acceptance parse, tests, verdicts) (#5)
This commit is contained in:
commit
bcf03bd9cc
376
gitea_saas/reviewer.py
Normal file
376
gitea_saas/reviewer.py
Normal file
@ -0,0 +1,376 @@
|
||||
"""Review gate: automated PR reviewer (issue #5).
|
||||
|
||||
Turns "lanes produce PRs" into "clients get verified code": parses a PR's
|
||||
linked issue acceptance criteria, runs real checks (tests, build, diff
|
||||
hygiene, secret scan), and produces an evidence-based verdict.
|
||||
|
||||
Design
|
||||
------
|
||||
* ``Reviewer(repo_client, runner)`` — ``repo_client`` is a git/Gitea seam
|
||||
(see ``gitea_client.GiteaClient`` for the real one), ``runner`` is a
|
||||
subprocess seam (``Runner`` here; a scripted fake in tests).
|
||||
* ``review_pr(repo, pr_number, linked_issue, repo_dir)`` returns a
|
||||
``Verdict``; the caller decides what to do. ``post_verdict`` posts the
|
||||
formatted ``[review-gate]`` comment.
|
||||
* FAIL-CLOSED: the review gate NEVER merges — there is no merge call in
|
||||
this module. Policy and merge are issue #7.
|
||||
* Every finding MUST quote the diff. The default (no LLM configured) pass
|
||||
is deterministic: secret-pattern and TODO/FIXME scans over added lines,
|
||||
each with the exact added-line text as ``evidence_quote``. An LLM pass
|
||||
can be supplied via config ``llm``; its output is validated and any
|
||||
finding without ``evidence_quote`` is dropped (never assert vaguely).
|
||||
|
||||
Verdict rules
|
||||
-------------
|
||||
1. tests non-zero exit -> BOUNCE (unless the issue body contains
|
||||
``allow_test_failure``); timeout -> BOUNCE always
|
||||
2. build non-zero exit (if a build command is configured) -> BOUNCE
|
||||
3. scope violation (changed file outside the issue's declared allowed set)
|
||||
-> BOUNCE
|
||||
4. any blocking finding -> BOUNCE
|
||||
5. acceptance list empty -> NEEDS_GUIDANCE (never silent-approve)
|
||||
6. otherwise -> APPROVE
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Callable, Optional
|
||||
|
||||
APPROVE = "APPROVE"
|
||||
BOUNCE = "BOUNCE"
|
||||
NEEDS_GUIDANCE = "NEEDS_GUIDANCE"
|
||||
|
||||
_DEFAULT_SECRET_PATTERNS = [
|
||||
("stripe live key", re.compile(r"sk_live_[A-Za-z0-9]{8,}")),
|
||||
("aws access key", re.compile(r"\bAKIA[0-9A-Z]{16}\b")),
|
||||
("github token", re.compile(r"\bgh[pousr]_[A-Za-z0-9]{20,}\b")),
|
||||
("private key block", re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----")),
|
||||
("assigned credential", re.compile(
|
||||
r"""(?i)(?:api[_-]?key|secret|token|password)\s*[:=]\s*["'][^"']{12,}["']""")),
|
||||
]
|
||||
|
||||
|
||||
def _tail(text: str, limit: int = 800) -> str:
|
||||
text = (text or "").strip()
|
||||
if len(text) <= limit:
|
||||
return text
|
||||
return "…" + text[-limit:]
|
||||
|
||||
|
||||
def parse_acceptance(issue_body: str) -> list[str]:
|
||||
"""Extract the acceptance checklist from an issue body.
|
||||
|
||||
Matches ``- [ ] ...`` / ``- [x] ...`` checklist lines anywhere, falling
|
||||
back to bullet lines under a ``## Acceptance`` section. Empty result
|
||||
means the issue declares no verifiable acceptance criteria.
|
||||
"""
|
||||
body = issue_body or ""
|
||||
items = re.findall(r"^\s*-\s*\[[ xX]\]\s+(.+?)\s*$", body, re.MULTILINE)
|
||||
if not items:
|
||||
m = re.search(r"(?im)^#{1,3}\s*Acceptance\s*$([\s\S]*?)(?=^#{1,3}\s|\Z)", body)
|
||||
section = m.group(1) if m else ""
|
||||
items = [line.strip().lstrip("-*").strip()
|
||||
for line in section.splitlines()
|
||||
if line.strip().startswith(("-", "*")) and line.strip() not in ("-", "*")]
|
||||
return [i.strip() for i in items if i.strip()]
|
||||
|
||||
|
||||
def parse_allowed_files(issue_body: str) -> list[str]:
|
||||
"""Extract the backticked "Files allowed" set from an issue body.
|
||||
|
||||
Returns [] when the issue declares no allowed-set constraint.
|
||||
"""
|
||||
body = issue_body or ""
|
||||
m = re.search(r"(?i)files?\s+allowed\s*[:：]([\s\S]*?)(?=do\s+not\s+modify|\Z)", body)
|
||||
if not m:
|
||||
return []
|
||||
return re.findall(r"`([^`]+)`", m.group(1))
|
||||
|
||||
|
||||
def _matches_allowed(filename: str, allowed: list[str]) -> bool:
|
||||
filename = filename.lstrip("./")
|
||||
for pattern in allowed:
|
||||
pattern = pattern.lstrip("./")
|
||||
if pattern.endswith("/"):
|
||||
if filename.startswith(pattern):
|
||||
return True
|
||||
elif filename == pattern:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _iter_added_lines(diff_text: str):
|
||||
"""Yield (filename, new_line, content) for every added ('+') diff line."""
|
||||
current_file: Optional[str] = None
|
||||
new_line: Optional[int] = None
|
||||
for raw in diff_text.splitlines():
|
||||
if raw.startswith("+++ "):
|
||||
current_file = raw[4:].split("\t")[0].lstrip("a/").lstrip("b/")
|
||||
new_line = None
|
||||
elif raw.startswith("@@ "):
|
||||
m = re.search(r"\+(\d+)(?:,\d+)? ", raw)
|
||||
new_line = int(m.group(1)) if m else None
|
||||
elif raw.startswith("+") and not raw.startswith("+++"):
|
||||
yield (current_file or "?", new_line, raw[1:])
|
||||
if new_line is not None:
|
||||
new_line += 1
|
||||
elif raw.startswith("-") or raw == r"\ No newline at end of file":
|
||||
pass # removed side: does not advance the new side
|
||||
else:
|
||||
if new_line is not None:
|
||||
new_line += 1
|
||||
|
||||
|
||||
def _redact_quote(text: str, match: "re.Match[str]") -> str:
|
||||
"""Quote the diff line but redact the matched secret value."""
|
||||
start, end = match.span()
|
||||
return text[:start] + "***REDACTED***" + text[end:]
|
||||
|
||||
|
||||
class Runner:
|
||||
"""Subprocess seam. ``command_runner`` is injectable for hermetic tests.
|
||||
|
||||
Returns ``{"exit": int | "timed_out", "summary": tail-of-output}``.
|
||||
"""
|
||||
|
||||
def __init__(self, timeout_s: int = 300,
|
||||
command_runner: Optional[Callable] = None):
|
||||
self.timeout_s = timeout_s
|
||||
self._command_runner = command_runner or (
|
||||
lambda command, **kwargs: subprocess.run(command, shell=True, **kwargs))
|
||||
|
||||
def run(self, repo_dir, command: str, timeout_s: Optional[int] = None) -> dict:
|
||||
timeout = timeout_s or self.timeout_s
|
||||
try:
|
||||
completed = self._command_runner(
|
||||
command, cwd=str(repo_dir), timeout=timeout,
|
||||
capture_output=True, text=True)
|
||||
output = ((completed.stdout or "") + "\n" + (completed.stderr or "")).strip()
|
||||
return {"exit": completed.returncode, "summary": _tail(output)}
|
||||
except subprocess.TimeoutExpired as exc:
|
||||
stdout = exc.stdout or b""
|
||||
stderr = exc.stderr or b""
|
||||
if isinstance(stdout, bytes):
|
||||
stdout = stdout.decode("utf-8", "replace")
|
||||
if isinstance(stderr, bytes):
|
||||
stderr = stderr.decode("utf-8", "replace")
|
||||
return {"exit": "timed_out",
|
||||
"summary": _tail(f"{stdout}\n{stderr}\n"
|
||||
f"[timed out after {timeout}s]")}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Verdict:
|
||||
"""Evidence-based outcome of a review gate run."""
|
||||
|
||||
repo: str
|
||||
pr: int
|
||||
head_sha: str
|
||||
acceptance: list[str] = field(default_factory=list)
|
||||
tests: dict = field(default_factory=dict)
|
||||
build: Optional[dict] = None
|
||||
scope_violations: list[str] = field(default_factory=list)
|
||||
findings: list[dict] = field(default_factory=list)
|
||||
verdict: str = NEEDS_GUIDANCE
|
||||
summary: str = ""
|
||||
|
||||
def to_comment(self) -> str:
|
||||
"""Format the [review-gate] comment (checklist + findings)."""
|
||||
lines = [f"[review-gate] VERDICT: **{self.verdict}**", "",
|
||||
f"Repo: `{self.repo}` · PR: #{self.pr} · head `{self.head_sha[:12]}`", ""]
|
||||
lines.append(f"Acceptance parsed ({len(self.acceptance)}):")
|
||||
if self.acceptance:
|
||||
lines += [f"- [ ] {a}" for a in self.acceptance]
|
||||
else:
|
||||
lines.append("- (none found — no verifiable acceptance criteria)")
|
||||
lines += ["",
|
||||
f"Tests: exit={self.tests.get('exit')}",
|
||||
f"```\n{self.tests.get('summary', '')[:400]}\n```"]
|
||||
if self.build is not None:
|
||||
lines += ["",
|
||||
f"Build ({self.build.get('command', '')}): exit={self.build.get('exit')}",
|
||||
f"```\n{self.build.get('summary', '')[:400]}\n```"]
|
||||
if self.scope_violations:
|
||||
lines += ["", "Scope violations:"]
|
||||
lines += [f"- `{f}`" for f in self.scope_violations]
|
||||
lines += ["", f"Findings ({len(self.findings)}):"]
|
||||
if self.findings:
|
||||
for f in self.findings:
|
||||
loc = f"{f.get('file')}" + (f":{f.get('line')}" if f.get("line") is not None else "")
|
||||
lines.append(f"- [{f.get('severity')}] {loc}: {f.get('note')}")
|
||||
lines.append(f" `{f.get('evidence_quote', '')[:200]}`")
|
||||
else:
|
||||
lines.append("- none")
|
||||
lines += ["",
|
||||
f"Summary: {self.summary}",
|
||||
"", "No merge performed by the review gate (policy/merge is issue #7)."]
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
class Reviewer:
|
||||
"""Automated PR reviewer (issue #5). Never merges anything."""
|
||||
|
||||
def __init__(self, repo_client, runner, config: Optional[dict] = None):
|
||||
"""``repo_client`` and ``runner`` are duck-typed seams (see module docstring)."""
|
||||
self.repo_client = repo_client
|
||||
self.runner = runner
|
||||
self.config = {
|
||||
"build_cmd": os.environ.get("REVIEW_BUILD_CMD"),
|
||||
"test_timeout_s": int(os.environ.get("REVIEW_TEST_TIMEOUT_S", "300")),
|
||||
"secret_scan": os.environ.get("REVIEW_SECRET_SCAN", "1") != "0",
|
||||
"llm": None, # callable(prompt) -> list[dict] | None, if configured
|
||||
}
|
||||
if config:
|
||||
self.config.update(config)
|
||||
|
||||
# -- main entry points -------------------------------------------------
|
||||
|
||||
def review_pr(self, repo: str, pr_number: int, linked_issue, repo_dir) -> Verdict:
|
||||
issue_body = linked_issue.get("body") if isinstance(linked_issue, dict) else str(linked_issue or "")
|
||||
|
||||
pr = self.repo_client.get_pr(repo, pr_number)
|
||||
head_sha = pr.get("head_sha") or "unknown"
|
||||
changed_files = pr.get("changed_files") or []
|
||||
diff_text = self.repo_client.get_pr_diff(repo, pr_number) or ""
|
||||
|
||||
acceptance = parse_acceptance(issue_body)
|
||||
allowed = parse_allowed_files(issue_body)
|
||||
scope_violations = [f for f in changed_files
|
||||
if allowed and not _matches_allowed(f, allowed)]
|
||||
|
||||
tests = self.runner.run(repo_dir, "python -m pytest -q",
|
||||
timeout_s=self.config["test_timeout_s"])
|
||||
build = None
|
||||
if self.config.get("build_cmd"):
|
||||
build = self.runner.run(repo_dir, self.config["build_cmd"],
|
||||
timeout_s=self.config["test_timeout_s"])
|
||||
build["command"] = self.config["build_cmd"]
|
||||
|
||||
hygiene = self.runner.run(repo_dir, "git diff --check",
|
||||
timeout_s=self.config["test_timeout_s"])
|
||||
|
||||
findings = self._collect_findings(diff_text, tests, hygiene)
|
||||
|
||||
verdict, summary = self._decide(
|
||||
issue_body, acceptance, tests, build, scope_violations, findings, hygiene)
|
||||
|
||||
return Verdict(
|
||||
repo=repo, pr=pr_number, head_sha=head_sha,
|
||||
acceptance=acceptance, tests=tests, build=build,
|
||||
scope_violations=scope_violations, findings=findings,
|
||||
verdict=verdict, summary=summary,
|
||||
)
|
||||
|
||||
def post_verdict(self, repo: str, pr_number: int, verdict: Verdict) -> str:
|
||||
"""Post the formatted [review-gate] comment. Returns the comment."""
|
||||
comment = verdict.to_comment()
|
||||
self.repo_client.post_comment(repo, pr_number, comment)
|
||||
return comment
|
||||
|
||||
# -- checks ------------------------------------------------------------
|
||||
|
||||
def _collect_findings(self, diff_text: str, tests: dict, hygiene: dict) -> list[dict]:
|
||||
findings: list[dict] = []
|
||||
|
||||
llm = self.config.get("llm")
|
||||
if llm is not None:
|
||||
findings += self._llm_pass(llm, diff_text, tests)
|
||||
else:
|
||||
findings += self._heuristic_findings(diff_text)
|
||||
|
||||
if hygiene.get("exit") not in (0, "not_run", None):
|
||||
findings.append({
|
||||
"file": "(repo)", "line": None, "severity": "blocking",
|
||||
"evidence_quote": _tail(hygiene.get("summary", ""), 200),
|
||||
"note": "git diff --check reported whitespace/conflict errors",
|
||||
})
|
||||
return findings
|
||||
|
||||
def _heuristic_findings(self, diff_text: str) -> list[dict]:
|
||||
findings: list[dict] = []
|
||||
patterns = _DEFAULT_SECRET_PATTERNS if self.config.get("secret_scan") else []
|
||||
for filename, lineno, content in _iter_added_lines(diff_text):
|
||||
if not content.strip():
|
||||
continue
|
||||
for name, pattern in patterns:
|
||||
m = pattern.search(content)
|
||||
if m:
|
||||
findings.append({
|
||||
"file": filename, "line": lineno, "severity": "blocking",
|
||||
"evidence_quote": _redact_quote(content, m),
|
||||
"note": f"possible {name} leaked in added line",
|
||||
})
|
||||
break # one finding per line is enough
|
||||
if re.search(r"\b(?:TODO|FIXME|HACK)\b", content):
|
||||
findings.append({
|
||||
"file": filename, "line": lineno, "severity": "warning",
|
||||
"evidence_quote": content,
|
||||
"note": "TODO/FIXME/HACK marker in added line",
|
||||
})
|
||||
return findings
|
||||
|
||||
def _llm_pass(self, llm: Callable, diff_text: str, tests: dict) -> list[dict]:
|
||||
prompt = (
|
||||
"You are the review gate for a code PR. Acceptance criteria and "
|
||||
"test results are below. Return ONLY a JSON list of findings, each "
|
||||
"{\"file\", \"line\"?, \"severity\": blocking|warning|info, "
|
||||
"\"evidence_quote\" (EXACT text quoted from the diff), \"note\"}. "
|
||||
"Never invent evidence that is not in the diff.\n\n"
|
||||
f"Test result: exit={tests.get('exit')}\n{diff_text[:4000]}"
|
||||
)
|
||||
try:
|
||||
raw = llm(prompt)
|
||||
except Exception:
|
||||
return []
|
||||
if not isinstance(raw, list):
|
||||
return []
|
||||
validated = []
|
||||
for f in raw:
|
||||
if not isinstance(f, dict):
|
||||
continue
|
||||
if "evidence_quote" not in f or "severity" not in f or not f["evidence_quote"]:
|
||||
continue # unquoted/vague finding: dropped, fail-closed
|
||||
severity = str(f.get("severity", "info")).lower()
|
||||
if severity not in ("blocking", "warning", "info"):
|
||||
severity = "info"
|
||||
validated.append({
|
||||
"file": str(f.get("file", "?")),
|
||||
"line": f.get("line"),
|
||||
"severity": severity,
|
||||
"evidence_quote": str(f["evidence_quote"])[:500],
|
||||
"note": str(f.get("note", "")),
|
||||
})
|
||||
return validated
|
||||
|
||||
# -- verdict -----------------------------------------------------------
|
||||
|
||||
def _decide(self, issue_body: str, acceptance: list[str], tests: dict,
|
||||
build: Optional[dict], scope_violations: list[str],
|
||||
findings: list[dict], hygiene: dict) -> tuple[str, str]:
|
||||
allow_test_failure = "allow_test_failure" in (issue_body or "").lower()
|
||||
blocking = [f for f in findings if f.get("severity") in ("blocking", "critical")]
|
||||
|
||||
if tests.get("exit") == "timed_out":
|
||||
return BOUNCE, "tests timed out — rerun or fix before review"
|
||||
if tests.get("exit") not in (0, None) and not allow_test_failure:
|
||||
return BOUNCE, f"tests failed (exit={tests.get('exit')}) — see tests field"
|
||||
if build is not None and build.get("exit") not in (0, None):
|
||||
return BOUNCE, f"build failed (exit={build.get('exit')}) — see build field"
|
||||
if scope_violations:
|
||||
return BOUNCE, (f"scope violation: changed files outside the issue's "
|
||||
f"allowed set: {', '.join(scope_violations)}")
|
||||
if blocking:
|
||||
return BOUNCE, f"{len(blocking)} blocking finding(s) — see findings"
|
||||
if not acceptance:
|
||||
return NEEDS_GUIDANCE, ("no acceptance criteria parsed from the linked issue — "
|
||||
"cannot verify; never silent-approve")
|
||||
if tests.get("exit") is None:
|
||||
return NEEDS_GUIDANCE, "unverifiable test state (tests did not run)"
|
||||
return APPROVE, "all acceptance criteria checkable; tests/build/scope/hygiene green"
|
||||
|
||||
# never a merge call in this class
|
||||
320
tests/test_reviewer.py
Normal file
320
tests/test_reviewer.py
Normal file
@ -0,0 +1,320 @@
|
||||
"""Hermetic tests for the review gate (issue #5).
|
||||
|
||||
No network, no real subprocesses: FakeGiteaClient serves PR/issue/diff
|
||||
data in-memory and FakeRunner scripted exits. The only subprocess-ish seam
|
||||
tested is Runner's timeout handling, via an injected command runner.
|
||||
|
||||
Run with: python -m pytest tests/test_reviewer.py -q
|
||||
"""
|
||||
|
||||
import ast
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
# PEP 420 namespace package: make the project root importable regardless of
|
||||
# how pytest was invoked.
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from gitea_saas.reviewer import ( # noqa: E402
|
||||
APPROVE,
|
||||
BOUNCE,
|
||||
NEEDS_GUIDANCE,
|
||||
Reviewer,
|
||||
Runner,
|
||||
Verdict,
|
||||
parse_acceptance,
|
||||
parse_allowed_files,
|
||||
)
|
||||
|
||||
REVIEWER_SRC = Path(__file__).resolve().parent.parent / "gitea_saas" / "reviewer.py"
|
||||
|
||||
|
||||
class FakeGiteaClient:
|
||||
"""In-memory Gitea seam: PRs, issues and posted comments."""
|
||||
|
||||
def __init__(self, prs=None, issues=None):
|
||||
self.prs = prs or {} # number -> {"head_sha", "changed_files", "diff"}
|
||||
self.issues = issues or {} # number -> {"body", ...}
|
||||
self.comments = [] # (repo, number, body)
|
||||
|
||||
def get_pr(self, repo, number):
|
||||
pr = self.prs[number]
|
||||
return {"head_sha": pr["head_sha"], "changed_files": list(pr["changed_files"])}
|
||||
|
||||
def get_pr_diff(self, repo, number):
|
||||
return self.prs[number].get("diff", "")
|
||||
|
||||
def get_issue(self, repo, number):
|
||||
return dict(self.issues[number])
|
||||
|
||||
def post_comment(self, repo, number, body):
|
||||
self.comments.append((repo, number, body))
|
||||
|
||||
|
||||
class FakeRunner:
|
||||
"""Subprocess seam: scripted results, records commands."""
|
||||
|
||||
def __init__(self, results=None):
|
||||
self.results = list(results or []) # popped per run
|
||||
self.commands = []
|
||||
|
||||
def run(self, repo_dir, command, timeout_s=None):
|
||||
self.commands.append((str(repo_dir), command, timeout_s))
|
||||
if self.results:
|
||||
return self.results.pop(0)
|
||||
return {"exit": 0, "summary": ""}
|
||||
|
||||
|
||||
def sample_issue(acceptance=None, allowed_files=None, extra=""):
|
||||
body = "## Goal\nAutomated review gate.\n\n"
|
||||
if allowed_files is not None:
|
||||
body += ("**Files allowed:** " + ", ".join(f"`{f}`" for f in allowed_files)
|
||||
+ ". Do NOT modify anything else.\n\n")
|
||||
body += "## Acceptance\n"
|
||||
body += "".join(f"- [ ] {a}\n" for a in (acceptance or ["tests pass", "no regressions"]))
|
||||
body += extra
|
||||
return {"body": body}
|
||||
|
||||
|
||||
def sample_pr(diff="", files=None, head_sha="a" * 40):
|
||||
return {"head_sha": head_sha,
|
||||
"changed_files": files or ["gitea_saas/reviewer.py", "tests/test_reviewer.py"],
|
||||
"diff": diff}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Parsing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_parse_acceptance_checklist_and_section():
|
||||
body = ("## Goal\nx\n\n## Acceptance\n- [ ] first criterion\n- [X] second criterion\n"
|
||||
"- [x] third criterion\n\n## Notes\n- not acceptance\n")
|
||||
assert parse_acceptance(body) == ["first criterion", "second criterion", "third criterion"]
|
||||
assert parse_acceptance("no checklist here") == []
|
||||
|
||||
|
||||
def test_parse_allowed_files_backticks():
|
||||
body = "**Files allowed:** `gitea_saas/reviewer.py`, `gitea_saas/gitea_client.py` (extend), `tests/test_reviewer.py`. Do NOT modify anything else."
|
||||
assert parse_allowed_files(body) == [
|
||||
"gitea_saas/reviewer.py", "gitea_saas/gitea_client.py", "tests/test_reviewer.py"]
|
||||
assert parse_allowed_files("no constraint declared") == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# APPROVE / BOUNCE / NEEDS_GUIDANCE
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_approve_when_tests_pass_and_no_findings():
|
||||
client = FakeGiteaClient(prs={1: sample_pr(files=["gitea_saas/reviewer.py"])},
|
||||
issues={5: sample_issue(allowed_files=["gitea_saas/reviewer.py"])})
|
||||
runner = FakeRunner([{"exit": 0, "summary": "12 passed"}, # pytest
|
||||
{"exit": 0, "summary": "clean"}]) # git diff --check
|
||||
reviewer = Reviewer(client, runner, config={"secret_scan": True})
|
||||
|
||||
verdict = reviewer.review_pr("gitea-saas-core", 1, client.issues[5], "/fake/repo")
|
||||
|
||||
assert verdict.verdict == APPROVE
|
||||
assert verdict.tests == {"exit": 0, "summary": "12 passed"}
|
||||
assert verdict.acceptance == ["tests pass", "no regressions"]
|
||||
assert verdict.scope_violations == [] and verdict.findings == []
|
||||
assert verdict.head_sha == "a" * 40
|
||||
|
||||
comment = reviewer.post_verdict("gitea-saas-core", 1, verdict)
|
||||
assert "[review-gate] VERDICT: **APPROVE**" in comment
|
||||
assert "- [ ] tests pass" in comment # checklist present
|
||||
assert "No merge performed by the review gate" in comment
|
||||
assert client.comments and client.comments[0][:2] == ("gitea-saas-core", 1)
|
||||
|
||||
|
||||
def test_bounce_when_tests_fail():
|
||||
client = FakeGiteaClient(prs={1: sample_pr()}, issues={5: sample_issue()})
|
||||
runner = FakeRunner([{"exit": 1, "summary": "1 failed, 11 passed"}, {"exit": 0, "summary": "clean"}])
|
||||
verdict = Reviewer(client, runner).review_pr("gitea-saas-core", 1, client.issues[5], "/fake/repo")
|
||||
assert verdict.verdict == BOUNCE
|
||||
assert "tests failed (exit=1)" in verdict.summary
|
||||
# BOUNCE comment carries the test evidence
|
||||
reviewer = Reviewer(client, runner)
|
||||
comment = reviewer.post_verdict("gitea-saas-core", 1, verdict)
|
||||
assert "VERDICT: **BOUNCE**" in comment and "1 failed, 11 passed" in comment
|
||||
|
||||
|
||||
def test_bounce_on_scope_violation_even_with_green_tests():
|
||||
issue = sample_issue(allowed_files=["gitea_saas/reviewer.py"])
|
||||
pr = sample_pr(files=["gitea_saas/reviewer.py", "gitea_saas/server.py"])
|
||||
client = FakeGiteaClient(prs={1: pr}, issues={5: issue})
|
||||
runner = FakeRunner([{"exit": 0, "summary": "12 passed"}, {"exit": 0, "summary": "clean"}])
|
||||
verdict = Reviewer(client, runner).review_pr("gitea-saas-core", 1, issue, "/fake/repo")
|
||||
assert verdict.verdict == BOUNCE
|
||||
assert verdict.scope_violations == ["gitea_saas/server.py"]
|
||||
|
||||
|
||||
def test_needs_guidance_when_no_acceptance_criteria():
|
||||
issue = {"body": "## Goal\nnothing to verify here\n"}
|
||||
client = FakeGiteaClient(prs={1: sample_pr()}, issues={5: issue})
|
||||
runner = FakeRunner([{"exit": 0, "summary": "12 passed"}, {"exit": 0, "summary": "clean"}])
|
||||
verdict = Reviewer(client, runner).review_pr("gitea-saas-core", 1, issue, "/fake/repo")
|
||||
assert verdict.verdict == NEEDS_GUIDANCE
|
||||
assert "never silent-approve" in verdict.summary
|
||||
|
||||
|
||||
def test_allow_test_failure_in_issue_body_overrides_bounce():
|
||||
issue = sample_issue(extra="note: allow_test_failure for this transitional PR\n")
|
||||
client = FakeGiteaClient(prs={1: sample_pr()}, issues={5: issue})
|
||||
runner = FakeRunner([{"exit": 1, "summary": "known failure"}, {"exit": 0, "summary": "clean"}])
|
||||
verdict = Reviewer(client, runner).review_pr("gitea-saas-core", 1, issue, "/fake/repo")
|
||||
assert verdict.verdict == APPROVE # tolerated, acceptance present, nothing else blocking
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Findings quote the diff
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
SECRET_DIFF = (
|
||||
"diff --git a/gitea_saas/server.py b/gitea_saas/server.py\n"
|
||||
"--- a/gitea_saas/server.py\n+++ b/gitea_saas/server.py\n"
|
||||
"@@ -10,6 +10,8 @@\n"
|
||||
" def main():\n"
|
||||
"+ api_key = 'sk_live_1234567890ABCDEF'\n"
|
||||
"+ print('injected debug line')\n"
|
||||
"+ # TODO: replace placeholder\n"
|
||||
)
|
||||
|
||||
|
||||
def test_secret_finding_quotes_and_redacts_diff_line():
|
||||
issue = sample_issue(allowed_files=["gitea_saas/server.py"])
|
||||
pr = sample_pr(diff=SECRET_DIFF, files=["gitea_saas/server.py"])
|
||||
client = FakeGiteaClient(prs={1: pr}, issues={5: issue})
|
||||
runner = FakeRunner([{"exit": 0, "summary": "ok"}, {"exit": 0, "summary": "clean"}])
|
||||
verdict = Reviewer(client, runner).review_pr("gitea-saas-core", 1, issue, "/fake/repo")
|
||||
blocking = [f for f in verdict.findings if f["severity"] == "blocking"]
|
||||
assert blocking, "secret scan should produce a blocking finding"
|
||||
f = blocking[0]
|
||||
assert f["file"] == "gitea_saas/server.py"
|
||||
assert f["line"] == 11
|
||||
# Must quote the actual added line (with the secret value redacted).
|
||||
assert f["evidence_quote"].startswith(" api_key = '")
|
||||
assert "sk_live_" not in f["evidence_quote"] and "REDACTED" in f["evidence_quote"]
|
||||
assert verdict.verdict == BOUNCE
|
||||
# No unquoted assertion: every finding carries evidence_quote.
|
||||
assert all(fi.get("evidence_quote") for fi in verdict.findings)
|
||||
|
||||
|
||||
def test_todo_finding_is_warning_not_blocking():
|
||||
issue = sample_issue()
|
||||
pr = sample_pr(diff=SECRET_DIFF, files=["gitea_saas/server.py"])
|
||||
client = FakeGiteaClient(prs={1: pr}, issues={5: issue})
|
||||
runner = FakeRunner([{"exit": 0, "summary": "ok"}, {"exit": 0, "summary": "clean"}])
|
||||
verdict = Reviewer(client, runner, config={"secret_scan": False}).review_pr(
|
||||
"gitea-saas-core", 1, issue, "/fake/repo")
|
||||
todos = [f for f in verdict.findings if f["severity"] == "warning"]
|
||||
assert todos and todos[0]["note"] == "TODO/FIXME/HACK marker in added line"
|
||||
assert todos[0]["evidence_quote"] == " # TODO: replace placeholder"
|
||||
assert verdict.verdict == APPROVE # warnings alone do not bounce
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Timeouts
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_timeout_bounces_with_timed_out_exit():
|
||||
issue = sample_issue()
|
||||
client = FakeGiteaClient(prs={1: sample_pr()}, issues={5: issue})
|
||||
runner = FakeRunner([{"exit": "timed_out", "summary": "[timed out after 300s]"},
|
||||
{"exit": 0, "summary": "clean"}])
|
||||
verdict = Reviewer(client, runner).review_pr("gitea-saas-core", 1, issue, "/fake/repo")
|
||||
assert verdict.verdict == BOUNCE
|
||||
assert verdict.tests["exit"] == "timed_out"
|
||||
assert "timed out" in verdict.summary
|
||||
|
||||
|
||||
def test_runner_timeout_respected_without_real_subprocess():
|
||||
def exploding_runner(command, **kwargs):
|
||||
raise subprocess.TimeoutExpired(command, timeout=kwargs.get("timeout") or 0)
|
||||
|
||||
calls = []
|
||||
def spy(command, **kwargs):
|
||||
calls.append((command, kwargs.get("timeout")))
|
||||
return exploding_runner(command, **kwargs)
|
||||
|
||||
runner = Runner(timeout_s=300, command_runner=spy)
|
||||
result = runner.run("/fake/repo", "python -m pytest -q", timeout_s=7)
|
||||
assert result["exit"] == "timed_out"
|
||||
assert calls == [("python -m pytest -q", 7)] # timeout passed through
|
||||
|
||||
|
||||
def test_runner_default_timeout_used():
|
||||
seen = []
|
||||
def spy(command, **kwargs):
|
||||
seen.append(kwargs.get("timeout"))
|
||||
class C:
|
||||
returncode = 0
|
||||
stdout = "ok"
|
||||
stderr = ""
|
||||
return C()
|
||||
runner = Runner(timeout_s=120, command_runner=spy)
|
||||
result = runner.run("/fake/repo", "true")
|
||||
assert result["exit"] == 0 and seen == [120]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Build command / hygiene / misc
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_build_command_run_when_configured_and_build_failure_bounces():
|
||||
issue = sample_issue()
|
||||
client = FakeGiteaClient(prs={1: sample_pr()}, issues={5: issue})
|
||||
runner = FakeRunner([{"exit": 0, "summary": "12 passed"}, # pytest
|
||||
{"exit": 1, "summary": "build error"}, # build
|
||||
{"exit": 0, "summary": "clean"}]) # diff --check
|
||||
reviewer = Reviewer(client, runner, config={"build_cmd": "python -m compileall -q ."})
|
||||
verdict = reviewer.review_pr("gitea-saas-core", 1, issue, "/fake/repo")
|
||||
build = verdict.build
|
||||
assert build is not None
|
||||
assert build["exit"] == 1 and build["command"] == "python -m compileall -q ."
|
||||
assert verdict.verdict == BOUNCE
|
||||
assert "python -m compileall -q ." in "".join(runner.commands[1][1:2])
|
||||
|
||||
|
||||
def test_diff_hygiene_failure_is_blocking_finding():
|
||||
issue = sample_issue()
|
||||
client = FakeGiteaClient(prs={1: sample_pr()}, issues={5: issue})
|
||||
runner = FakeRunner([{"exit": 0, "summary": "12 passed"},
|
||||
{"exit": 1, "summary": "trailing whitespace at gitea_saas/x.py:3"}])
|
||||
verdict = Reviewer(client, runner).review_pr("gitea-saas-core", 1, issue, "/fake/repo")
|
||||
assert verdict.verdict == BOUNCE
|
||||
assert any("diff --check" in f["note"] for f in verdict.findings)
|
||||
|
||||
|
||||
def test_verdict_shape_and_comment_never_mentions_merge_action():
|
||||
issue = sample_issue()
|
||||
client = FakeGiteaClient(prs={1: sample_pr()}, issues={5: issue})
|
||||
runner = FakeRunner([{"exit": 0, "summary": "12 passed"}, {"exit": 0, "summary": "clean"}])
|
||||
verdict = Reviewer(client, runner).review_pr("gitea-saas-core", 1, issue, "/fake/repo")
|
||||
assert verdict.repo == "gitea-saas-core" and verdict.pr == 1
|
||||
assert isinstance(verdict.acceptance, list)
|
||||
assert isinstance(verdict.tests, dict) and "exit" in verdict.tests
|
||||
assert isinstance(verdict.scope_violations, list)
|
||||
assert isinstance(verdict.findings, list)
|
||||
assert verdict.verdict in (APPROVE, BOUNCE, NEEDS_GUIDANCE)
|
||||
comment = verdict.to_comment()
|
||||
assert "[review-gate]" in comment
|
||||
# Fail-closed: the comment only ever *disclaims* merging — never claims it.
|
||||
assert "No merge performed by the review gate" in comment
|
||||
assert "merged" not in comment
|
||||
|
||||
|
||||
def test_no_merge_call_anywhere_in_reviewer_src():
|
||||
tree = ast.parse(REVIEWER_SRC.read_text(encoding="utf-8"))
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
assert node.name != "merge", f"merge function found: {node.lineno}"
|
||||
if isinstance(node, ast.Call):
|
||||
func = node.func
|
||||
name = (func.id if isinstance(func, ast.Name) else
|
||||
func.attr if isinstance(func, ast.Attribute) else "")
|
||||
assert name != "merge", f"merge call found at {node.lineno}"
|
||||
if isinstance(node, ast.Attribute):
|
||||
assert node.attr != "merge", f"merge attribute access at {node.lineno}"
|
||||
Loading…
Reference in New Issue
Block a user