338 lines
14 KiB
Python
338 lines
14 KiB
Python
import contextlib
|
|
import dataclasses
|
|
import io
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
from tests.fakes import FakeGitea, FakeModel, config, deps, diff_for, gitea_file, pr_ref, review_data
|
|
from werkrevisor import prlens, report
|
|
from werkrevisor.cli import execute
|
|
from werkrevisor.config import ConfigError
|
|
from werkrevisor.redact import Redactor
|
|
from werkrevisor.verdict import Verdict
|
|
|
|
DOC = "doc/PR/1.md"
|
|
SVG = b'<svg xmlns="http://www.w3.org/2000/svg"/>'
|
|
|
|
|
|
def docs_gitea():
|
|
return FakeGitea(files=[gitea_file(DOC)], diff=diff_for(DOC, added=["Some text."]))
|
|
|
|
|
|
class FakeDraw:
|
|
def __init__(self, count=2, error=None):
|
|
self.calls = []
|
|
self.count = count
|
|
self.error = error
|
|
|
|
def __call__(self, cfg, ref, patch):
|
|
self.calls.append(patch)
|
|
if self.error:
|
|
raise self.error
|
|
return [(f"pr-lens-view-{i}.svg", SVG) for i in range(self.count)]
|
|
|
|
|
|
def run(gitea, draw, redactor=None, model=None, use_llm=True, **diagrams):
|
|
# Most tests review a GREEN docs PR; the tests of skip_green set it themselves.
|
|
diagrams.setdefault("skip_green", False)
|
|
d = deps(gitea, model or FakeModel(prompt_files=[f["filename"] for f in gitea.files]))
|
|
d = dataclasses.replace(d, draw_diagrams=draw)
|
|
cfg = config(gate={"yellow_gate": "block"}, diagrams=diagrams)
|
|
err = io.StringIO()
|
|
with contextlib.redirect_stderr(err):
|
|
outcome = execute(cfg, pr_ref(), d, redactor or Redactor(), publish=True, use_llm=use_llm)
|
|
return outcome, err.getvalue()
|
|
|
|
|
|
def result_body(gitea):
|
|
return gitea.comments[-1]["body"]
|
|
|
|
|
|
class DiagramsInTheResultCommentTest(unittest.TestCase):
|
|
def test_an_enabled_repository_gets_the_diagrams_in_its_result_comment(self):
|
|
"""Scenario#121.01"""
|
|
gitea, draw = docs_gitea(), FakeDraw()
|
|
|
|
outcome, _ = run(gitea, draw, enabled=True)
|
|
|
|
body = result_body(gitea)
|
|
self.assertEqual((outcome.result.verdict, outcome.result.exit_code), (Verdict.GREEN, 0))
|
|
self.assertEqual([a["name"] for a in gitea.comments[-1]["assets"]],
|
|
["pr-lens-view-0.svg", "pr-lens-view-1.svg"])
|
|
self.assertIn(prlens.MARKER, body)
|
|
self.assertIn("not a statement of what it does", body)
|
|
self.assertIn("](https://git.example.org/attachments/2-pr-lens-view-1.svg)", body)
|
|
self.assertIn(DOC, draw.calls[0])
|
|
|
|
def test_diagrams_are_off_by_default(self):
|
|
"""Scenario#121.02"""
|
|
gitea, draw = docs_gitea(), FakeDraw()
|
|
|
|
run(gitea, draw)
|
|
|
|
self.assertEqual(draw.calls, [])
|
|
self.assertNotIn("asset", gitea.writes)
|
|
self.assertNotIn(prlens.MARKER, result_body(gitea))
|
|
|
|
def test_a_failure_of_pr_lens_leaves_the_review_unchanged(self):
|
|
"""Scenario#121.03"""
|
|
gitea = docs_gitea()
|
|
|
|
outcome, err = run(gitea, FakeDraw(error=prlens.DiagramError("PR Lens failed: s3cret")),
|
|
redactor=Redactor(["s3cret"]), enabled=True)
|
|
|
|
self.assertEqual((outcome.result.verdict, outcome.result.exit_code), (Verdict.GREEN, 0))
|
|
self.assertNotIn(prlens.MARKER, result_body(gitea))
|
|
self.assertIn("diagrams not attached", err)
|
|
self.assertNotIn("s3cret", err)
|
|
|
|
def test_a_rerun_on_the_same_head_reuses_the_attached_diagrams(self):
|
|
"""Scenario#121.04"""
|
|
gitea = docs_gitea()
|
|
run(gitea, FakeDraw(), enabled=True)
|
|
comment = gitea.comments[-1]
|
|
comment["body"] = comment["body"].split("\n\n" + prlens.MARKER)[0]
|
|
again = FakeDraw()
|
|
|
|
run(gitea, again, enabled=True)
|
|
|
|
self.assertEqual(again.calls, [])
|
|
self.assertEqual(len(gitea.comments[-1]["assets"]), 2)
|
|
self.assertEqual(result_body(gitea).count(prlens.MARKER), 1)
|
|
|
|
def test_a_possible_secret_keeps_the_diff_from_pr_lens(self):
|
|
"""Scenario#121.07"""
|
|
gitea = FakeGitea(files=[gitea_file(DOC)], diff=diff_for(DOC, added=["-----BEGIN " + "PRIVATE KEY-----"]))
|
|
draw = FakeDraw()
|
|
|
|
outcome, _ = run(gitea, draw, enabled=True)
|
|
|
|
self.assertIs(outcome.result.verdict, Verdict.YELLOW)
|
|
self.assertEqual(draw.calls, [])
|
|
self.assertNotIn("asset", gitea.writes)
|
|
|
|
def test_pr_lens_gets_the_diff_that_passed_the_secret_scan(self):
|
|
"""Scenario#121.07"""
|
|
gitea, draw = docs_gitea(), FakeDraw()
|
|
reviewed = gitea.diff
|
|
set_status = gitea.set_status
|
|
|
|
def push_meanwhile(*args, **kw):
|
|
gitea.diff = diff_for(DOC, added=["-----BEGIN " + "PRIVATE KEY-----"])
|
|
return set_status(*args, **kw)
|
|
gitea.set_status = push_meanwhile
|
|
|
|
run(gitea, draw, enabled=True)
|
|
|
|
self.assertEqual(draw.calls, [reviewed])
|
|
|
|
def test_no_llm_draws_no_diagrams(self):
|
|
"""Scenario#121.08"""
|
|
gitea, draw = docs_gitea(), FakeDraw()
|
|
|
|
run(gitea, draw, use_llm=False, enabled=True)
|
|
|
|
self.assertEqual(draw.calls, [])
|
|
|
|
def test_a_rerun_without_attached_diagrams_draws_none(self):
|
|
"""Scenario#121.07"""
|
|
gitea = FakeGitea(files=[gitea_file(DOC)], diff=diff_for(DOC, added=["-----BEGIN " + "PRIVATE KEY-----"]))
|
|
run(gitea, FakeDraw(), enabled=True)
|
|
again = FakeDraw()
|
|
|
|
run(gitea, again, enabled=True)
|
|
|
|
self.assertEqual(again.calls, [])
|
|
|
|
def test_diagrams_beyond_the_attachment_limit_of_gitea_are_left_out(self):
|
|
"""Scenario#121.06"""
|
|
gitea, draw = docs_gitea(), FakeDraw(count=3)
|
|
create = gitea.create_comment_asset
|
|
|
|
def limited(comment_id, name, content, content_type):
|
|
if name.endswith("-2.svg"):
|
|
raise OSError("HTTP 422")
|
|
return create(comment_id, name, content, content_type)
|
|
gitea.create_comment_asset = limited
|
|
|
|
_, err = run(gitea, draw, enabled=True)
|
|
|
|
self.assertEqual(result_body(gitea).count("![pr-lens-"), 2)
|
|
self.assertIn("1 of 3 diagrams not attached", err)
|
|
self.assertIn("[attachment] MAX_FILES", err)
|
|
|
|
def test_a_dry_run_writes_the_diagrams_to_a_directory(self):
|
|
"""Scenario#129.01"""
|
|
gitea, draw = docs_gitea(), FakeDraw()
|
|
model = FakeModel(data=review_data(issues=[DOC], header="[high] Possible Bug"), prompt_files=[DOC])
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
open(os.path.join(tmp, "pr-lens-old.svg"), "w").write("<svg/>")
|
|
d = dataclasses.replace(deps(gitea, model), draw_diagrams=draw, diagrams_out=tmp)
|
|
with contextlib.redirect_stderr(io.StringIO()):
|
|
execute(config(gate={"yellow_gate": "block"}, diagrams={"enabled": True}), pr_ref(), d, Redactor(),
|
|
publish=False, use_llm=True)
|
|
|
|
self.assertEqual(sorted(os.listdir(tmp)), ["pr-lens-view-0.svg", "pr-lens-view-1.svg"])
|
|
self.assertEqual(len(draw.calls), 1)
|
|
self.assertEqual(gitea.writes, [])
|
|
|
|
def test_a_failed_dry_run_leaves_no_diagrams_of_an_earlier_run(self):
|
|
"""Scenario#129.01"""
|
|
gitea, draw = docs_gitea(), FakeDraw()
|
|
model = FakeModel(error=RuntimeError("model down"), prompt_files=[DOC])
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
open(os.path.join(tmp, "pr-lens-old.svg"), "w").write("<svg/>")
|
|
d = dataclasses.replace(deps(gitea, model), draw_diagrams=draw, diagrams_out=tmp)
|
|
with contextlib.redirect_stderr(io.StringIO()):
|
|
outcome = execute(config(gate={"yellow_gate": "block"}, diagrams={"enabled": True}), pr_ref(), d,
|
|
Redactor(), publish=False, use_llm=True)
|
|
|
|
self.assertIs(outcome.result.verdict, Verdict.ERROR)
|
|
self.assertEqual(os.listdir(tmp), [])
|
|
|
|
def test_a_green_dry_run_writes_no_diagrams(self):
|
|
"""Scenario#129.01"""
|
|
gitea, draw = docs_gitea(), FakeDraw()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
open(os.path.join(tmp, "pr-lens-old.svg"), "w").write("<svg/>")
|
|
d = dataclasses.replace(deps(gitea, FakeModel(prompt_files=[DOC])), draw_diagrams=draw, diagrams_out=tmp)
|
|
execute(config(gate={"yellow_gate": "block"}, diagrams={"enabled": True}), pr_ref(), d, Redactor(),
|
|
publish=False, use_llm=True)
|
|
|
|
self.assertEqual(os.listdir(tmp), [])
|
|
self.assertEqual(draw.calls, [])
|
|
|
|
def test_a_fold_during_the_drawing_stays(self):
|
|
"""Scenario#127.01"""
|
|
gitea = docs_gitea()
|
|
|
|
class FoldMeanwhile(FakeDraw):
|
|
def __call__(self, cfg, ref, patch):
|
|
comment = gitea.comments[-1]
|
|
comment["body"] = report.fold(comment["body"], "https://git.example.org/newer", "f" * 40)
|
|
return super().__call__(cfg, ref, patch)
|
|
|
|
run(gitea, FoldMeanwhile(), enabled=True)
|
|
|
|
body = result_body(gitea)
|
|
self.assertIn(report.FOLDED_MARKER, body)
|
|
self.assertIn("https://git.example.org/newer", body)
|
|
self.assertEqual(body.count("![pr-lens-"), 2)
|
|
self.assertLess(body.index(prlens.MARKER), body.rindex("</details>"))
|
|
|
|
def test_a_green_pr_gets_no_diagrams(self):
|
|
"""Scenario#125.01"""
|
|
gitea, draw = docs_gitea(), FakeDraw()
|
|
|
|
outcome, _ = run(gitea, draw, enabled=True, skip_green=True)
|
|
|
|
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
|
self.assertEqual(draw.calls, [])
|
|
self.assertNotIn("asset", gitea.writes)
|
|
|
|
def test_a_pr_that_a_human_reviews_gets_diagrams(self):
|
|
"""Scenario#125.02"""
|
|
gitea, draw = docs_gitea(), FakeDraw()
|
|
model = FakeModel(data=review_data(issues=[DOC], header="[high] Possible Bug"), prompt_files=[DOC])
|
|
|
|
outcome, _ = run(gitea, draw, model=model, enabled=True, skip_green=True)
|
|
|
|
self.assertIs(outcome.result.verdict, Verdict.ORANGE)
|
|
self.assertEqual(len(draw.calls), 1)
|
|
self.assertIn(prlens.MARKER, result_body(gitea))
|
|
|
|
def test_an_error_result_gets_no_diagrams(self):
|
|
gitea, draw = docs_gitea(), FakeDraw()
|
|
|
|
outcome, _ = run(gitea, draw, model=FakeModel(error=RuntimeError("model down"), prompt_files=[DOC]),
|
|
enabled=True)
|
|
|
|
self.assertIs(outcome.result.verdict, Verdict.ERROR)
|
|
self.assertEqual(draw.calls, [])
|
|
|
|
|
|
def completed(stdout="", returncode=0, stderr=""):
|
|
return subprocess.CompletedProcess([], returncode, stdout=stdout, stderr=stderr)
|
|
|
|
|
|
class DrawTest(unittest.TestCase):
|
|
PATCH = "diff --git a/a.py b/a.py\n--- a/a.py\n+++ b/a.py\n@@ -1 +1,2 @@\n-x\n+y\n+z\n"
|
|
|
|
def draw(self, answer, **diagrams):
|
|
calls = []
|
|
|
|
def fake_run(argv, **kw):
|
|
calls.append((argv, kw))
|
|
if isinstance(answer, Exception):
|
|
raise answer
|
|
return answer
|
|
|
|
settings = config(diagrams={"enabled": True, **diagrams})
|
|
environ = {"OPENROUTER_API_KEY": "or-key", "GITEA_BOT_TOKEN": "gitea-token", "HOME": "/root"}
|
|
return prlens.draw(settings.diagrams, settings.model.id, pr_ref(), self.PATCH, environ, fake_run), calls
|
|
|
|
def test_pr_lens_gets_the_model_key_and_no_other_secret(self):
|
|
"""Scenario#121.05"""
|
|
_, calls = self.draw(completed(json.dumps({"svgs": []})))
|
|
|
|
(argv, kw), = calls
|
|
self.assertEqual(kw["env"], {"PATH": "/usr/bin:/bin", "HOME": "/tmp", "OPENROUTER_API_KEY": "or-key"})
|
|
self.assertNotIn("gitea-token", json.dumps([argv, kw]))
|
|
request = json.loads(kw["input"])
|
|
self.assertEqual((request["model"], request["baseUrl"]),
|
|
("anthropic/claude-sonnet-5.5", "https://openrouter.ai/api/v1"))
|
|
self.assertEqual(request["files"], [{"path": "a.py", "additions": 2, "deletions": 1}])
|
|
|
|
def test_at_most_max_diagrams_are_attached(self):
|
|
"""Scenario#121.06"""
|
|
svgs = [{"view": f"Flow {i}", "svg": "<svg/>"} for i in range(7)]
|
|
|
|
drawn, _ = self.draw(completed(json.dumps({"svgs": svgs})), max_diagrams=5)
|
|
|
|
self.assertEqual([name for name, _ in drawn], [f"pr-lens-flow-{i}.svg" for i in range(5)])
|
|
|
|
def test_a_failed_or_slow_pr_lens_raises_a_diagram_error(self):
|
|
with self.assertRaisesRegex(prlens.DiagramError, "MODEL_OUTPUT_INVALID"):
|
|
self.draw(completed(returncode=1, stderr="trace\nMODEL_OUTPUT_INVALID: no JSON"))
|
|
with self.assertRaisesRegex(prlens.DiagramError, r"PR Lens failed: pr-lens: MODEL_OUTPUT_INVALID: twice$"):
|
|
self.draw(completed(returncode=1, stderr="pr-lens: MODEL_OUTPUT_INVALID: twice\n\nNode.js v20.19.2\n"))
|
|
with self.assertRaisesRegex(prlens.DiagramError, "longer than 300 s"):
|
|
self.draw(subprocess.TimeoutExpired("node", 300))
|
|
with self.assertRaisesRegex(prlens.DiagramError, "without diagrams"):
|
|
self.draw(completed("not json"))
|
|
|
|
|
|
class DiagramsConfigTest(unittest.TestCase):
|
|
def test_hs_review_draws_also_for_green(self):
|
|
"""Scenario#129.03"""
|
|
from tests.fakes import example_config
|
|
hs_review = example_config("hsadmin-ng.toml").diagrams
|
|
gitea, draw = docs_gitea(), FakeDraw()
|
|
cfg = dataclasses.replace(config(gate={"yellow_gate": "block"}), diagrams=hs_review)
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
d = dataclasses.replace(deps(gitea, FakeModel(prompt_files=[DOC])), draw_diagrams=draw, diagrams_out=tmp)
|
|
outcome = execute(cfg, pr_ref(), d, Redactor(), publish=False, use_llm=True)
|
|
|
|
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
|
self.assertEqual(sorted(os.listdir(tmp)), ["pr-lens-view-0.svg", "pr-lens-view-1.svg"])
|
|
|
|
def test_the_review_model_is_the_default(self):
|
|
self.assertEqual(config(diagrams={"enabled": True}).diagrams.model_for("openrouter/x/y"), "x/y")
|
|
self.assertEqual(config(diagrams={"model": "a/b"}).diagrams.model_for("openrouter/x/y"), "a/b")
|
|
|
|
def test_an_enabled_section_needs_an_openrouter_model(self):
|
|
with self.assertRaisesRegex(ConfigError, r"\[diagrams\] model"):
|
|
config(model={"id": "anthropic/claude"}, diagrams={"enabled": True})
|
|
with self.assertRaisesRegex(ConfigError, r"\[diagrams\] max_diagrams"):
|
|
config(diagrams={"max_diagrams": 0})
|
|
with self.assertRaisesRegex(ConfigError, r"\[diagrams\] skip_green"):
|
|
config(diagrams={"skip_green": "yes"})
|
|
self.assertTrue(config(diagrams={"enabled": True}).diagrams.skip_green)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|