464 lines
24 KiB
Python
464 lines
24 KiB
Python
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"<!-- werkrevisor:{HEAD} -->\n<!-- werkrevisor-verdict:{verdict} -->"})
|
|
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"<!-- werkrevisor:{other} -->\n<!-- werkrevisor-verdict:GREEN -->"})
|
|
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("<!-- werkrevisor-codex-critical:false -->", 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"<!-- werkrevisor:{HEAD} -->\n<!-- werkrevisor-verdict:GREEN -->"})
|
|
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)
|