Merge pull request 'feat: hard-stop review mutations on already-landed PR heads (Closes #292)' (#368) from feat/issue-292-already-landed-gate into master

tests/test_review_proofs.py

@@ -23,6 +23,8 @@ sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.par
 from review_proofs import (  # noqa: E402
     ISSUE_SELECTION_CONTINUATION_EXPLICIT,
     ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
+    assess_already_landed_report_state,
+    assess_already_landed_review_gate,
     assess_author_pr_report,
     assess_validation_environment_proof,
     assess_full_suite_failure_approval_gate,
@@ -2639,6 +2641,139 @@ class TestReviewerBaselineValidationProof(unittest.TestCase):
         self.assertFalse(report["baseline_validation_proven"])
 
 
+class TestAlreadyLandedReviewGate(unittest.TestCase):
+    """Issue #292: PR head already on target blocks approval and merge."""
+
+    def _gate(self, **overrides):
+        kwargs = {
+            "pr_number": 278,
+            "candidate_head_sha": PINNED,
+            "target_branch": "master",
+            "target_branch_sha": OTHER,
+            "head_is_ancestor_of_target": True,
+            "live_head_sha": PINNED,
+        }
+        kwargs.update(overrides)
+        return assess_already_landed_review_gate(**kwargs)
+
+    def test_already_landed_blocks_approval_and_merge(self):
+        result = self._gate()
+        self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED")
+        self.assertFalse(result["approve_allowed"])
+        self.assertFalse(result["merge_api_allowed"])
+        self.assertFalse(result["request_changes_allowed"])
+        self.assertTrue(result["reconciliation_handoff_required"])
+
+    def test_already_landed_with_real_blocker_allows_request_changes(self):
+        result = self._gate(real_blocker=True)
+        self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED")
+        self.assertFalse(result["approve_allowed"])
+        self.assertFalse(result["merge_api_allowed"])
+        self.assertTrue(result["request_changes_allowed"])
+
+    def test_normal_candidate_passes_gate(self):
+        result = self._gate(head_is_ancestor_of_target=False)
+        self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE")
+        self.assertTrue(result["approve_allowed"])
+        self.assertTrue(result["merge_api_allowed"])
+        self.assertFalse(result["reconciliation_handoff_required"])
+
+    def test_non_mergeable_normal_pr_still_passes_this_gate(self):
+        # Mergeability/conflicts are separate gates; ancestry gate only
+        # decides already-landed vs normal candidate.
+        result = self._gate(
+            head_is_ancestor_of_target=False, mergeable=False)
+        self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE")
+        self.assertTrue(result["request_changes_allowed"])
+
+    def test_unchecked_ancestry_blocks_review_mutations(self):
+        result = self._gate(head_is_ancestor_of_target=None)
+        self.assertEqual(result["state"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["approve_allowed"])
+        self.assertFalse(result["merge_api_allowed"])
+
+    def test_changed_head_since_pin_blocks_until_recheck(self):
+        result = self._gate(
+            head_is_ancestor_of_target=False, live_head_sha=OTHER)
+        self.assertEqual(result["state"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["approve_allowed"])
+        self.assertTrue(any(
+            "head" in reason.lower() for reason in result["reasons"]
+        ))
+
+    def test_invalid_shas_block_gate(self):
+        result = self._gate(candidate_head_sha="deadbeef")
+        self.assertEqual(result["state"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["approve_allowed"])
+
+        result = self._gate(target_branch_sha=None)
+        self.assertEqual(result["state"], "GATE_NOT_PROVEN")
+
+    def test_already_landed_handoff_lists_required_fields(self):
+        result = self._gate()
+        for field in (
+            "PR number/title",
+            "candidate head SHA",
+            "target branch",
+            "target branch SHA",
+            "ancestor proof",
+            "linked issue status",
+            "recommended reconciliation action",
+            "capability proof for close/comment mutations",
+        ):
+            self.assertIn(field, result["required_handoff_fields"])
+
+
+class TestAlreadyLandedReportState(unittest.TestCase):
+    """Issue #292: reports cannot say APPROVED after the gate fired."""
+
+    GOOD_REPORT = (
+        "Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n"
+        "Candidate head SHA: " + PINNED + "\n"
+        "Review decision: none\n"
+        "Merge result: none\n"
+    )
+
+    def test_gate_fired_with_reconcile_state_passes(self):
+        result = assess_already_landed_report_state(
+            self.GOOD_REPORT, gate_fired=True)
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["block"])
+
+    def test_gate_fired_with_approved_state_blocks(self):
+        report = self.GOOD_REPORT.replace(
+            "Review decision: none", "Review decision: APPROVED")
+        result = assess_already_landed_report_state(report, gate_fired=True)
+        self.assertTrue(result["block"])
+
+    def test_gate_fired_with_merged_state_blocks(self):
+        report = self.GOOD_REPORT.replace(
+            "Merge result: none", "Merge result: MERGED")
+        result = assess_already_landed_report_state(report, gate_fired=True)
+        self.assertTrue(result["block"])
+
+    def test_gate_fired_with_ready_to_merge_blocks(self):
+        result = assess_already_landed_report_state(
+            self.GOOD_REPORT + "Current status: READY_TO_MERGE\n",
+            gate_fired=True)
+        self.assertTrue(result["block"])
+
+    def test_gate_fired_without_reconcile_state_blocks(self):
+        result = assess_already_landed_report_state(
+            "Current status: review complete.\n", gate_fired=True)
+        self.assertTrue(result["block"])
+        self.assertTrue(any(
+            "already_landed_reconcile_required" in reason.lower()
+            for reason in result["reasons"]
+        ))
+
+    def test_gate_not_fired_reports_pass_untouched(self):
+        result = assess_already_landed_report_state(
+            "Review decision: APPROVED\nMerge result: MERGED\n",
+            gate_fired=False)
+        self.assertTrue(result["complete"])
+
+
 class TestFullSuiteFailureApprovalGate(unittest.TestCase):
     """Issue #323: full-suite failure blocks approval without baseline proof."""