284 lines
14 KiB
Python
284 lines
14 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
|
|
from tests.fakes import FakeGitea
|
|
from werkrevisor import codex, reviewreport
|
|
|
|
HEAD = "6d1842aa2111a50cbb0614aa840ed1cb8a75985c"
|
|
DRY_RUN = f"""<!-- werkrevisor:{HEAD} -->
|
|
<!-- werkrevisor-verdict:YELLOW -->
|
|
## 🟡 werkrevisor: YELLOW / HUMAN_REVIEW
|
|
|
|
Change kind: `mixed`.
|
|
|
|
### Reasons
|
|
- **YELLOW** risky paths changed: build.gradle.kts
|
|
|
|
<sub>
|
|
Usage: model `m`, cost 0.17 USD (reported).<br>
|
|
Reviewed head `{HEAD}`.
|
|
</sub>
|
|
|
|
{{"verdict": "YELLOW"}}
|
|
"""
|
|
FINDING = {"severity": "medium", "file": "tools/x", "line": 148, "title": "Claims collapse", "explanation": "Why."}
|
|
|
|
|
|
def record(answer, exit_code=0):
|
|
return {"model": "gpt-6-astra", "effort": "high", "exit_code": exit_code, "seconds": 72.2,
|
|
"usage": {"input_tokens": 311095, "cached_input_tokens": 256000, "output_tokens": 2143,
|
|
"reasoning_output_tokens": 643}, "commands": [], "errors": [], "answer": answer}
|
|
|
|
|
|
class ReviewReportTest(unittest.TestCase):
|
|
def test_the_werkrevisor_part_has_no_markers(self):
|
|
title, body, usage = reviewreport.werkrevisor_part(DRY_RUN)
|
|
self.assertEqual(title, "🟡 YELLOW / HUMAN_REVIEW")
|
|
self.assertIn("#### Reasons", body)
|
|
self.assertNotIn("<!--", body)
|
|
self.assertNotIn('"verdict"', body)
|
|
self.assertEqual(usage, "model `m`, cost 0.17 USD (reported).")
|
|
self.assertIsNone(reviewreport.werkrevisor_part("werkrevisor: configuration error"))
|
|
|
|
def test_the_report_holds_both_reviews_and_asks_for_a_check(self):
|
|
answer = {"critical": True, "critical_reason": "Changes shared test infrastructure.", "findings": [FINDING]}
|
|
report = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN),
|
|
reviewreport.codex_part(record(answer)))
|
|
self.assertTrue(report.startswith("## AI review of `6d1842aa` (werkrevisor and Codex)"))
|
|
self.assertIn("check every finding in the code before posting", report)
|
|
self.assertIn("### werkrevisor: 🟡 YELLOW / HUMAN_REVIEW", report)
|
|
self.assertIn("Codex rates the PR as critical: Changes shared test infrastructure.", report)
|
|
self.assertIn("1. **[medium] Claims collapse** at `tools/x:148`", report)
|
|
self.assertIn("input tokens 311095 (256000 cached)", report)
|
|
|
|
def test_a_failed_review_is_named_not_left_out(self):
|
|
report = reviewreport.compose(HEAD, None, reviewreport.codex_part(record({"unparsed": "x"})),
|
|
"docker: not found", "quota exceeded")
|
|
self.assertIn("### werkrevisor\n\nThe review did not complete: docker: not found", report)
|
|
self.assertIn("### Codex\n\nThe review did not complete: quota exceeded", report)
|
|
self.assertIsNone(reviewreport.codex_part(record({"critical": False, "critical_reason": "", "findings": []},
|
|
exit_code=1)))
|
|
self.assertTrue(codex.valid_answer({"critical": False, "critical_reason": "", "findings": []}))
|
|
|
|
def test_an_empty_reason_leaves_no_colon(self):
|
|
text, _ = reviewreport.codex_part(record({"critical": False, "critical_reason": " ", "findings": []}))
|
|
self.assertTrue(text.startswith("Codex rates the PR as not critical.\n"))
|
|
|
|
def test_each_error_is_cut_on_its_own(self):
|
|
self.assertEqual(codex.error_text(["a" * 600, "first", "b" * 10 + "end"], limit=5), "first; bbend")
|
|
|
|
def test_reviews_of_different_commits_are_flagged(self):
|
|
other = "f" * 40
|
|
self.assertEqual(reviewreport.reviewed_head(DRY_RUN), HEAD)
|
|
werkrevisor = reviewreport.werkrevisor_part(DRY_RUN)
|
|
self.assertNotIn("Warning", reviewreport.compose(HEAD, werkrevisor, None, werkrevisor_head=HEAD))
|
|
self.assertIn("> **Warning:** werkrevisor reviewed `6d1842aa`, Codex `ffffffff`",
|
|
reviewreport.compose(other, werkrevisor, None, werkrevisor_head=HEAD))
|
|
|
|
def test_a_failed_codex_run_keeps_the_head_and_warns_of_no_mismatch(self):
|
|
report = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None, codex_error="timeout",
|
|
werkrevisor_head=None)
|
|
self.assertTrue(report.startswith("## AI review of `6d1842aa`"))
|
|
self.assertNotIn("Warning", report)
|
|
|
|
def test_the_draft_links_the_diagrams_next_to_it(self):
|
|
"""Scenario#129.02"""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
source, target = os.path.join(tmp, "drawn"), os.path.join(tmp, "draft")
|
|
os.makedirs(source)
|
|
os.makedirs(target)
|
|
for name in ("pr-lens-flow.svg", "pr-lens-containers.svg", "notes.txt"):
|
|
open(os.path.join(source, name), "w").write("<svg/>")
|
|
open(os.path.join(target, "pr-lens-old.svg"), "w").write("<svg/>")
|
|
|
|
names = reviewreport.copy_diagrams(source, target)
|
|
report = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None, codex_error="x",
|
|
diagrams=names)
|
|
|
|
self.assertEqual(names, ["pr-lens-containers.svg", "pr-lens-flow.svg"])
|
|
self.assertEqual(sorted(os.listdir(target)), names)
|
|
self.assertIn("### Diagrams", report)
|
|
self.assertIn("", report)
|
|
self.assertLess(report.index("### Diagrams"), report.index("### Codex"))
|
|
|
|
def test_a_run_without_new_diagrams_shows_none_of_an_earlier_run(self):
|
|
"""Scenario#129.02"""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
source, target = os.path.join(tmp, "drawn"), os.path.join(tmp, "draft")
|
|
os.makedirs(source)
|
|
os.makedirs(target)
|
|
for directory in (source, target):
|
|
open(os.path.join(directory, "pr-lens-earlier.svg"), "w").write("<svg/>")
|
|
|
|
reviewreport.clear_diagrams(source)
|
|
names = reviewreport.copy_diagrams(source, target)
|
|
|
|
self.assertEqual((names, os.listdir(source), os.listdir(target)), ([], [], []))
|
|
self.assertNotIn("### Diagrams", reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None,
|
|
diagrams=names))
|
|
|
|
def test_without_diagrams_the_draft_has_no_section(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
self.assertEqual(reviewreport.copy_diagrams(os.path.join(tmp, "none"), tmp), [])
|
|
self.assertNotIn("### Diagrams", reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None))
|
|
|
|
|
|
class AttachTest(unittest.TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.dir = tmp.name
|
|
open(os.path.join(self.dir, "pr-lens-flow.svg"), "w").write("<svg/>")
|
|
report = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None, codex_error="x",
|
|
diagrams=["pr-lens-flow.svg"])
|
|
self.checked = report.replace(report.split("\n")[2] + "\n", "")
|
|
self.gitea = FakeGitea()
|
|
|
|
def test_a_checked_draft_is_posted_with_its_diagrams(self):
|
|
"""Scenario#130.01"""
|
|
url = reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
|
|
|
|
comment = self.gitea.comments[-1]
|
|
self.assertEqual(url, comment["html_url"])
|
|
self.assertIn(f"<!-- hs-review:{HEAD} -->", comment["body"])
|
|
self.assertIn("", comment["body"])
|
|
self.assertNotIn("Attach the SVGs", comment["body"])
|
|
self.assertEqual([a["name"] for a in comment["assets"]], ["pr-lens-flow.svg"])
|
|
|
|
def test_an_unchecked_draft_is_refused(self):
|
|
"""Scenario#130.02"""
|
|
draft = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None, codex_error="x")
|
|
with self.assertRaisesRegex(reviewreport.AttachError, "not checked"):
|
|
reviewreport.attach(draft, self.dir, self.gitea, 7, HEAD)
|
|
self.assertEqual(self.gitea.writes, [])
|
|
|
|
def test_a_draft_of_another_head_is_refused(self):
|
|
"""Scenario#130.02"""
|
|
with self.assertRaisesRegex(reviewreport.AttachError, "run hs-review again"):
|
|
reviewreport.attach(self.checked, self.dir, self.gitea, 7, "f" * 40)
|
|
self.assertEqual(self.gitea.writes, [])
|
|
|
|
def test_a_second_post_for_the_same_head_is_refused(self):
|
|
"""Scenario#130.02"""
|
|
reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
|
|
writes = list(self.gitea.writes)
|
|
|
|
with self.assertRaisesRegex(reviewreport.AttachError, "posted already"):
|
|
reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
|
|
self.assertEqual(self.gitea.writes, writes)
|
|
|
|
def test_a_failed_upload_leaves_no_comment_and_allows_a_retry(self):
|
|
"""Scenario#130.01"""
|
|
create = self.gitea.create_comment_asset
|
|
|
|
def refused(*args):
|
|
raise OSError("HTTP 413")
|
|
self.gitea.create_comment_asset = refused
|
|
|
|
with self.assertRaises(OSError):
|
|
reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
|
|
self.assertEqual(self.gitea.comments, [])
|
|
|
|
self.gitea.create_comment_asset = create
|
|
reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
|
|
self.assertEqual(len(self.gitea.comments), 1)
|
|
|
|
def test_a_finding_that_quotes_the_draft_line_is_no_refusal(self):
|
|
"""Scenario#130.02"""
|
|
quoted = self.checked.replace("### Codex", "> **Draft:** quoted in a finding\n\n### Codex")
|
|
|
|
reviewreport.attach(quoted, self.dir, self.gitea, 7, HEAD)
|
|
|
|
self.assertIn("quoted in a finding", self.gitea.comments[-1]["body"])
|
|
|
|
def test_a_reason_replaces_the_draft_line_and_is_posted(self):
|
|
"""Scenario#136.01"""
|
|
draft = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None, codex_error="x",
|
|
diagrams=["pr-lens-flow.svg"])
|
|
reason = "Finding 1 is fixed in a1b2c3d.\n\nFinding 2 does not hold: the check runs in the repository."
|
|
|
|
reviewreport.attach(draft, self.dir, self.gitea, 7, HEAD, checked=reason)
|
|
|
|
body = self.gitea.comments[-1]["body"]
|
|
self.assertNotIn("> **Draft:**", body)
|
|
self.assertIn("> **Checked before posting:**\n> Finding 1 is fixed in a1b2c3d.\n>\n"
|
|
"> Finding 2 does not hold: the check runs in the repository.\n", body)
|
|
self.assertLess(body.index("Checked before posting"), body.index("### werkrevisor"))
|
|
|
|
def test_a_reason_that_quotes_the_draft_line_stays_whole(self):
|
|
"""Scenario#136.01"""
|
|
draft = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None, codex_error="x")
|
|
|
|
reviewreport.attach(draft, self.dir, self.gitea, 7, HEAD,
|
|
checked="**Draft:** the remaining finding is already covered.\nSecond line.")
|
|
|
|
body = self.gitea.comments[-1]["body"]
|
|
self.assertIn("> **Checked before posting:**\n> **Draft:** the remaining finding is already covered.\n"
|
|
"> Second line.\n", body)
|
|
|
|
def test_an_empty_reason_is_refused_before_any_write(self):
|
|
"""Scenario#136.02"""
|
|
draft = reviewreport.compose(HEAD, reviewreport.werkrevisor_part(DRY_RUN), None, codex_error="x")
|
|
for reason in ("", " \n\t\n"):
|
|
with self.subTest(reason=reason), self.assertRaisesRegex(reviewreport.AttachError, "needs a reason"):
|
|
reviewreport.attach(draft, self.dir, self.gitea, 7, HEAD, checked=reason)
|
|
self.assertEqual(self.gitea.writes, [])
|
|
|
|
def test_a_reason_also_follows_a_draft_line_deleted_by_hand(self):
|
|
reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD, checked="All points hold; fixed.")
|
|
|
|
body = self.gitea.comments[-1]["body"]
|
|
self.assertIn("> **Checked before posting:**\n> All points hold; fixed.", body)
|
|
self.assertLess(body.index("Checked before posting"), body.index("### werkrevisor"))
|
|
|
|
def test_a_missing_svg_is_refused_before_any_write(self):
|
|
os.remove(os.path.join(self.dir, "pr-lens-flow.svg"))
|
|
with self.assertRaisesRegex(reviewreport.AttachError, "pr-lens-flow.svg"):
|
|
reviewreport.attach(self.checked, self.dir, self.gitea, 7, HEAD)
|
|
self.assertEqual(self.gitea.writes, [])
|
|
|
|
|
|
class ErrorTextTest(unittest.TestCase):
|
|
def test_the_cause_of_a_failure_comes_first(self):
|
|
"""Scenario#131.01"""
|
|
stderr = ("werkrevisor: operator config from /werkrevisor/config/hsadmin-ng.toml\n"
|
|
"werkrevisor: cannot read the PR: GET /repos/o/r/pulls/343 failed: The handshake operation timed out\n")
|
|
|
|
text = reviewreport.error_text(stderr)
|
|
report = reviewreport.compose(HEAD, None, None, werkrevisor_error=text, codex_error="x")
|
|
|
|
self.assertTrue(text.startswith("werkrevisor: cannot read the PR"))
|
|
self.assertIn("The review did not complete: werkrevisor: cannot read the PR", report)
|
|
self.assertNotIn("operator config", report)
|
|
|
|
def test_a_gitea_request_waits_90_seconds(self):
|
|
"""Scenario#131.02"""
|
|
from werkrevisor.gitea import GiteaClient
|
|
self.assertEqual(GiteaClient("https://g.example.org", "o", "r", "t")._timeout, 90)
|
|
|
|
|
|
class TitleTest(unittest.TestCase):
|
|
URL = "https://git.example.org/team/app/pulls/343"
|
|
|
|
def test_the_title_links_the_pr_and_keeps_the_commit(self):
|
|
"""Scenario#134.02"""
|
|
report = reviewreport.compose(HEAD, None, None, codex_error="x", pr=(343, self.URL))
|
|
|
|
self.assertTrue(report.startswith(f"## AI review of [PR#343]({self.URL}) (`6d1842aa`)\n"))
|
|
|
|
def test_attach_reads_the_commit_from_the_new_and_the_old_title(self):
|
|
"""Scenario#134.02"""
|
|
for pr in ((343, self.URL), None):
|
|
with self.subTest(pr=pr):
|
|
draft = reviewreport.compose(HEAD, None, None, codex_error="x", pr=pr)
|
|
checked = "\n".join(line for line in draft.split("\n") if not line.startswith("> **Draft:**"))
|
|
gitea = FakeGitea()
|
|
|
|
reviewreport.attach(checked, tempfile.gettempdir(), gitea, 7, HEAD)
|
|
|
|
self.assertEqual(len(gitea.comments), 1)
|
|
with self.assertRaisesRegex(reviewreport.AttachError, "run hs-review again"):
|
|
reviewreport.attach(checked, tempfile.gettempdir(), FakeGitea(), 7, "f" * 40)
|
|
|