229 lines
11 KiB
Python
229 lines
11 KiB
Python
import unittest
|
|
|
|
from tests.fakes import BOT, FakeGitea, FakeJev, FakeModel, config, gitea_file, review_data
|
|
from tests.test_flow import run
|
|
from werkrevisor import carryover, report
|
|
from werkrevisor.diff import parse_unified_diff
|
|
from werkrevisor.verdict import Verdict
|
|
|
|
APP = "src/main/java/App.java"
|
|
APP_TEST = "src/test/java/AppTest.java"
|
|
OLD = "c" * 40
|
|
APPROVAL_ID = 5
|
|
CHECK = ["if (user == null) {", " throw new IllegalArgumentException();", "}"]
|
|
TEST = ["@Test void rejectsNull() {}"]
|
|
|
|
|
|
def hunk_diff(path, hunks):
|
|
"""hunks: (first new line, removed lines, added lines); the line numbers shift as they do after a rebase."""
|
|
lines = [f"diff --git a/{path} b/{path}", f"--- a/{path}", f"+++ b/{path}"]
|
|
for first, removed, added in hunks:
|
|
lines.append(f"@@ -{first},{len(removed) + 1} +{first},{len(added) + 1} @@")
|
|
lines += [" context"] + ["-" + r for r in removed] + ["+" + a for a in added]
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
APPROVED_DIFF = hunk_diff(APP, [(2, [], CHECK)]) + hunk_diff(APP_TEST, [(5, [], TEST)])
|
|
REBASED_DIFF = hunk_diff(APP, [(12, [], CHECK)]) + hunk_diff(APP_TEST, [(9, [], TEST)])
|
|
CORRECTED_DIFF = hunk_diff(APP, [(12, [], CHECK), (40, ["old call();"], ["new call(user);"])]) \
|
|
+ hunk_diff(APP_TEST, [(9, [], TEST)])
|
|
|
|
|
|
def result_comment(head, verdict="YELLOW", diff=APPROVED_DIFF):
|
|
marker = carryover.marker(carryover.blocks(parse_unified_diff(diff)))
|
|
return {"id": 1, "user": {"login": BOT},
|
|
"body": f"<!-- werkrevisor:{head} -->\n<!-- werkrevisor-verdict:{verdict} -->\n{marker}\n## werkrevisor"}
|
|
|
|
|
|
def approval(dismissal=carryover.AUTO_DISMISSAL):
|
|
review = {"id": APPROVAL_ID, "state": "APPROVED", "commit_id": OLD, "user": {"login": "alice"}, "dismissed": True}
|
|
event = {"type": "dismiss_review", "review_id": APPROVAL_ID, "body": dismissal}
|
|
return review, event
|
|
|
|
|
|
def carry_gitea(diff, dismissal=carryover.AUTO_DISMISSAL, human=True, verdict="YELLOW"):
|
|
review, event = approval(dismissal)
|
|
gitea = FakeGitea(files=[gitea_file(APP), gitea_file(APP_TEST)], diff=diff,
|
|
comments=[result_comment(OLD, verdict)], reviews=[review] if human else [])
|
|
gitea.timeline.append(event)
|
|
return gitea
|
|
|
|
|
|
def carry_config():
|
|
return config(engine={"mode": "combined", "model_role": "advises"},
|
|
gate={"yellow_gate": "approval", "human_reviewers": ["alice"]},
|
|
path_judge={"judge": "jev", "scope": "all"})
|
|
|
|
|
|
def carry_run(gitea, jev=None, model=None):
|
|
model = model or FakeModel(prompt_files=[APP, APP_TEST])
|
|
outcome, _ = run(gitea, model, cfg=carry_config(), jev=jev or FakeJev())
|
|
return outcome, model
|
|
|
|
|
|
def codes(outcome):
|
|
return {r.code: r for r in outcome.result.reasons}
|
|
|
|
|
|
def finding_on_the_check(prompt_files=(APP,)):
|
|
data = review_data(issues=[APP], header="[high] Possible Bug")
|
|
for issue in data["review"]["key_issues_to_review"]:
|
|
issue.update(start_line=13, end_line=15)
|
|
return FakeModel(data=data, prompt_files=list(prompt_files))
|
|
|
|
|
|
class CarryOverTest(unittest.TestCase):
|
|
def test_a_rebase_keeps_the_approval(self):
|
|
"""Scenario#73.01"""
|
|
gitea = carry_gitea(REBASED_DIFF)
|
|
jev = FakeJev()
|
|
outcome, model = carry_run(gitea, jev)
|
|
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
|
self.assertEqual(model.calls, [])
|
|
self.assertEqual(jev.requests, [])
|
|
self.assertEqual(codes(outcome)["approval_carried_over"].text,
|
|
f"approval of alice for `{OLD[:7]}` carried over: "
|
|
"the changed blocks are those of the approved head")
|
|
self.assertNotIn("category", codes(outcome))
|
|
self.assertEqual([r[0] for r in gitea.created_reviews], ["APPROVED"])
|
|
|
|
def test_a_harmless_correction_keeps_the_approval(self):
|
|
"""Scenario#73.02"""
|
|
jev = FakeJev()
|
|
outcome, model = carry_run(carry_gitea(CORRECTED_DIFF), jev)
|
|
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
|
self.assertEqual(model.calls, [])
|
|
judged = {state.split("\n", 1)[0] for state, _ in jev.requests}
|
|
self.assertEqual(judged, {f"Changed file: {APP}"})
|
|
self.assertIn(f"Jev finds no risk in the new blocks of {APP}", codes(outcome)["approval_carried_over"].text)
|
|
|
|
def test_a_risky_correction_gets_a_review(self):
|
|
"""Scenario#73.03"""
|
|
outcome, model = carry_run(carry_gitea(CORRECTED_DIFF), FakeJev(problems={(APP, "security"): 0.97}))
|
|
self.assertIs(outcome.result.verdict, Verdict.YELLOW)
|
|
self.assertEqual(len(model.calls), 1)
|
|
self.assertIn(APP_TEST, model.excluded[0])
|
|
self.assertNotIn(APP, model.excluded[0])
|
|
self.assertIn("approved on an earlier head", model.calls[0])
|
|
self.assertNotIn("approval_carried_over", codes(outcome))
|
|
|
|
def test_a_finding_on_approved_lines_is_a_hint(self):
|
|
"""Scenario#73.03"""
|
|
unit = f"{APP}:13-15"
|
|
jev = FakeJev(problems={(APP, "security"): 0.97, (unit, "real_defect"): 0.95, (unit, "severity"): 0.95})
|
|
outcome, _ = carry_run(carry_gitea(CORRECTED_DIFF), jev, finding_on_the_check())
|
|
self.assertEqual(outcome.result.finding_verdicts, [Verdict.GREEN])
|
|
self.assertNotIn("model_findings", codes(outcome))
|
|
self.assertEqual(codes(outcome)["approved_lines"].text,
|
|
f"1 finding(s) on lines approved by alice for `{OLD[:7]}`: {APP}:13-15")
|
|
|
|
def test_an_approval_dismissed_by_hand_is_not_carried_over(self):
|
|
"""Scenario#73.04"""
|
|
outcome, model = carry_run(carry_gitea(REBASED_DIFF, dismissal="Please split the PR."))
|
|
self.assertEqual(len(model.calls), 1)
|
|
self.assertNotIn("approval_carried_over", codes(outcome))
|
|
|
|
def test_a_revoked_approval_is_not_carried_over(self):
|
|
gitea = carry_gitea(REBASED_DIFF)
|
|
gitea.reviews.append({"id": APPROVAL_ID + 1, "state": "REQUEST_CHANGES", "commit_id": OLD,
|
|
"user": {"login": "alice"}})
|
|
outcome, model = carry_run(gitea)
|
|
self.assertEqual(len(model.calls), 1)
|
|
self.assertNotIn("approval_carried_over", codes(outcome))
|
|
|
|
def test_fingerprints_in_a_comment_of_another_account_count_for_nothing(self):
|
|
gitea = carry_gitea(REBASED_DIFF)
|
|
gitea.comments[0]["user"] = {"login": "dev"}
|
|
outcome, model = carry_run(gitea)
|
|
self.assertEqual(len(model.calls), 1)
|
|
self.assertIn("left no fingerprints", codes(outcome)["approval_not_carried_over"].text)
|
|
|
|
def test_carry_over_approvals_false_carries_nothing_over(self):
|
|
cfg = config(engine={"mode": "combined", "model_role": "advises"},
|
|
gate={"yellow_gate": "approval", "human_reviewers": ["alice"], "carry_over_approvals": False},
|
|
path_judge={"judge": "jev", "scope": "all"})
|
|
model = FakeModel(prompt_files=[APP, APP_TEST])
|
|
outcome, _ = run(carry_gitea(REBASED_DIFF), model, cfg=cfg, jev=FakeJev())
|
|
self.assertEqual(len(model.calls), 1)
|
|
self.assertNotIn("approval_carried_over", codes(outcome))
|
|
|
|
def test_an_approval_before_the_result_of_its_head_is_not_carried_over(self):
|
|
gitea = carry_gitea(REBASED_DIFF)
|
|
gitea.reviews[0]["submitted_at"] = "2026-10-10T13:51:44+02:00"
|
|
gitea.comments[0]["created_at"] = "2026-10-10T13:56:11+02:00"
|
|
outcome, model = carry_run(gitea)
|
|
self.assertEqual(len(model.calls), 1)
|
|
self.assertIn("it came before the result of that head", codes(outcome)["approval_not_carried_over"].text)
|
|
|
|
def test_a_dropped_block_ends_the_carry_over(self):
|
|
"""Scenario#73.05"""
|
|
outcome, model = carry_run(carry_gitea(hunk_diff(APP, [(12, [], CHECK)]) + hunk_diff(APP_TEST, [(9, [], [])])))
|
|
self.assertEqual(len(model.calls), 1)
|
|
self.assertEqual(codes(outcome)["approval_not_carried_over"].text,
|
|
f"approval of alice for `{OLD[:7]}` not carried over: 1 approved block(s) are missing")
|
|
|
|
def test_blocks_of_a_green_review_do_not_block_again(self):
|
|
"""Scenario#73.06"""
|
|
unit = f"{APP}:13-15"
|
|
jev = FakeJev(problems={(unit, "real_defect"): 0.95, (unit, "severity"): 0.95})
|
|
outcome, model = carry_run(carry_gitea(REBASED_DIFF, human=False, verdict="GREEN"), jev,
|
|
finding_on_the_check((APP, APP_TEST)))
|
|
self.assertEqual(len(model.calls), 1)
|
|
self.assertEqual(model.excluded[0], ())
|
|
self.assertEqual(outcome.result.finding_verdicts, [Verdict.GREEN])
|
|
self.assertIn("approved by werkrevisor", codes(outcome)["approved_lines"].text)
|
|
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
|
|
|
def test_a_yellow_review_of_werkrevisor_approves_nothing(self):
|
|
unit = f"{APP}:13-15"
|
|
jev = FakeJev(problems={(unit, "real_defect"): 0.95, (unit, "severity"): 0.95})
|
|
outcome, _ = carry_run(carry_gitea(REBASED_DIFF, human=False), jev, finding_on_the_check((APP, APP_TEST)))
|
|
self.assertEqual(outcome.result.finding_verdicts, [Verdict.ORANGE])
|
|
|
|
def test_the_result_comment_holds_the_fingerprints(self):
|
|
gitea = FakeGitea(files=[gitea_file(APP)], diff=hunk_diff(APP, [(2, [], CHECK)]))
|
|
run(gitea, FakeModel(prompt_files=[APP]))
|
|
markers = [carryover.parse_marker(c) for c in gitea.created_comments]
|
|
self.assertEqual([sum(m.values()) for m in markers], [1])
|
|
folded = report.fold(gitea.created_comments[0], "https://example.org", "e" * 40)
|
|
self.assertEqual(carryover.parse_marker(folded), markers[0])
|
|
|
|
|
|
class ChangeKindTest(unittest.TestCase):
|
|
def test_the_kind_comes_from_the_files_outside_the_approval(self):
|
|
from werkrevisor.rules import evaluate_rules, files_from_gitea
|
|
files = files_from_gitea([gitea_file(APP), gitea_file(APP_TEST)])
|
|
diffs = parse_unified_diff(REBASED_DIFF)
|
|
self.assertEqual(evaluate_rules(files, diffs, carry_config(), {APP}).change_kind, "tests_only_added")
|
|
self.assertEqual(evaluate_rules(files, diffs, carry_config(), {APP, APP_TEST}).change_kind,
|
|
evaluate_rules(files, diffs, carry_config()).change_kind)
|
|
|
|
|
|
class BlocksTest(unittest.TestCase):
|
|
def test_a_shift_keeps_the_fingerprints(self):
|
|
before = carryover.blocks(parse_unified_diff(APPROVED_DIFF))
|
|
after = carryover.blocks(parse_unified_diff(REBASED_DIFF))
|
|
self.assertEqual([b.fingerprint for b in before], [b.fingerprint for b in after])
|
|
self.assertEqual(after[0].lines, (13, 14, 15))
|
|
|
|
def test_a_block_of_removed_lines_covers_the_line_where_they_were(self):
|
|
[block] = carryover.blocks(parse_unified_diff(hunk_diff(APP, [(7, ["gone();", "gone too();"], [])])))
|
|
self.assertEqual(block.lines, (8,))
|
|
|
|
def test_the_same_lines_in_another_file_are_another_block(self):
|
|
[a] = carryover.blocks(parse_unified_diff(hunk_diff(APP, [(1, [], TEST)])))
|
|
[b] = carryover.blocks(parse_unified_diff(hunk_diff(APP_TEST, [(1, [], TEST)])))
|
|
self.assertNotEqual(a.fingerprint, b.fingerprint)
|
|
|
|
def test_a_fingerprint_has_128_bits(self):
|
|
[block] = carryover.blocks(parse_unified_diff(hunk_diff(APP, [(1, [], TEST)])))
|
|
self.assertEqual(len(block.fingerprint) * 4, 128)
|
|
|
|
def test_a_large_pr_gets_no_marker(self):
|
|
found = [carryover.Block(APP, "0" * carryover.FINGERPRINT_CHARS, (n,)) for n in range(carryover.MAX_BLOCKS + 1)]
|
|
self.assertEqual(carryover.marker(found), "")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|