Merge pull request 'feat: dedicated reconciler capability for closing landed PRs (Closes #309)' (#379) from feat/issue-309-reconciler-close-capability into master

tests/test_reconciler_close_gate.py

@@ -0,0 +1,188 @@
+"""Issue #309: dedicated reconciler capability to close already-landed PRs.
+
+The reconciler path must prove exact ``gitea.pr.close`` capability, must
+never imply review/approve/request-changes/merge capability, and may close
+a PR only after full already-landed proof.
+"""
+
+import sys
+import unittest
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import role_session_router  # noqa: E402
+import task_capability_map  # noqa: E402
+from review_proofs import assess_reconciler_close_gate  # noqa: E402
+
+PINNED = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
+OTHER = "a4060c5de00f2b1c9e88f4f6f0f3f9a7b2c1d0e9"
+
+
+class TestReconcilerCapabilityMap(unittest.TestCase):
+    def test_reconcile_close_pr_task_maps_to_pr_close(self):
+        self.assertEqual(
+            task_capability_map.required_permission(
+                "reconcile_close_landed_pr"),
+            "gitea.pr.close",
+        )
+        self.assertEqual(
+            task_capability_map.required_role("reconcile_close_landed_pr"),
+            "reconciler",
+        )
+
+    def test_reconcile_close_issue_task_maps_to_issue_close(self):
+        self.assertEqual(
+            task_capability_map.required_permission(
+                "reconcile_close_landed_issue"),
+            "gitea.issue.close",
+        )
+        self.assertEqual(
+            task_capability_map.required_role(
+                "reconcile_close_landed_issue"),
+            "reconciler",
+        )
+
+    def test_reconciler_tasks_do_not_grant_review_permissions(self):
+        for task in ("reconcile_close_landed_pr",
+                     "reconcile_close_landed_issue"):
+            permission = task_capability_map.required_permission(task)
+            self.assertNotIn("review", permission)
+            self.assertNotIn("approve", permission)
+            self.assertNotIn("merge", permission)
+            self.assertNotIn("request_changes", permission)
+
+
+class TestReconcilerRouting(unittest.TestCase):
+    def test_reconciler_task_requires_reconciler_role(self):
+        self.assertEqual(
+            role_session_router.required_role_for_task(
+                "reconcile_close_landed_pr"),
+            "reconciler",
+        )
+
+    def test_reconciler_task_blocked_in_author_session(self):
+        result = role_session_router.route_task_session(
+            "reconcile_close_landed_pr",
+            active_profile="prgs-author",
+            active_role_kind="author",
+            allowed_in_current_session=False,
+        )
+        self.assertEqual(result["route_result"], "wrong_role_stop")
+        self.assertFalse(result["downstream_allowed"])
+
+    def test_reconciler_task_allowed_in_matching_session(self):
+        result = role_session_router.route_task_session(
+            "reconcile_close_landed_pr",
+            active_profile="prgs-reconciler",
+            active_role_kind="reconciler",
+            allowed_in_current_session=True,
+        )
+        self.assertEqual(result["route_result"], "allowed_current_session")
+        self.assertTrue(result["downstream_allowed"])
+
+
+class TestReconcilerCloseGate(unittest.TestCase):
+    def _gate(self, **overrides):
+        kwargs = {
+            "pr_number": 278,
+            "pr_state": "open",
+            "candidate_head_sha": PINNED,
+            "live_head_sha": PINNED,
+            "target_branch": "master",
+            "target_branch_sha": OTHER,
+            "head_is_ancestor_of_target": True,
+            "close_pr_capability": True,
+            "close_issue_capability": False,
+            "linked_issue_state": "closed",
+        }
+        kwargs.update(overrides)
+        return assess_reconciler_close_gate(**kwargs)
+
+    def test_already_landed_open_pr_close_allowed(self):
+        result = self._gate()
+        self.assertEqual(result["outcome"], "CLOSE_ALLOWED")
+        self.assertTrue(result["pr_close_allowed"])
+
+    def test_not_landed_pr_cannot_be_closed(self):
+        result = self._gate(head_is_ancestor_of_target=False)
+        self.assertEqual(result["outcome"], "NOT_LANDED_CLOSE_BLOCKED")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_unchecked_ancestry_blocks_close(self):
+        result = self._gate(head_is_ancestor_of_target=None)
+        self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_stale_target_branch_sha_blocks_close(self):
+        result = self._gate(target_branch_sha=None)
+        self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_closed_pr_state_blocks_close(self):
+        result = self._gate(pr_state="closed")
+        self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_changed_head_blocks_close(self):
+        result = self._gate(live_head_sha=OTHER)
+        self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_missing_close_capability_produces_recovery_handoff(self):
+        result = self._gate(close_pr_capability=False)
+        self.assertEqual(result["outcome"], "RECOVERY_HANDOFF_REQUIRED")
+        self.assertFalse(result["pr_close_allowed"])
+        self.assertTrue(any(
+            "gitea.pr.close" in reason for reason in result["reasons"]
+        ))
+
+    def test_linked_issue_already_closed_skips_issue_close(self):
+        result = self._gate(linked_issue_state="closed",
+                            close_issue_capability=True)
+        self.assertFalse(result["issue_close_allowed"])
+
+    def test_linked_open_issue_close_requires_capability(self):
+        allowed = self._gate(
+            linked_issue_state="open",
+            issue_resolved_by_landing=True,
+            close_issue_capability=True,
+        )
+        self.assertTrue(allowed["issue_close_allowed"])
+
+        blocked = self._gate(
+            linked_issue_state="open",
+            issue_resolved_by_landing=True,
+            close_issue_capability=False,
+        )
+        self.assertFalse(blocked["issue_close_allowed"])
+
+    def test_gate_never_allows_review_mutations(self):
+        for result in (
+            self._gate(),
+            self._gate(head_is_ancestor_of_target=False),
+            self._gate(close_pr_capability=False),
+        ):
+            self.assertFalse(result["review_allowed"])
+            self.assertFalse(result["approve_allowed"])
+            self.assertFalse(result["request_changes_allowed"])
+            self.assertFalse(result["merge_allowed"])
+
+    def test_report_fields_listed_for_close(self):
+        result = self._gate()
+        for field in (
+            "identity/profile",
+            "close capability proof",
+            "PR live state",
+            "candidate head SHA",
+            "target branch SHA",
+            "ancestor proof",
+            "linked issue status",
+            "PR close result",
+            "issue close result",
+            "no review/merge confirmation",
+        ):
+            self.assertIn(field, result["required_report_fields"])
+
+
+if __name__ == "__main__":
+    unittest.main()