292 lines
14 KiB
Python
292 lines
14 KiB
Python
import unittest
|
|
|
|
from tests.fakes import BOT, FakeGitea, FakeModel, config, diff_for, gitea_file, review_data
|
|
from tests.test_flow import run
|
|
from tests.test_line_comments import APP, app_gitea
|
|
from werkrevisor import codex
|
|
from werkrevisor.verdict import Verdict
|
|
|
|
AUTOMERGE = {"yellow_gate": "approval", "human_reviewers": ["alice"], "automerge": True, "require_wip_transition": True}
|
|
|
|
|
|
OUTSIDE = "other.py"
|
|
|
|
|
|
def hint_run(gitea=None, issues=(APP,), start=40, **gate):
|
|
data = review_data(issues=list(issues), header="[low] Naming")
|
|
for issue in data["review"]["key_issues_to_review"]:
|
|
issue.update(start_line=start, end_line=start + 1 if start else 0)
|
|
gitea = gitea or app_gitea()
|
|
outcome, _ = run(gitea, FakeModel(data=data, prompt_files=[APP]), cfg=config(gate={**AUTOMERGE, **gate}))
|
|
return outcome, gitea
|
|
|
|
|
|
DOC = "docs/guide.md"
|
|
LONG_LINE = "A sentence that runs on and on " * 6
|
|
WRITING_BODY = "**Writing rule: line longer than 128 characters (180)** (werkrevisor writing rule)"
|
|
|
|
|
|
def writing_run(gitea=None):
|
|
gitea = gitea or FakeGitea(files=[gitea_file(DOC)], diff=diff_for(DOC, added=["Short.", LONG_LINE]))
|
|
gitea.raw_files[DOC] = f"Short.\n{LONG_LINE}\n".encode()
|
|
outcome, _ = run(gitea, FakeModel(prompt_files=[DOC]), cfg=config(gate=AUTOMERGE))
|
|
return outcome, gitea
|
|
|
|
|
|
class HintsHoldTheMergeTest(unittest.TestCase):
|
|
def test_a_writing_rule_violation_is_a_line_comment_that_holds_the_merge(self):
|
|
outcome, gitea = writing_run()
|
|
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
|
self.assertEqual([r[0] for r in gitea.created_reviews], ["APPROVED"])
|
|
[comment] = gitea.review_comments[0]
|
|
self.assertEqual((comment["path"], comment["new_position"]), (DOC, 2))
|
|
self.assertIn("line longer than 128 characters", comment["body"])
|
|
self.assertEqual(gitea.scheduled_merges, [])
|
|
self.assertIn("merge waits for 1 open hint(s)", outcome.result.gate_note)
|
|
|
|
def test_a_resolved_writing_rule_comment_releases_the_merge(self):
|
|
outcome, gitea = writing_run()
|
|
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
|
self.assertEqual(gitea.scheduled_merges, [])
|
|
for comments in gitea.line_comments.values():
|
|
for c in comments:
|
|
c["resolver"] = {"login": "claude-code"}
|
|
outcome, gitea = writing_run(gitea)
|
|
self.assertEqual(outcome.result.llm_status, "reused")
|
|
self.assertEqual(len(gitea.scheduled_merges), 1)
|
|
|
|
def test_a_codex_line_comment_holds_no_merge(self):
|
|
gitea = FakeGitea(files=[gitea_file(DOC)], diff=diff_for(DOC, added=["Short."]))
|
|
gitea.raw_files[DOC] = b"Short.\n"
|
|
gitea.create_review(7, "COMMENT", "Codex review\n" + codex.COMMENT_MARKER, gitea.heads[0],
|
|
[{"path": DOC, "new_position": 1, "old_position": 0,
|
|
"body": "**[high] Lost grant** (Codex review)\n" + codex.COMMENT_MARKER}])
|
|
outcome, _ = run(gitea, FakeModel(prompt_files=[DOC]), cfg=config(gate=AUTOMERGE))
|
|
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
|
self.assertEqual(len(gitea.scheduled_merges), 1)
|
|
|
|
def test_a_refused_writing_rule_comment_still_holds_the_merge(self):
|
|
gitea = FakeGitea(files=[gitea_file(DOC)], diff=diff_for(DOC, added=["Short.", LONG_LINE]))
|
|
gitea.refuse_line_comments = 500
|
|
outcome, gitea = writing_run(gitea)
|
|
self.assertEqual(gitea.scheduled_merges, [])
|
|
self.assertIn("merge waits for 1 open hint(s)", outcome.result.gate_note)
|
|
|
|
def test_an_open_writing_rule_comment_is_reused(self):
|
|
gitea = FakeGitea(files=[gitea_file(DOC)], diff=diff_for(DOC, added=["Short.", LONG_LINE]))
|
|
gitea.reviews.append({"id": 7, "state": "APPROVED", "commit_id": "c" * 40, "user": {"login": BOT}})
|
|
gitea.line_comments[7] = [{"id": 77, "user": {"login": BOT}, "path": DOC, "position": 2, "resolver": None,
|
|
"body": WRITING_BODY, "html_url": "https://x/77"}]
|
|
outcome, gitea = writing_run(gitea)
|
|
self.assertEqual(gitea.review_comments, [[]])
|
|
self.assertIn("merge waits for 1 open hint(s)", outcome.result.gate_note)
|
|
|
|
def test_a_finding_never_takes_over_a_writing_rule_comment(self):
|
|
gitea = app_gitea()
|
|
gitea.reviews.append({"id": 7, "state": "APPROVED", "commit_id": "c" * 40, "user": {"login": BOT}})
|
|
gitea.line_comments[7] = [{"id": 77, "user": {"login": BOT}, "path": APP, "position": 41, "resolver": None,
|
|
"body": WRITING_BODY, "html_url": "https://x/77"}]
|
|
outcome, gitea = hint_run(gitea)
|
|
self.assertEqual(len(gitea.review_comments[0]), 1)
|
|
self.assertEqual([c for c, _ in gitea.edited_comments if c == 77], [])
|
|
|
|
def test_a_green_review_with_open_hints_approves_but_schedules_no_merge(self):
|
|
gitea = app_gitea()
|
|
gitea.scheduled_merges.append((7, "merge", False))
|
|
outcome, gitea = hint_run(gitea)
|
|
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
|
self.assertEqual([r[0] for r in gitea.created_reviews], ["APPROVED"])
|
|
self.assertEqual(len(gitea.review_comments[0]), 1)
|
|
# The merge scheduled for an earlier head would still merge this one.
|
|
self.assertEqual((gitea.scheduled_merges, gitea.cancelled_merges), ([], [7]))
|
|
self.assertIn("merge waits for 1 open hint(s)", outcome.result.gate_note)
|
|
self.assertIn("merge waits for 1 open hint(s)", gitea.edited_comments[-1][1])
|
|
|
|
def test_a_hint_without_a_line_asks_for_an_answer_in_a_pr_comment(self):
|
|
"""Scenario#95.01"""
|
|
outcome, gitea = hint_run(start=0)
|
|
self.assertEqual(gitea.review_comments, [[]])
|
|
self.assertEqual(gitea.scheduled_merges, [])
|
|
self.assertIn("merge waits for 1 open hint(s): answer the 1 hint(s) that stand only in the result comment "
|
|
"in a PR comment, then restart the review or push a commit", gitea.edited_comments[-1][1])
|
|
self.assertNotIn("line comment and resolve", outcome.result.gate_note)
|
|
|
|
def test_line_comments_and_hints_without_a_line_are_named_apart(self):
|
|
"""Scenario#95.02"""
|
|
gitea = app_gitea()
|
|
gitea.refuse_line_comments = True
|
|
gitea.line_comments[7] = [{"id": 77, "user": {"login": BOT}, "path": APP, "position": 2, "resolver": None,
|
|
"body": WRITING_BODY, "html_url": "https://x/77"}]
|
|
gitea.reviews.append({"id": 7, "state": "APPROVED", "commit_id": "c" * 40, "user": {"login": BOT}})
|
|
outcome, gitea = hint_run(gitea)
|
|
self.assertIn("merge waits for 2 open hint(s): reply to each line comment and resolve it; "
|
|
"answer the 1 hint(s) that stand only in the result comment in a PR comment, "
|
|
"then restart the review or push a commit", outcome.result.gate_note)
|
|
|
|
def test_a_restart_releases_a_hint_without_a_line(self):
|
|
outcome, gitea = hint_run(start=0)
|
|
outcome, gitea = hint_run(gitea, start=0)
|
|
self.assertEqual(outcome.result.llm_status, "reused")
|
|
self.assertEqual(len(gitea.scheduled_merges), 1)
|
|
|
|
def test_a_hint_outside_the_diff_holds_no_merge(self):
|
|
"""Scenario#97.01"""
|
|
outcome, gitea = hint_run(issues=[OUTSIDE])
|
|
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
|
self.assertEqual(gitea.review_comments, [[]])
|
|
self.assertIn(f"`{OUTSIDE}:40-41`", gitea.created_comments[-1])
|
|
self.assertEqual(len(gitea.scheduled_merges), 1)
|
|
|
|
def test_a_restarted_review_schedules_the_merge_once_all_hints_are_resolved(self):
|
|
outcome, gitea = hint_run()
|
|
self.assertEqual(gitea.scheduled_merges, [])
|
|
for comments in gitea.line_comments.values():
|
|
for c in comments:
|
|
c["resolver"] = {"login": "claude-code"}
|
|
outcome, gitea = hint_run(gitea)
|
|
self.assertEqual(outcome.result.llm_status, "reused")
|
|
self.assertEqual(len(gitea.scheduled_merges), 1)
|
|
self.assertIn("merge scheduled", outcome.result.gate_note)
|
|
self.assertIn("review restarted", gitea.created_comments[-1])
|
|
self.assertIn("merge scheduled for when the checks succeed", gitea.created_comments[-1])
|
|
|
|
def test_an_open_reply_to_a_resolved_hint_does_not_hold_the_merge(self):
|
|
outcome, gitea = hint_run()
|
|
[(review_id, [hint])] = gitea.line_comments.items()
|
|
hint["resolver"] = {"login": "claude-code"}
|
|
gitea.line_comments[review_id].append({**hint, "id": 1, "user": {"login": "claude-code"}, "resolver": None})
|
|
outcome, gitea = hint_run(gitea)
|
|
self.assertEqual(len(gitea.scheduled_merges), 1)
|
|
|
|
def test_a_scheduled_merge_withdraws_the_review_request_of_an_earlier_round(self):
|
|
gitea = app_gitea()
|
|
gitea.requested = ["alice", "carol"]
|
|
outcome, gitea = hint_run(gitea, hints_hold_merge=False)
|
|
self.assertEqual(len(gitea.scheduled_merges), 1)
|
|
# Gitea blocks a merge while an official review request is open.
|
|
self.assertEqual(gitea.requested, ["carol"])
|
|
self.assertIn("review request of alice withdrawn", outcome.result.gate_note)
|
|
|
|
def test_a_request_listed_only_as_review_is_withdrawn_too(self):
|
|
gitea = app_gitea()
|
|
gitea.reviews.append({"id": 9, "state": "REQUEST_REVIEW", "user": {"login": "alice"}, "commit_id": ""})
|
|
outcome, gitea = hint_run(gitea, hints_hold_merge=False)
|
|
self.assertEqual(gitea.withdrawn, ["alice"])
|
|
|
|
def test_a_merge_not_scheduled_keeps_the_review_request(self):
|
|
gitea = app_gitea()
|
|
gitea.requested = ["alice"]
|
|
gitea.branch["required_approvals"] = 0
|
|
outcome, gitea = hint_run(gitea, hints_hold_merge=False)
|
|
self.assertIn("merge not scheduled", outcome.result.gate_note)
|
|
self.assertEqual(gitea.requested, ["alice"])
|
|
|
|
def test_a_held_merge_keeps_the_review_request(self):
|
|
gitea = app_gitea()
|
|
gitea.requested = ["alice"]
|
|
outcome, gitea = hint_run(gitea)
|
|
self.assertEqual(gitea.requested, ["alice"])
|
|
|
|
def test_a_failed_withdrawal_is_named_and_keeps_the_review_alive(self):
|
|
gitea = app_gitea()
|
|
gitea.requested = ["alice"]
|
|
gitea.fail_review_request = True
|
|
outcome, gitea = hint_run(gitea, hints_hold_merge=False)
|
|
self.assertEqual(outcome.result.exit_code, 0)
|
|
self.assertIn("review request of alice not withdrawn", outcome.result.gate_note)
|
|
|
|
def test_an_open_reply_to_an_open_hint_is_no_second_hint(self):
|
|
outcome, gitea = hint_run()
|
|
[(review_id, [hint])] = gitea.line_comments.items()
|
|
gitea.line_comments[review_id].append({**hint, "id": 1, "user": {"login": "claude-code"}})
|
|
outcome, gitea = hint_run(gitea)
|
|
self.assertIn("merge waits for 1 open hint(s)", outcome.result.gate_note)
|
|
|
|
def test_a_later_edit_shortens_the_result_once_the_line_comments_exist(self):
|
|
gitea = app_gitea()
|
|
edit, calls = gitea.edit_issue_comment, []
|
|
|
|
def first_fails(comment_id, body):
|
|
calls.append(comment_id)
|
|
if len(calls) == 1:
|
|
raise OSError("HTTP 502")
|
|
return edit(comment_id, body)
|
|
gitea.edit_issue_comment = first_fails
|
|
outcome, gitea = hint_run(gitea)
|
|
body = gitea.comments[-1]["body"]
|
|
self.assertEqual(len(gitea.review_comments[0]), 1)
|
|
self.assertIn("as a line comment", body)
|
|
self.assertIn("merge waits for 1 open hint(s)", body)
|
|
|
|
def test_a_restart_with_open_hints_comments_that_the_merge_still_waits(self):
|
|
outcome, gitea = hint_run()
|
|
outcome, gitea = hint_run(gitea)
|
|
self.assertEqual(gitea.scheduled_merges, [])
|
|
self.assertIn("merge waits for 1 open hint(s)", gitea.created_comments[-1])
|
|
|
|
def test_hints_refused_as_line_comments_hold_the_merge_too(self):
|
|
gitea = app_gitea()
|
|
gitea.refuse_line_comments = True
|
|
outcome, gitea = hint_run(gitea)
|
|
self.assertIn("line_comments_refused", [r.code for r in outcome.result.reasons])
|
|
self.assertEqual(gitea.scheduled_merges, [])
|
|
self.assertIn("merge waits for 1 open hint(s)", outcome.result.gate_note)
|
|
outcome, gitea = hint_run(gitea)
|
|
self.assertEqual(len(gitea.scheduled_merges), 1)
|
|
|
|
def test_a_failed_cancel_is_named(self):
|
|
from werkrevisor.gitea import GiteaError
|
|
gitea = app_gitea()
|
|
|
|
def refuse(number):
|
|
raise GiteaError("DELETE merge failed with HTTP 403", 403)
|
|
gitea.cancel_scheduled_merge = refuse
|
|
outcome, gitea = hint_run(gitea)
|
|
self.assertIn("a merge scheduled earlier could not be cancelled: DELETE merge failed with HTTP 403",
|
|
gitea.edited_comments[-1][1])
|
|
# The failed gate check stops the merge that Gitea still has scheduled.
|
|
self.assertEqual(outcome.result.exit_code, 10)
|
|
|
|
def test_unreadable_hints_hold_the_merge(self):
|
|
gitea = app_gitea()
|
|
|
|
def unreadable(number, review_id):
|
|
raise OSError("HTTP 502")
|
|
outcome, gitea = hint_run(gitea)
|
|
for comments in gitea.line_comments.values():
|
|
for c in comments:
|
|
c["resolver"] = {"login": "claude-code"}
|
|
gitea.list_review_comments = unreadable
|
|
outcome, gitea = hint_run(gitea)
|
|
self.assertEqual(gitea.scheduled_merges, [])
|
|
self.assertEqual(outcome.result.exit_code, 10)
|
|
self.assertIn("the open hints could not be read: HTTP 502", outcome.result.gate_note)
|
|
|
|
def test_a_restart_comments_only_a_change_of_the_merge(self):
|
|
outcome, gitea = hint_run()
|
|
for comments in gitea.line_comments.values():
|
|
for c in comments:
|
|
c["resolver"] = {"login": "claude-code"}
|
|
hint_run(gitea)
|
|
comments = len(gitea.created_comments)
|
|
outcome, gitea = hint_run(gitea)
|
|
self.assertIn("merge already scheduled", outcome.result.gate_note)
|
|
self.assertEqual(len(gitea.created_comments), comments)
|
|
|
|
def test_the_hold_can_be_switched_off(self):
|
|
outcome, gitea = hint_run(hints_hold_merge=False)
|
|
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
|
self.assertEqual(len(gitea.scheduled_merges), 1)
|
|
self.assertEqual(gitea.cancelled_merges, [])
|
|
|
|
|
|
class HintsHoldMergeConfigTest(unittest.TestCase):
|
|
def test_the_switch_must_be_a_boolean(self):
|
|
from werkrevisor.config import ConfigError
|
|
with self.assertRaisesRegex(ConfigError, r"\[gate\] hints_hold_merge must be true or false"):
|
|
config(gate={**AUTOMERGE, "hints_hold_merge": "false"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|