Gitea-Tools/tests/test_close_pr_capability.py

"""Tests for the gated PR close path (Issue #216).

``gitea_edit_pr(state='closed')`` requires the distinct ``gitea.pr.close``
capability: without it the close fails closed (no auth lookup, no API call,
structured permission report). With it — the explicit operator-directed
contaminated-PR closure path — the close proceeds and is audited as a
distinct ``close_pr`` action carrying the required capability, so final
reports can prove exactly which mutation capability was exercised.

Ordinary edits (title/body/base) and reopening never require the close
capability: PR comment, PR edit, and PR close remain distinct capabilities.
"""
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_edit_pr

FAKE_AUTH = "token fake"

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

AUTHOR_WITH_CLOSE = {
    "profile_name": "prgs-author-closer",
    "allowed_operations": AUTHOR_NO_CLOSE["allowed_operations"] + ["gitea.pr.close"],
    "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
    "audit_label": "prgs-author-closer",
}


class TestEditPrCloseGate(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=FAKE_AUTH).start()
        # Deterministic permission reports: no real operator config, no switching.
        patch("gitea_config.load_config", return_value={}).start()
        patch("gitea_config.is_runtime_switching_enabled", return_value=False).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_close_capability(self):
        # Author profile without gitea.pr.close: the broad edit path must not
        # be usable as an untracked close fallback.
        self._set_profile(AUTHOR_NO_CLOSE)
        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
        self.assertFalse(res["success"])
        self.assertFalse(res["performed"])
        self.assertEqual(res["requested_state"], "closed")
        self.assertEqual(res["required_permission"], "gitea.pr.close")
        self.assertTrue(res["reasons"])
        report = res["permission_report"]
        self.assertEqual(report["required_permission"], "gitea.pr.close")
        self.assertEqual(report["active_profile"], "prgs-author")
        self.mock_api.assert_not_called()
        self.mock_auth.assert_not_called()

    def test_close_blocked_when_close_forbidden(self):
        profile = dict(AUTHOR_WITH_CLOSE)
        profile["forbidden_operations"] = ["gitea.pr.close"]
        self._set_profile(profile)
        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
        self.assertFalse(res["success"])
        self.assertIn("profile forbids 'gitea.pr.close'", res["reasons"])
        self.mock_api.assert_not_called()

    def test_close_blocked_when_profile_unresolvable(self):
        patch("mcp_server.get_profile", side_effect=RuntimeError("no profile")).start()
        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
        self.assertFalse(res["success"])
        self.assertTrue(any("fail closed" in r for r in res["reasons"]))
        self.mock_api.assert_not_called()

    def test_operator_directed_close_allowed_and_audited(self):
        # Explicit operator-directed contaminated-PR closure: the operator
        # granted gitea.pr.close, so the close proceeds and the audit trail
        # records a distinct close_pr action with the capability proof.
        self._set_profile(AUTHOR_WITH_CLOSE)
        patch("gitea_audit.audit_enabled", return_value=True).start()
        mock_write = patch("gitea_audit.write_event").start()

        def api_side_effect(method, url, auth, payload=None):
            if method == "GET" and "/user" in url:
                return {"login": "jcwalker3"}
            if method == "PATCH" and "pulls/205" in url:
                self.assertEqual(payload["state"], "closed")
                return {
                    "number": 205,
                    "title": "Contaminated PR",
                    "state": "closed",
                    "html_url": "url",
                    "body": "No issue link",
                    "head": {"ref": "feat/invalid-provenance"},
                }
            return {}
        self.mock_api.side_effect = api_side_effect

        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
        self.assertTrue(res["success"])
        self.assertEqual(res["state"], "closed")
        mock_write.assert_called()
        event = mock_write.call_args[0][0]
        self.assertEqual(event["action"], "close_pr")
        self.assertEqual(
            event["request_metadata"]["required_permission"], "gitea.pr.close")

    def test_title_edit_needs_no_close_capability(self):
        # PR edit and PR close are distinct capabilities: retitling stays on
        # the ordinary edit path.
        self._set_profile(AUTHOR_NO_CLOSE)
        self.mock_api.return_value = {
            "number": 7, "title": "Renamed", "state": "open",
            "body": "", "html_url": "u"}
        res = gitea_edit_pr(pr_number=7, title="Renamed", remote="prgs")
        self.assertTrue(res["success"])

    def test_reopen_needs_no_close_capability(self):
        self._set_profile(AUTHOR_NO_CLOSE)
        self.mock_api.return_value = {
            "number": 7, "title": "T", "state": "open",
            "body": "", "html_url": "u"}
        res = gitea_edit_pr(pr_number=7, state="open", remote="prgs")
        self.assertTrue(res["success"])

    def test_invalid_state_fails_closed_before_auth(self):
        # A case-variant state can neither bypass the close gate nor reach
        # the API: it is rejected as pure validation, before auth.
        self._set_profile(AUTHOR_WITH_CLOSE)
        with self.assertRaises(ValueError):
            gitea_edit_pr(pr_number=7, state="CLOSED", remote="prgs")
        self.mock_api.assert_not_called()
        self.mock_auth.assert_not_called()


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