Gitea-Tools/tests/test_reconciler_close_gate.py

"""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()