Gitea-Tools/tests/test_reconcile_already_landed_pr_tool.py

"""Tests for gitea_reconcile_already_landed_pr MCP tool (#310)."""
import sys
import unittest
from unittest.mock import patch

sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))

import mcp_server
from mcp_server import gitea_reconcile_already_landed_pr

RECONCILER_PROFILE = {
    "profile_name": "prgs-reconciler",
    "allowed_operations": [
        "gitea.read",
        "gitea.pr.comment",
        "gitea.issue.comment",
        "gitea.issue.close",
        "gitea.pr.close",
    ],
    "forbidden_operations": [
        "gitea.pr.approve",
        "gitea.pr.merge",
        "gitea.pr.create",
        "gitea.branch.push",
    ],
    "audit_label": "prgs-reconciler",
}

AUTHOR_PROFILE = {
    "profile_name": "prgs-author",
    "allowed_operations": [
        "gitea.read",
        "gitea.pr.create",
        "gitea.issue.comment",
    ],
    "forbidden_operations": ["gitea.pr.close", "gitea.pr.approve", "gitea.pr.merge"],
    "audit_label": "prgs-author",
}

OPEN_LANDED_PR = {
    "number": 278,
    "title": "Already landed (Closes #263)",
    "body": "",
    "state": "open",
    "head": {"ref": "feat/issue-263", "sha": "2a544b78d1371925761d16f1c451bb3b2984470e"},
    "base": {"ref": "master"},
}


class TestReconcileAlreadyLandedPrTool(unittest.TestCase):
    def setUp(self):
        self.mock_api = patch("mcp_server.api_request").start()
        self.mock_auth = patch(
            "mcp_server.get_auth_header", return_value="token test"
        ).start()
        patch("mcp_server.verify_preflight_purity").start()
        patch("gitea_audit.audit_enabled", return_value=False).start()
        mcp_server._IDENTITY_CACHE.clear()

    def tearDown(self):
        patch.stopall()
        mcp_server._IDENTITY_CACHE.clear()

    def _set_profile(self, profile):
        patch("mcp_server.get_profile", return_value=profile).start()

    def test_close_blocked_without_pr_close_capability(self):
        self._set_profile(AUTHOR_PROFILE)
        res = gitea_reconcile_already_landed_pr(
            pr_number=278,
            close_pr=True,
            remote="prgs",
        )
        self.assertFalse(res["success"])
        self.assertFalse(res["performed"])
        self.assertEqual(res["required_permission"], "gitea.pr.close")
        self.mock_api.assert_not_called()

    def test_not_landed_pr_close_denied_after_live_fetch(self):
        self._set_profile(RECONCILER_PROFILE)
        self.mock_api.return_value = OPEN_LANDED_PR
        with patch(
            "mcp_server.already_landed_reconcile.assess_already_landed_reconciliation",
            return_value={
                "pr_state": "open",
                "candidate_head_sha": OPEN_LANDED_PR["head"]["sha"],
                "target_branch": "master",
                "target_branch_sha": "deadbeef",
                "ancestor_proof": False,
                "eligibility_class": "NOT_ALREADY_LANDED",
                "linked_issue": 263,
                "close_allowed": False,
                "git_ref_mutations": ["git fetch prgs master"],
                "reasons": ["not ancestor"],
            },
        ):
            res = gitea_reconcile_already_landed_pr(
                pr_number=278,
                close_pr=True,
                remote="prgs",
            )
        self.assertFalse(res["success"])
        self.assertFalse(res.get("pr_closed"))
        patch_calls = [
            c for c in self.mock_api.call_args_list if c.args[0] == "PATCH"
        ]
        self.assertEqual(patch_calls, [])

    def test_already_landed_close_allowed_with_capability(self):
        self._set_profile(RECONCILER_PROFILE)

        def api_side_effect(method, url, auth, payload=None):
            if method == "GET" and "/pulls/278" in url:
                return dict(OPEN_LANDED_PR)
            if method == "PATCH" and "/pulls/278" in url:
                closed = dict(OPEN_LANDED_PR)
                closed["state"] = "closed"
                return closed
            return {}

        self.mock_api.side_effect = api_side_effect
        with patch(
            "mcp_server.already_landed_reconcile.assess_already_landed_reconciliation",
            return_value={
                "pr_state": "open",
                "candidate_head_sha": OPEN_LANDED_PR["head"]["sha"],
                "target_branch": "master",
                "target_branch_sha": "e017d80256217f17b0f421cd051cfa5cbcf5ae0f",
                "ancestor_proof": True,
                "eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED",
                "linked_issue": 263,
                "close_allowed": True,
                "git_ref_mutations": ["git fetch prgs master"],
                "reasons": [],
            },
        ):
            res = gitea_reconcile_already_landed_pr(
                pr_number=278,
                close_pr=True,
                remote="prgs",
            )
        self.assertTrue(res["success"])
        self.assertTrue(res["performed"])
        self.assertTrue(res["pr_closed"])
        self.assertEqual(res["eligibility_class"], "ALREADY_LANDED_RECONCILE_REQUIRED")


if __name__ == "__main__":
    unittest.main()