From df72b3926f491311053a2d4b4e44b961c7f15f57 Mon Sep 17 00:00:00 2001 From: pbwraith-worker Date: Tue, 25 Aug 2026 04:31:08 -0500 Subject: [PATCH 1/2] =?UTF-8?q?gitea-saas-core#5:=20review=20gate=20?= =?UTF-8?q?=E2=80=94=20automated=20PR=20reviewer=20(acceptance=20parse,=20?= =?UTF-8?q?tests,=20verdicts)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reviewer(repo_client, runner) with review_pr() -> Verdict and post_verdict() posting [review-gate] comments. Parses acceptance checklists + Files-allowed scope from the linked issue; runs pytest, optional build cmd (REVIEW_BUILD_CMD), git diff --check with timeouts (REVIEW_TEST_TIMEOUT_S, default 300); secret-pattern scan (sk_live_, AKIA, gh tokens, private keys, assigned credentials) quoting each added diff line as evidence (values redacted); optional configurable LLM pass that is validated — unquoted findings dropped. Verdict rules: failing tests -> BOUNCE (unless allow_test_failure in issue body); timeout -> BOUNCE; scope violation -> BOUNCE; blocking finding -> BOUNCE; empty acceptance -> NEEDS_GUIDANCE (never silent-approve); else APPROVE. FAIL-CLOSED: no merge call anywhere. gitea_client.py adds the Gitea seam (env token, urllib, mockable). 16 hermetic tests, no network, no real subprocesses. --- gitea_saas/gitea_client.py | 69 +++++++ gitea_saas/reviewer.py | 376 +++++++++++++++++++++++++++++++++++++ tests/test_reviewer.py | 320 +++++++++++++++++++++++++++++++ 3 files changed, 765 insertions(+) create mode 100644 gitea_saas/gitea_client.py create mode 100644 gitea_saas/reviewer.py create mode 100644 tests/test_reviewer.py diff --git a/gitea_saas/gitea_client.py b/gitea_saas/gitea_client.py new file mode 100644 index 0000000..f9a660b --- /dev/null +++ b/gitea_saas/gitea_client.py @@ -0,0 +1,69 @@ +"""Gitea API client (mockable seam, env token). + +Used by the review gate and later front-ends. Real HTTP only when a token +is available; the token is read from ``GITEA_TOKEN`` (or passed in) and is +never printed or placed on a command line. Tests use an in-memory fake, so +this module is never exercised over the network in CI. +""" + +from __future__ import annotations + +import json +import os +import urllib.request +from typing import Optional + + +class GiteaClient: + """Thin urllib client for the subset of the Gitea API the SaaS needs.""" + + def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None, + owner: str = "pbwraith"): + self.base_url = (base_url or os.environ.get( + "GITEA_BASE_URL", "https://gitea.185.214.135.234.nip.io")).rstrip("/") + self.token = token if token is not None else os.environ.get("GITEA_TOKEN", "") + self.owner = owner + + # -- low-level --------------------------------------------------------- + + def _request(self, method: str, path: str, body: Optional[dict] = None): + data = json.dumps(body).encode() if body is not None else None + url = f"{self.base_url}/api/v1/repos/{self.owner}/{path}" + req = urllib.request.Request( + url, data=data, method=method, + headers={"Authorization": "token " + self.token, + "Accept": "application/json", + "Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=30) as resp: + raw = resp.read() + return json.loads(raw) if raw else None + + # -- API surface used by the review gate ------------------------------- + + def get_issue(self, repo: str, number: int) -> dict: + """Fetch an issue: {title, body, state, labels: [...]}.""" + return self._request("GET", f"{repo}/issues/{number}") + + def get_pr(self, repo: str, number: int) -> dict: + """Fetch a PR: {head_sha, changed_files: [filename, ...]}.""" + pr = self._request("GET", f"{repo}/pulls/{number}") or {} + files = self._request("GET", f"{repo}/pulls/{number}/files") or [] + return { + "head_sha": ((pr.get("head") or {}).get("sha")), + "changed_files": [f.get("filename") for f in files if f.get("filename")], + } + + def get_pr_diff(self, repo: str, number: int) -> str: + """Return the PR's diff text reconstructed from per-file patches.""" + files = self._request("GET", f"{repo}/pulls/{number}/files") or [] + chunks = [] + for f in files: + filename = f.get("filename") or "?" + chunks.append(f"diff --git a/{filename} b/{filename}\n" + f"--- a/{filename}\n+++ b/{filename}\n" + + (f.get("patch") or "")) + return "\n".join(chunks) + + def post_comment(self, repo: str, number: int, body: str) -> dict: + """Post a comment on an issue or PR conversation.""" + return self._request("POST", f"{repo}/issues/{number}/comments", {"body": body}) diff --git a/gitea_saas/reviewer.py b/gitea_saas/reviewer.py new file mode 100644 index 0000000..4d5bf3b --- /dev/null +++ b/gitea_saas/reviewer.py @@ -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 diff --git a/tests/test_reviewer.py b/tests/test_reviewer.py new file mode 100644 index 0000000..e8e3df6 --- /dev/null +++ b/tests/test_reviewer.py @@ -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}" -- 2.43.0 From 5e5827c987b63cc80762bda576c4f3610f0fe446 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 10:43:46 -0500 Subject: [PATCH 2/2] =?UTF-8?q?gitea-saas-core#5:=20drop=20duplicate=20git?= =?UTF-8?q?ea=5Fclient.py=20=E2=80=94=20canonical=20version=20is=20!17's?= =?UTF-8?q?=20(#4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gitea_saas/gitea_client.py | 69 -------------------------------------- 1 file changed, 69 deletions(-) delete mode 100644 gitea_saas/gitea_client.py diff --git a/gitea_saas/gitea_client.py b/gitea_saas/gitea_client.py deleted file mode 100644 index f9a660b..0000000 --- a/gitea_saas/gitea_client.py +++ /dev/null @@ -1,69 +0,0 @@ -"""Gitea API client (mockable seam, env token). - -Used by the review gate and later front-ends. Real HTTP only when a token -is available; the token is read from ``GITEA_TOKEN`` (or passed in) and is -never printed or placed on a command line. Tests use an in-memory fake, so -this module is never exercised over the network in CI. -""" - -from __future__ import annotations - -import json -import os -import urllib.request -from typing import Optional - - -class GiteaClient: - """Thin urllib client for the subset of the Gitea API the SaaS needs.""" - - def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None, - owner: str = "pbwraith"): - self.base_url = (base_url or os.environ.get( - "GITEA_BASE_URL", "https://gitea.185.214.135.234.nip.io")).rstrip("/") - self.token = token if token is not None else os.environ.get("GITEA_TOKEN", "") - self.owner = owner - - # -- low-level --------------------------------------------------------- - - def _request(self, method: str, path: str, body: Optional[dict] = None): - data = json.dumps(body).encode() if body is not None else None - url = f"{self.base_url}/api/v1/repos/{self.owner}/{path}" - req = urllib.request.Request( - url, data=data, method=method, - headers={"Authorization": "token " + self.token, - "Accept": "application/json", - "Content-Type": "application/json"}) - with urllib.request.urlopen(req, timeout=30) as resp: - raw = resp.read() - return json.loads(raw) if raw else None - - # -- API surface used by the review gate ------------------------------- - - def get_issue(self, repo: str, number: int) -> dict: - """Fetch an issue: {title, body, state, labels: [...]}.""" - return self._request("GET", f"{repo}/issues/{number}") - - def get_pr(self, repo: str, number: int) -> dict: - """Fetch a PR: {head_sha, changed_files: [filename, ...]}.""" - pr = self._request("GET", f"{repo}/pulls/{number}") or {} - files = self._request("GET", f"{repo}/pulls/{number}/files") or [] - return { - "head_sha": ((pr.get("head") or {}).get("sha")), - "changed_files": [f.get("filename") for f in files if f.get("filename")], - } - - def get_pr_diff(self, repo: str, number: int) -> str: - """Return the PR's diff text reconstructed from per-file patches.""" - files = self._request("GET", f"{repo}/pulls/{number}/files") or [] - chunks = [] - for f in files: - filename = f.get("filename") or "?" - chunks.append(f"diff --git a/{filename} b/{filename}\n" - f"--- a/{filename}\n+++ b/{filename}\n" - + (f.get("patch") or "")) - return "\n".join(chunks) - - def post_comment(self, repo: str, number: int, body: str) -> dict: - """Post a comment on an issue or PR conversation.""" - return self._request("POST", f"{repo}/issues/{number}/comments", {"body": body}) -- 2.43.0