130 lines
6.2 KiB
Python
130 lines
6.2 KiB
Python
import unittest
|
|
|
|
from tests.fakes import HEAD, FakeGitea, FakeModel, config, diff_for, gitea_file, review_data
|
|
from tests.test_flow import DOC, docs_gitea, run
|
|
from werkrevisor.verdict import Verdict, early_approvers, human_approval
|
|
|
|
RESULT_AT = "2026-10-10T13:56:11+02:00"
|
|
BEFORE = "2026-10-10T13:51:44+02:00"
|
|
AFTER = "2026-10-10T14:05:00+02:00"
|
|
CFG = {"gate": {"yellow_gate": "approval", "human_reviewers": ["alice"], "require_wip_transition": True,
|
|
"automerge": True}}
|
|
|
|
|
|
def approval(submitted_at, review_id=9):
|
|
return {"id": review_id, "state": "APPROVED", "commit_id": HEAD, "user": {"login": "alice"},
|
|
"submitted_at": submitted_at}
|
|
|
|
|
|
class TimedGitea(FakeGitea):
|
|
"""Gitea answers a new comment with its creation time."""
|
|
|
|
def create_issue_comment(self, number, body):
|
|
created = super().create_issue_comment(number, body)
|
|
self.comments[-1]["created_at"] = RESULT_AT
|
|
return {**created, "created_at": RESULT_AT}
|
|
|
|
|
|
def timed_gitea():
|
|
return TimedGitea(files=[gitea_file(DOC)], diff=diff_for(DOC, added=["Some text."]))
|
|
|
|
|
|
def yellow_run(gitea):
|
|
model = FakeModel(data=review_data(risk="high"), prompt_files=[DOC])
|
|
return run(gitea, model, cfg=config(**CFG))[0]
|
|
|
|
|
|
class ApprovalAfterResultTest(unittest.TestCase):
|
|
def test_an_approval_before_the_result_is_asked_for_again(self):
|
|
"""Scenario#75.01"""
|
|
gitea = timed_gitea()
|
|
gitea.reviews.append(approval(BEFORE))
|
|
outcome = yellow_run(gitea)
|
|
self.assertIs(outcome.result.verdict, Verdict.YELLOW)
|
|
self.assertEqual(gitea.requested, ["alice"])
|
|
self.assertEqual(gitea.scheduled_merges, [(7, "squash", True)])
|
|
self.assertIn("the approval of alice came before this result; review requested again, "
|
|
"please approve again after reading it", outcome.result.gate_note)
|
|
self.assertIn("please approve again after reading it", gitea.edited_comments[-1][1])
|
|
|
|
def test_an_approval_after_the_result_counts(self):
|
|
"""Scenario#75.02"""
|
|
gitea = docs_gitea()
|
|
gitea.comments.append({"id": 1, "user": {"login": "werkrevisor-bot"}, "created_at": RESULT_AT,
|
|
"body": f"<!-- werkrevisor:{HEAD} -->\n<!-- werkrevisor-verdict:YELLOW -->\n"})
|
|
gitea.reviews += [approval(BEFORE), approval(AFTER, review_id=10)]
|
|
outcome = yellow_run(gitea)
|
|
self.assertEqual(outcome.result.gate_note,
|
|
"approved by alice for this head commit; merge scheduled for when the checks succeed")
|
|
self.assertEqual(gitea.requested, [])
|
|
|
|
def test_a_failed_request_keeps_the_gate_closed(self):
|
|
gitea = timed_gitea()
|
|
gitea.reviews.append(approval(BEFORE))
|
|
gitea.fail_review_request = True
|
|
outcome = yellow_run(gitea)
|
|
self.assertEqual(outcome.result.exit_code, Verdict.YELLOW.exit_code)
|
|
self.assertIn("the new review request failed", outcome.result.gate_note)
|
|
|
|
|
|
class OtherGateModesTest(unittest.TestCase):
|
|
def test_block_does_not_count_an_approval_from_before_the_review(self):
|
|
"""Scenario#75.04"""
|
|
gitea = timed_gitea()
|
|
gitea.reviews.append(approval(BEFORE))
|
|
model = FakeModel(data=review_data(risk="high"), prompt_files=[DOC])
|
|
outcome, _ = run(gitea, model, cfg=config(gate={"yellow_gate": "block", "human_reviewers": ["alice"]}))
|
|
self.assertEqual(outcome.result.exit_code, Verdict.YELLOW.exit_code)
|
|
self.assertEqual(gitea.requested, ["alice"])
|
|
self.assertIn("the approval of alice came before this result", outcome.result.gate_note)
|
|
|
|
def test_block_names_a_failed_request(self):
|
|
gitea = timed_gitea()
|
|
gitea.reviews.append(approval(BEFORE))
|
|
gitea.fail_review_request = True
|
|
model = FakeModel(data=review_data(risk="high"), prompt_files=[DOC])
|
|
outcome, _ = run(gitea, model, cfg=config(gate={"yellow_gate": "block", "human_reviewers": ["alice"]}))
|
|
self.assertEqual(outcome.result.exit_code, Verdict.YELLOW.exit_code)
|
|
self.assertIn("came before this result, but the new review request failed: HTTP 422",
|
|
outcome.result.gate_note)
|
|
|
|
def test_pass_with_review_request_asks_again(self):
|
|
"""Scenario#75.04"""
|
|
gitea = timed_gitea()
|
|
gitea.reviews.append(approval(BEFORE))
|
|
model = FakeModel(data=review_data(risk="high"), prompt_files=[DOC])
|
|
cfg = config(gate={"yellow_gate": "pass_with_review_request", "human_reviewers": ["alice"]})
|
|
outcome, _ = run(gitea, model, cfg=cfg)
|
|
self.assertEqual(outcome.result.exit_code, 0)
|
|
self.assertEqual(gitea.requested, ["alice"])
|
|
self.assertIn("gate passes because a human review request is pending; the approval of alice came before",
|
|
outcome.result.gate_note)
|
|
|
|
def test_block_counts_an_approval_after_an_earlier_result(self):
|
|
gitea = docs_gitea()
|
|
gitea.comments.append({"id": 1, "user": {"login": "werkrevisor-bot"}, "created_at": RESULT_AT,
|
|
"body": f"<!-- werkrevisor:{HEAD} -->\n<!-- werkrevisor-verdict:YELLOW -->\n"})
|
|
gitea.reviews.append(approval(AFTER))
|
|
outcome, _ = run(gitea, cfg=config(gate={"yellow_gate": "block", "human_reviewers": ["alice"]}))
|
|
self.assertEqual(outcome.result.exit_code, 0)
|
|
self.assertEqual(outcome.result.gate_note, "unlocked by the approval of alice for this head commit")
|
|
|
|
|
|
class EarlyApproversTest(unittest.TestCase):
|
|
def test_times_decide(self):
|
|
reviews = [approval(BEFORE)]
|
|
self.assertIsNone(human_approval(reviews, HEAD, ["alice"], RESULT_AT))
|
|
self.assertEqual(early_approvers(reviews, HEAD, ["alice"], RESULT_AT), ["alice"])
|
|
reviews.append(approval(AFTER, review_id=10))
|
|
self.assertEqual(human_approval(reviews, HEAD, ["alice"], RESULT_AT), "alice")
|
|
self.assertEqual(early_approvers(reviews, HEAD, ["alice"], RESULT_AT), [])
|
|
|
|
def test_unknown_times_keep_the_approval(self):
|
|
for submitted, since in ((None, RESULT_AT), (BEFORE, None), ("garbled", RESULT_AT)):
|
|
with self.subTest(submitted=submitted, since=since):
|
|
self.assertEqual(human_approval([approval(submitted)], HEAD, ["alice"], since), "alice")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|