sysadmin
d4dc9aa854
PR #416 introduced pr_work_lease reviewer blocks and mandatory full 40-hex expected_head_sha on mark/merge paths. Update audit, review, permission-report, terminal hard-stop, inventory, and schema tests to patch lease comment fetches, seed ready review decisions, and use consistent HEAD_SHA constants so the full suite passes again.
195 lines
7.2 KiB
Python
195 lines
7.2 KiB
Python
"""Tests for the session hard-stop after a terminal review mutation (#332).
|
|
|
|
Extends the single-terminal-decision machinery (#211): after a live
|
|
REQUEST_CHANGES the run must stop; after an APPROVE only the merge sequence
|
|
for that same PR may continue; a different PR can never be reviewed or
|
|
merged in the same run; duplicate REQUEST_CHANGES at an unchanged head is
|
|
suppressed.
|
|
"""
|
|
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import mcp_server
|
|
|
|
HEAD_SHA = "a" * 40
|
|
|
|
|
|
def _lock(mutations=None, correction=False):
|
|
return {
|
|
"task": "review_pr",
|
|
"remote": "prgs",
|
|
"session_pid": os.getpid(),
|
|
"session_profile": "prgs-reviewer",
|
|
"session_profile_lock": "",
|
|
"final_review_decision_ready": False,
|
|
"ready_pr_number": None,
|
|
"ready_action": None,
|
|
"ready_expected_head_sha": None,
|
|
"ready_remote": None,
|
|
"ready_org": None,
|
|
"ready_repo": None,
|
|
"live_mutations": list(mutations or []),
|
|
"correction_authorized": correction,
|
|
"correction_reason": None,
|
|
}
|
|
|
|
|
|
def _seed(mutations=None, correction=False):
|
|
mcp_server._save_review_decision_lock(_lock(mutations, correction))
|
|
|
|
|
|
APPROVED_A = {"pr_number": 5, "action": "approve", "review_id": 1,
|
|
"review_state": "approve"}
|
|
RC_A = {"pr_number": 5, "action": "request_changes", "review_id": 2,
|
|
"review_state": "request_changes"}
|
|
|
|
|
|
class TestTerminalHardStopReasons(unittest.TestCase):
|
|
def tearDown(self):
|
|
mcp_server._save_review_decision_lock(None)
|
|
|
|
def test_no_lock_no_reasons(self):
|
|
mcp_server._save_review_decision_lock(None)
|
|
self.assertEqual(
|
|
mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
|
|
|
|
def test_no_terminal_mutation_no_reasons(self):
|
|
_seed()
|
|
self.assertEqual(
|
|
mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
|
|
|
|
def test_request_changes_blocks_everything(self):
|
|
_seed([RC_A])
|
|
for op in ("merge", "mark_ready", "review"):
|
|
for pr in (5, 6):
|
|
reasons = mcp_server.terminal_review_hard_stop_reasons(pr, op)
|
|
self.assertTrue(reasons, f"{op}/{pr}")
|
|
self.assertIn("request_changes on PR #5", reasons[0])
|
|
self.assertIn("#332", reasons[0])
|
|
|
|
def test_approve_allows_same_pr_merge_only(self):
|
|
_seed([APPROVED_A])
|
|
self.assertEqual(
|
|
mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
|
|
self.assertTrue(
|
|
mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
|
|
self.assertTrue(
|
|
mcp_server.terminal_review_hard_stop_reasons(6, "mark_ready"))
|
|
self.assertTrue(
|
|
mcp_server.terminal_review_hard_stop_reasons(6, "review"))
|
|
|
|
def test_correction_reopens_review_path_only(self):
|
|
_seed([RC_A], correction=True)
|
|
self.assertEqual(
|
|
mcp_server.terminal_review_hard_stop_reasons(5, "mark_ready"), [])
|
|
self.assertEqual(
|
|
mcp_server.terminal_review_hard_stop_reasons(5, "review"), [])
|
|
# correction never opens a cross-PR merge
|
|
self.assertTrue(
|
|
mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
|
|
|
|
|
|
class TestMergeHardStopWiring(unittest.TestCase):
|
|
def tearDown(self):
|
|
mcp_server._save_review_decision_lock(None)
|
|
|
|
def test_merge_blocked_after_request_changes(self):
|
|
_seed([RC_A])
|
|
result = mcp_server.gitea_merge_pr(
|
|
pr_number=6, confirmation="MERGE PR 6", remote="prgs")
|
|
self.assertFalse(result["performed"])
|
|
self.assertTrue(
|
|
any("#332" in r for r in result["reasons"]), result["reasons"])
|
|
|
|
def test_merge_of_other_pr_blocked_after_approve(self):
|
|
_seed([APPROVED_A])
|
|
result = mcp_server.gitea_merge_pr(
|
|
pr_number=6, confirmation="MERGE PR 6", remote="prgs")
|
|
self.assertFalse(result["performed"])
|
|
self.assertTrue(
|
|
any("#332" in r for r in result["reasons"]), result["reasons"])
|
|
|
|
|
|
class TestMarkFinalHardStopWiring(unittest.TestCase):
|
|
def tearDown(self):
|
|
mcp_server._save_review_decision_lock(None)
|
|
|
|
def test_mark_ready_blocked_after_terminal_mutation(self):
|
|
_seed([RC_A])
|
|
result = mcp_server.gitea_mark_final_review_decision(
|
|
pr_number=6, action="approve", remote="prgs")
|
|
self.assertFalse(result["marked_ready"])
|
|
self.assertTrue(
|
|
any("#332" in r for r in result["reasons"]), result["reasons"])
|
|
|
|
|
|
def _feedback(blocking, stale=False, success=True):
|
|
return {
|
|
"success": success,
|
|
"has_blocking_change_requests": blocking,
|
|
"review_feedback_stale": stale,
|
|
"current_head_sha": HEAD_SHA,
|
|
}
|
|
|
|
|
|
def _mark(action, pr_number=6, **kwargs):
|
|
kwargs.setdefault("expected_head_sha", HEAD_SHA)
|
|
no_lease_block = {"block": False, "reasons": [], "mutation_allowed": True}
|
|
with patch("mcp_server._list_pr_lease_comments", return_value=[]), \
|
|
patch("mcp_server._pr_work_lease_reviewer_block", return_value=no_lease_block):
|
|
return mcp_server.gitea_mark_final_review_decision(
|
|
pr_number=pr_number, action=action, remote="prgs", **kwargs
|
|
)
|
|
|
|
|
|
class TestDuplicateRequestChangesSuppression(unittest.TestCase):
|
|
def tearDown(self):
|
|
mcp_server._save_review_decision_lock(None)
|
|
|
|
def test_duplicate_request_changes_blocked_at_same_head(self):
|
|
_seed()
|
|
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
|
|
return_value=_feedback(blocking=True, stale=False)):
|
|
result = _mark("request_changes")
|
|
self.assertFalse(result["marked_ready"])
|
|
self.assertTrue(
|
|
any("duplicate" in r for r in result["reasons"]),
|
|
result["reasons"])
|
|
|
|
def test_request_changes_allowed_when_prior_blocker_stale(self):
|
|
_seed()
|
|
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
|
|
return_value=_feedback(blocking=True, stale=True)):
|
|
result = _mark("request_changes")
|
|
self.assertTrue(result["marked_ready"], result.get("reasons"))
|
|
|
|
def test_request_changes_allowed_when_no_blocker(self):
|
|
_seed()
|
|
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
|
|
return_value=_feedback(blocking=False)):
|
|
result = _mark("request_changes")
|
|
self.assertTrue(result["marked_ready"], result.get("reasons"))
|
|
|
|
def test_request_changes_fails_closed_when_feedback_unavailable(self):
|
|
_seed()
|
|
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
|
|
return_value=_feedback(blocking=False,
|
|
success=False)):
|
|
result = _mark("request_changes")
|
|
self.assertFalse(result["marked_ready"])
|
|
self.assertTrue(
|
|
any("could not verify" in r for r in result["reasons"]),
|
|
result["reasons"])
|
|
|
|
def test_approve_does_not_fetch_feedback(self):
|
|
_seed()
|
|
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
|
|
side_effect=AssertionError("must not be called")):
|
|
result = _mark("approve")
|
|
self.assertTrue(result["marked_ready"], result.get("reasons"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|