Merge pull request 'feat: enforce PR review/merge workflow state machine gates (Closes #290)' (#378) from feat/issue-290-review-merge-state-machine into master

2026-07-07 09:02:42 -05:00
parent ae566d5cd1 60040e789b
commit 0c18d0f54f
3 changed files with 521 additions and 0 deletions
@@ -0,0 +1,121 @@
+"""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()