sysadmin
8d32af14d6
Add fail-closed review/merge state progression helpers, pre-merge gate requirements, blocked recovery handoff checks, and MCP assessment tool. Closes #290
121 lines
4.5 KiB
Python
121 lines
4.5 KiB
Python
"""Tests for enforced PR review/merge state machine (#290)."""
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
import review_merge_state_machine as rmsm # noqa: E402
|
|
|
|
FULL_REVIEW_COMPLETION = {state: True for state in rmsm.REVIEW_MERGE_STATES}
|
|
|
|
|
|
def _completion_through(state_name: str) -> dict[str, bool]:
|
|
idx = rmsm.state_index(state_name)
|
|
return {state: i <= idx for i, state in enumerate(rmsm.REVIEW_MERGE_STATES)}
|
|
|
|
|
|
class TestWorkflowBlockers(unittest.TestCase):
|
|
def test_infra_stop_blocks_all_states(self):
|
|
result = rmsm.assess_workflow_blockers(infra_stop=True)
|
|
self.assertTrue(result["block"])
|
|
self.assertEqual(result["forbidden_states"], list(rmsm.REVIEW_MERGE_STATES))
|
|
|
|
def test_infra_stop_forbids_advancement(self):
|
|
result = rmsm.assess_state_advancement(
|
|
{},
|
|
target_state="INVENTORY",
|
|
infra_stop=True,
|
|
)
|
|
self.assertTrue(result["block"])
|
|
self.assertFalse(result["allowed"])
|
|
|
|
|
|
class TestStateAdvancement(unittest.TestCase):
|
|
def test_cannot_skip_inventory(self):
|
|
result = rmsm.assess_state_advancement(
|
|
{"PRECHECK": True},
|
|
target_state="SELECT_PR",
|
|
)
|
|
self.assertTrue(result["block"])
|
|
self.assertEqual(result["next_allowed_state"], "INVENTORY")
|
|
|
|
def test_sequential_advance_allowed(self):
|
|
completion = _completion_through("INVENTORY")
|
|
result = rmsm.assess_state_advancement(
|
|
completion,
|
|
target_state="SELECT_PR",
|
|
)
|
|
self.assertFalse(result["block"])
|
|
self.assertTrue(result["allowed"])
|
|
|
|
|
|
class TestApproveAndMergeGates(unittest.TestCase):
|
|
def test_approve_blocked_without_validate(self):
|
|
completion = _completion_through("PIN_HEAD_SHA")
|
|
result = rmsm.can_approve(completion)
|
|
self.assertTrue(result["block"])
|
|
self.assertIn("VALIDATE", ",".join(result["missing_states"]))
|
|
|
|
def test_approve_allowed_when_review_path_complete(self):
|
|
completion = _completion_through("REVIEW_DECISION")
|
|
result = rmsm.can_approve(completion)
|
|
self.assertFalse(result["block"])
|
|
self.assertTrue(result["allowed"])
|
|
|
|
def test_merge_blocked_without_pre_merge_gates(self):
|
|
completion = _completion_through("APPROVE_OR_REQUEST_CHANGES")
|
|
result = rmsm.can_merge(completion)
|
|
self.assertTrue(result["block"])
|
|
self.assertTrue(result["missing_pre_merge_gates"])
|
|
|
|
def test_merge_allowed_with_pre_merge_gates(self):
|
|
completion = _completion_through("PRE_MERGE_RECHECK")
|
|
gates = {gate: True for gate in rmsm._PRE_MERGE_REQUIRED_GATES}
|
|
result = rmsm.can_merge(completion, pre_merge_gates=gates)
|
|
self.assertFalse(result["block"])
|
|
self.assertTrue(result["allowed"])
|
|
|
|
def test_merge_blocked_when_head_sha_gate_fails(self):
|
|
completion = _completion_through("PRE_MERGE_RECHECK")
|
|
gates = {gate: True for gate in rmsm._PRE_MERGE_REQUIRED_GATES}
|
|
gates["reviewed_head_sha_unchanged"] = False
|
|
result = rmsm.can_merge(completion, pre_merge_gates=gates)
|
|
self.assertTrue(result["block"])
|
|
self.assertIn("reviewed_head_sha_unchanged", result["missing_pre_merge_gates"])
|
|
|
|
|
|
class TestRecoveryHandoff(unittest.TestCase):
|
|
def test_blocked_handoff_without_replay_passes(self):
|
|
report = (
|
|
"Blocked recovery handoff.\n"
|
|
"Restart the full workflow after infra_stop clears.\n"
|
|
"No approve/merge performed."
|
|
)
|
|
result = rmsm.assess_blocked_recovery_handoff(report)
|
|
self.assertFalse(result["block"])
|
|
|
|
def test_blocked_handoff_with_merge_replay_fails(self):
|
|
report = "Please merge PR #12 now using gitea_merge_pr."
|
|
result = rmsm.assess_blocked_recovery_handoff(report)
|
|
self.assertTrue(result["block"])
|
|
|
|
|
|
class TestFinalReportClaims(unittest.TestCase):
|
|
def test_ready_to_merge_claim_without_gates_fails(self):
|
|
result = rmsm.assess_final_report_state_claims(
|
|
"PR is ready to merge after validation.",
|
|
state_completion=_completion_through("VALIDATE"),
|
|
)
|
|
self.assertTrue(result["block"])
|
|
|
|
def test_reviewed_claim_without_validate_fails(self):
|
|
result = rmsm.assess_final_report_state_claims(
|
|
"PR reviewed and looks good.",
|
|
state_completion={"PRECHECK": True},
|
|
)
|
|
self.assertTrue(result["block"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |