import dataclasses import io import json import os import tempfile import unittest import unittest.mock from werkrevisor import codex, codexrunner from werkrevisor.gitea import GiteaError HEAD, BASE, BOT = "h" * 40, "b" * 40, "ai-reviewer" FINDING = {"severity": "high", "file": "src/A.java", "line": 12, "title": "Lost grant", "explanation": "Why."} class FakeGitea: def __init__(self, mode="gate", title="Title", werkrevisor_result=True, verdict="GREEN", result_at="2026-10-11T10:00:00Z"): self.mode = mode self.title = title self.comments = [] if werkrevisor_result: self.comments.append({"id": 1, "user": {"login": BOT}, "created_at": result_at, "body": f"\n"}) self.reviews = [] self.review_comments = {} self.statuses = [] self.created_reviews = [] self.refuse_line_comments = False def get_repo(self): return {"default_branch": "main"} def get_branch(self, name): return {"commit": {"id": "m" * 40}} def get_raw_file(self, path, ref): if self.mode is None: raise GiteaError("GET raw failed with HTTP 404", 404) return f'[codex]\nmode = "{self.mode}"\n'.encode() def current_user(self): return {"login": BOT} def list_open_prs(self): return [{"number": 7, "title": self.title, "body": "Adds a scope.", "head": {"sha": HEAD}, "base": {"sha": BASE}, "merge_base": BASE}] def list_issue_comments(self, number): return self.comments def create_issue_comment(self, number, body): comment = {"id": len(self.comments) + 1, "user": {"login": BOT}, "body": body, "created_at": "2026-10-11T10:05:00Z"} self.comments.append(comment) return comment def list_pr_files(self, number): return [{"filename": "src/A.java"}] def create_review(self, number, event, body, commit_id, comments=()): if self.refuse_line_comments: raise GiteaError("POST reviews failed with HTTP 422", 422) review_id = len(self.reviews) + 100 self.reviews.append({"id": review_id, "user": {"login": BOT}, "state": "COMMENT", "commit_id": commit_id}) self.review_comments[review_id] = [{"user": {"login": BOT}, "resolver": None, "position": c["new_position"], "body": c["body"]} for c in comments] self.created_reviews.append((event, body, list(comments))) def list_reviews(self, number): return self.reviews def list_review_comments(self, number, review_id): return self.review_comments.get(review_id, []) def list_statuses(self, sha): return [{"context": c, "status": s, "description": d} for c, s, d in reversed(self.statuses)] def set_status(self, sha, state, context, description, target_url=""): self.statuses.append((context, state, description)) def approve(self, login="mi", submitted_at="2026-10-11T10:10:00Z"): self.reviews.append({"id": 900, "user": {"login": login}, "state": "APPROVED", "commit_id": HEAD, "submitted_at": submitted_at}) def record(critical=False, findings=(), exit_code=0, error="quota", effect="none", fails_closed=False, tested=False): return {"model": "gpt-6-astra", "effort": "high", "exit_code": exit_code, "seconds": 90.0, "usage": {"input_tokens": 1000, "cached_input_tokens": 800, "output_tokens": 50, "reasoning_output_tokens": 10}, "commands": ["git diff"], "errors": [] if exit_code == 0 else [error], "answer": {"critical": critical, "critical_reason": "Changes RBAC.", "security_effect": effect, "fails_closed": fails_closed, "hardening_tested": tested, "findings": list(findings)}} class Harness: def __init__(self, test, gitea, records): """`gitea` is a FakeGitea for the repository team/app, or a dict of them by repository.""" self.repositories = gitea if isinstance(gitea, dict) else {"team/app": gitea} self.gitea = next(iter(self.repositories.values())) self.records = list(records) self.runs = 0 self.reviewed = [] self.sandbox = None self.result = None self.tmp = tempfile.TemporaryDirectory() test.addCleanup(self.tmp.cleanup) server = codexrunner.Server("https://git.example.org", "GITEA_BOT_TOKEN", ("mi",), tuple(self.repositories)) self.config = codexrunner.RunnerConfig("gpt-6-astra", "high", 60, self.tmp.name, 3, (server,)) self.out = io.StringIO() def run(self): runner = codexrunner.Runner(self.config, {"GITEA_BOT_TOKEN": "secret-token-123", "HOME": self.tmp.name}, out=self.out, client=lambda base_url, owner, repo, token: self.repositories[f"{owner}/{repo}"], sandbox_check=lambda: self.sandbox) original_checkout, original_run = codex.checkout, codex.run def fake_checkout(base_url, owner, repo, *args): self.reviewed.append(f"{owner}/{repo}") return self.tmp.name def fake_run(*args): self.runs += 1 return self.records.pop(0) codex.checkout, codex.run = fake_checkout, fake_run try: self.result = runner.run() finally: codex.checkout, codex.run = original_checkout, original_run return self def attempts(self): try: with open(os.path.join(self.tmp.name, "attempts.json")) as f: return json.load(f) except FileNotFoundError: return {} def status(self): return self.gitea.statuses[-1][1:] if self.gitea.statuses else None class CodexRunnerTest(unittest.TestCase): def test_a_repository_without_a_codex_mode_gets_no_review(self): for mode in (None, "off"): with self.subTest(mode=mode): h = Harness(self, FakeGitea(mode=mode), [record()]).run() self.assertEqual((h.runs, h.gitea.statuses, len(h.gitea.comments)), (0, [], 1)) def test_each_ready_pr_is_checked_for_a_held_merge_also_without_a_codex_mode(self): """Scenario#135.01""" for mode in (None, "off", "gate"): with self.subTest(mode=mode), unittest.mock.patch.object( codexrunner.cli, "release_held_merge", return_value="merge scheduled") as release: h = Harness(self, FakeGitea(mode=mode), [record()]).run() [(gate, ref, gitea, login, _)] = [c.args for c in release.call_args_list] self.assertEqual((gate.automerge, gate.human_reviewers, ref.number, ref.head_sha, gitea, login), (True, ("mi",), 7, HEAD, h.gitea, BOT)) self.assertIn("team/app#7: held merge: merge scheduled", h.out.getvalue()) def test_the_codex_status_is_up_to_date_before_a_release(self): h = Harness(self, FakeGitea(), [record(findings=[FINDING])]).run() for comments in h.gitea.review_comments.values(): for c in comments: c["resolver"] = {"login": "claude-code"} h.run() self.assertEqual(h.status(), ("success", codexrunner.PASSED)) for comments in h.gitea.review_comments.values(): for c in comments: c["resolver"] = None seen = [] with unittest.mock.patch.object(codexrunner.cli, "release_held_merge", side_effect=lambda *args: seen.append(h.status()) or ""): h.run() self.assertEqual(seen, [("pending", "author ✍️: resolve 1 open Codex finding(s)")]) def test_no_release_when_the_codex_status_cannot_be_brought_up_to_date(self): gitea = FakeGitea() gitea.list_issue_comments = unittest.mock.Mock(side_effect=OSError("HTTP 502")) with unittest.mock.patch.object(codexrunner.cli, "release_held_merge", return_value="") as release: Harness(self, gitea, [record()]).run() release.assert_not_called() def test_a_wip_pr_is_not_released(self): with unittest.mock.patch.object(codexrunner.cli, "release_held_merge", return_value="") as release: Harness(self, FakeGitea(title="WIP: Title"), [record()]).run() release.assert_not_called() def test_a_failed_release_stops_not_the_codex_review(self): with unittest.mock.patch.object(codexrunner.cli, "release_held_merge", side_effect=OSError("HTTP 502")): h = Harness(self, FakeGitea(), [record()]).run() self.assertIn("team/app#7: merge not released: OSError: HTTP 502", h.out.getvalue()) self.assertEqual(h.status(), ("success", codexrunner.PASSED)) def test_a_pr_waits_for_werkrevisor_and_its_wip_end(self): for gitea in (FakeGitea(werkrevisor_result=False), FakeGitea(title="WIP: Title")): with self.subTest(title=gitea.title): h = Harness(self, gitea, [record()]).run() self.assertEqual(h.runs, 0) self.assertEqual(Harness(self, FakeGitea(werkrevisor_result=False), []).run().status(), ("pending", codexrunner.QUEUED)) def test_a_red_head_gets_no_codex_review(self): h = Harness(self, FakeGitea(verdict="RED"), [record()]).run() self.assertEqual((h.runs, len(h.gitea.comments)), (0, 1)) self.assertEqual(h.status(), ("pending", codexrunner.RED)) def test_a_run_reviews_at_most_max_reviews_per_run(self): other = "g" * 40 for limit in (0, 1): with self.subTest(limit=limit): gitea = FakeGitea() gitea.comments.append({"id": 2, "user": {"login": BOT}, "created_at": "2026-10-11T10:00:00Z", "body": f"\n"}) first = gitea.list_open_prs()[0] gitea.list_open_prs = lambda: [first, {**first, "number": 8, "head": {"sha": other}}] h = Harness(self, gitea, [record(), record()]) h.config = dataclasses.replace(h.config, max_reviews_per_run=limit) self.assertEqual(h.run().runs, limit) def test_the_pr_that_waits_longest_goes_first(self): # 12:01+02:00 is 10:01Z: before 10:05Z, although it sorts after it as text. repositories = {"team/app": FakeGitea(result_at="2026-10-11T10:05:00Z"), "team/lib": FakeGitea(result_at="2026-10-11T12:01:00+02:00")} h = Harness(self, repositories, [record()]) h.config = dataclasses.replace(h.config, max_reviews_per_run=1) h.run() self.assertEqual(h.reviewed, ["team/lib"]) self.assertEqual(repositories["team/lib"].statuses[-1][1:], ("success", codexrunner.PASSED)) self.assertEqual(repositories["team/app"].statuses[-1][1:], ("pending", codexrunner.QUEUED)) def test_a_pr_without_a_readable_result_time_waits_behind_the_others(self): repositories = {"team/app": FakeGitea(result_at=None), "team/lib": FakeGitea(result_at="2026-10-11T10:05:00")} h = Harness(self, repositories, [record()]) h.config = dataclasses.replace(h.config, max_reviews_per_run=1) self.assertEqual(h.run().reviewed, ["team/lib"]) def test_a_failed_pr_keeps_its_place(self): repositories = {"team/app": FakeGitea(result_at="2026-10-11T10:05:00Z"), "team/lib": FakeGitea(result_at="2026-10-11T10:00:00Z")} h = Harness(self, repositories, [record(exit_code=1), record()]) h.config = dataclasses.replace(h.config, max_reviews_per_run=1) h.run().run() self.assertEqual(h.reviewed, ["team/lib", "team/lib"]) self.assertEqual(repositories["team/lib"].statuses[-1][1:], ("success", codexrunner.PASSED)) def test_a_failing_repository_stops_neither_the_others_nor_their_order(self): broken = FakeGitea() broken.list_open_prs = lambda: (_ for _ in ()).throw(GiteaError("GET pulls failed with HTTP 500", 500)) h = Harness(self, {"team/broken": broken, "team/app": FakeGitea()}, [record()]).run() self.assertEqual(h.reviewed, ["team/app"]) self.assertIn("codex-runner: team/broken: GiteaError", h.out.getvalue()) def test_a_failing_pr_hides_not_the_other_ready_prs_of_its_repository(self): gitea = FakeGitea() first = gitea.list_open_prs()[0] gitea.list_open_prs = lambda: [first, {**first, "number": 8, "head": {"sha": "g" * 40}}] list_issue_comments = gitea.list_issue_comments def failing_for_8(number): if number == 8: raise GiteaError("GET comments failed with HTTP 500", 500) return list_issue_comments(number) gitea.list_issue_comments = failing_for_8 h = Harness(self, gitea, [record()]).run() self.assertEqual(h.reviewed, ["team/app"]) self.assertIn("codex-runner: team/app#8: GiteaError", h.out.getvalue()) def test_a_failing_review_stops_not_the_next_one(self): repositories = {"team/app": FakeGitea(result_at="2026-10-11T10:00:00Z"), "team/lib": FakeGitea(result_at="2026-10-11T10:05:00Z")} repositories["team/app"].list_pr_files = lambda number: (_ for _ in ()).throw( GiteaError("GET files failed with HTTP 500", 500)) h = Harness(self, repositories, [record(), record()]).run() self.assertEqual(h.reviewed, ["team/app", "team/lib"]) self.assertIn("codex-runner: team/app#7: GiteaError", h.out.getvalue()) def test_a_broken_sandbox_stops_the_reviews(self): reviewed = FakeGitea(result_at="2026-10-11T09:00:00Z") repositories = {"team/app": FakeGitea(), "team/lib": reviewed} h = Harness(self, repositories, [record(findings=[FINDING]), record()]) h.config = dataclasses.replace(h.config, max_reviews_per_run=1) h.run() self.assertEqual(reviewed.statuses[-1][2], "author ✍️: resolve 1 open Codex finding(s)") h.sandbox = "bwrap: setting up uid map: Permission denied" for comments in reviewed.review_comments.values(): for c in comments: c["resolver"] = {"login": "claude-code"} h.run() self.assertEqual((h.runs, h.result, h.attempts()), (1, False, {})) self.assertEqual(repositories["team/app"].statuses[-1][1:], ("pending", codexrunner.SANDBOX)) self.assertEqual(reviewed.statuses[-1][1:], ("success", codexrunner.PASSED)) log = h.out.getvalue() self.assertIn("reviews nothing and counts no attempt: bwrap: setting up uid map: Permission denied", log) self.assertIn("journalctl -k | grep -E 'unpriv_bwrap|unprivileged_userns'", log) self.assertIn("/etc/apparmor.d/bwrap", log) h.sandbox = None h.run() self.assertEqual((h.runs, h.result), (2, True)) def test_a_jail_fault_in_a_review_costs_no_attempt(self): repositories = {"team/app": FakeGitea(result_at="2026-10-11T10:00:00Z"), "team/lib": FakeGitea(result_at="2026-10-11T10:05:00Z")} h = Harness(self, repositories, [record(exit_code=1, error="bwrap: setting up uid map: Permission denied"), record(), record()]) h.run() self.assertEqual((h.reviewed, h.result, h.attempts()), (["team/app"], False, {})) for gitea in repositories.values(): self.assertEqual(gitea.statuses[-1][1:], ("pending", codexrunner.SANDBOX)) h.run() self.assertEqual((h.reviewed, h.result), (["team/app", "team/app", "team/lib"], True)) def test_the_run_exits_with_1_after_a_sandbox_fault(self): config = os.path.join(self.enterContext(tempfile.TemporaryDirectory()), "codex-runner.toml") with open(config, "w") as f: f.write(f'work_dir = "{os.path.dirname(config)}"\n') for result, code in ((True, 0), (False, 1)): with self.subTest(result=result), \ unittest.mock.patch.object(codexrunner, "Runner") as runner: runner.return_value.run.return_value = result self.assertEqual(codexrunner.main(["--config", config], environ={}), code) def test_advise_comments_once_and_sets_no_status(self): h = Harness(self, FakeGitea(mode="advise"), [record(critical=True, findings=[FINDING])]).run() self.assertEqual(h.gitea.statuses, []) [(event, _, [comment])] = h.gitea.created_reviews self.assertEqual((event, comment["path"], comment["new_position"]), ("COMMENT", "src/A.java", 12)) self.assertIn(codex.COMMENT_MARKER, comment["body"]) body = h.gitea.comments[-1]["body"] self.assertIn(codexrunner.head_marker(HEAD), body) self.assertIn("## Codex review: critical", body) self.assertIn("only advises", body) h.run() self.assertEqual(h.runs, 1) def test_gate_passes_a_pr_without_findings(self): h = Harness(self, FakeGitea(), [record()]).run() self.assertEqual(h.status(), ("success", codexrunner.PASSED)) def test_gate_holds_until_each_line_comment_is_resolved(self): h = Harness(self, FakeGitea(), [record(findings=[FINDING])]).run() self.assertEqual(h.status(), ("pending", "author ✍️: resolve 1 open Codex finding(s)")) for comments in h.gitea.review_comments.values(): for c in comments: c["resolver"] = {"login": "claude-code"} h.run() self.assertEqual((h.runs, h.status()), (1, ("success", codexrunner.PASSED))) def test_a_critical_pr_waits_for_a_later_human_approval(self): h = Harness(self, FakeGitea(), [record(critical=True)]).run() self.assertEqual(h.status(), ("pending", codexrunner.HUMAN)) h.gitea.approve(submitted_at="2026-10-11T10:01:00Z") h.run() self.assertEqual(h.status()[0], "pending") h.gitea.approve() h.run() self.assertEqual(h.status(), ("success", codexrunner.PASSED)) def test_a_tested_hardening_that_fails_closed_needs_no_human(self): h = Harness(self, FakeGitea(), [record(critical=True, effect="hardens", fails_closed=True, tested=True)]).run() body = h.gitea.comments[-1]["body"] self.assertIn("## Codex review: not critical: a tested hardening that fails closed", body) self.assertIn("Security effect: hardens; fails closed: yes; a test of the closed case: yes.", body) self.assertIn("", body) self.assertEqual(h.status(), ("success", codexrunner.PASSED)) def test_any_other_critical_pr_waits_for_a_human(self): for effect, fails_closed, tested in (("weakens", True, True), ("mixed", True, True), ("none", True, True), ("hardens", False, True), ("hardens", True, False)): with self.subTest(effect=effect, fails_closed=fails_closed, tested=tested): h = Harness(self, FakeGitea(), [record(critical=True, effect=effect, fails_closed=fails_closed, tested=tested)]).run() self.assertIn("## Codex review: critical\n", h.gitea.comments[-1]["body"]) self.assertEqual(h.status(), ("pending", codexrunner.HUMAN)) def test_a_hardening_still_waits_for_its_line_comments(self): h = Harness(self, FakeGitea(), [record(critical=True, findings=[FINDING], effect="hardens", fails_closed=True, tested=True)]).run() self.assertEqual(h.status(), ("pending", "author ✍️: resolve 1 open Codex finding(s)")) def test_the_approval_of_werkrevisor_is_no_human_approval(self): h = Harness(self, FakeGitea(), [record(critical=True)]).run() h.gitea.approve(login=BOT) h.run() self.assertEqual(h.status(), ("pending", codexrunner.HUMAN)) def test_a_finding_outside_the_diff_needs_a_human(self): gitea = FakeGitea() gitea.refuse_line_comments = True h = Harness(self, gitea, [record(findings=[FINDING])]).run() self.assertIn("only here", gitea.comments[-1]["body"]) self.assertIn("Why.", gitea.comments[-1]["body"]) self.assertEqual(h.status()[0], "pending") def test_a_failed_run_is_retried_and_then_reported(self): h = Harness(self, FakeGitea(), [record(exit_code=1)] * 3).run() self.assertEqual((h.runs, len(h.gitea.comments), h.status()), (1, 1, ("pending", codexrunner.RUNNING))) h.run().run() self.assertEqual(h.runs, 3) self.assertIn(codexrunner.error_marker(HEAD), h.gitea.comments[-1]["body"]) self.assertEqual(h.status(), ("error", codexrunner.FAILED)) h.gitea.approve() h.run() self.assertEqual((h.runs, h.status()), (3, ("success", codexrunner.FAILED_APPROVED))) def test_a_status_that_waits_for_a_human_starts_with_human(self): """Scenario#138.01""" h = Harness(self, FakeGitea(), [record(critical=True)]).run() self.assertTrue(h.status()[1].startswith("human 👤: "), h.status()) def test_each_description_starts_with_who_acts_next(self): """Scenario#138.02""" h = Harness(self, FakeGitea(werkrevisor_result=False), []).run() h.gitea.comments.append({"id": 1, "user": {"login": BOT}, "created_at": "2026-10-11T10:00:00Z", "body": f"\n"}) h.records = [record(findings=[FINDING])] h.run() for comments in h.gitea.review_comments.values(): for c in comments: c["resolver"] = {"login": "claude-code"} h.run() seen = [(state, d) for context, state, d in h.gitea.statuses] self.assertEqual(seen, [("pending", codexrunner.QUEUED), ("pending", codexrunner.RUNNING), ("pending", "author ✍️: resolve 1 open Codex finding(s)"), ("success", codexrunner.PASSED)]) def test_the_descriptions_are_the_documented_ones(self): """Scenario#138.02: literal, so that a changed text or prefix of a constant fails here.""" self.assertEqual([codexrunner.QUEUED, codexrunner.RED, codexrunner.SANDBOX, codexrunner.RUNNING, codexrunner.HUMAN, codexrunner.FAILED, codexrunner.PASSED, codexrunner.FAILED_APPROVED], ["queued ⏳: Codex reviews after werkrevisor", "author ✍️: werkrevisor found RED; push a commit that fixes the cause", "operator 🔧: the Codex sandbox fails on the workstation; see the runner log", "running ⚙️: Codex review", "human 👤: approve the head to unlock the merge", "human 👤: Codex review failed; approve the head to merge", "done ✅: Codex review passed", "done ☑️: Codex review failed; a human approved the head"]) def test_an_unchanged_status_is_not_repeated(self): h = Harness(self, FakeGitea(), [record()]).run() count = len(h.gitea.statuses) h.run() self.assertEqual(len(h.gitea.statuses), count) runner = codexrunner.Runner(h.config, {}) long = "x" * 300 runner.set_status(h.gitea, HEAD, "pending", long) h.gitea.statuses[-1] = (codexrunner.STATUS_CONTEXT, "pending", long[:250]) runner.set_status(h.gitea, HEAD, "pending", long) self.assertEqual(len(h.gitea.statuses), count + 1) def test_no_token_or_login_reaches_the_pr(self): leaked = dict(FINDING, explanation="token secret-token-123 and login " + "L" * 30) h = Harness(self, FakeGitea(mode="advise"), [record(findings=[leaked])]) os.makedirs(os.path.join(h.tmp.name, ".codex")) with open(os.path.join(h.tmp.name, ".codex", "auth.json"), "w") as f: f.write('{"tokens": {"access_token": "' + "L" * 30 + '"}}') h.run() posted = [h.gitea.comments[-1]["body"]] + [c["body"] for _, _, cs in h.gitea.created_reviews for c in cs] for text in posted: self.assertNotIn("secret-token-123", text) self.assertNotIn("L" * 30, text)