Gitea-Tools/tests/test_permission_reports.py

"""Tests for structured permission failure reports on gated tools (#142).

A blocked gated call must explain itself: which operation was requested,
what permission is missing, which profile/identity is active, whether any
configured profile could perform it, whether runtime switching is
supported, whether a different MCP namespace/session is required, and the
exact safe next action. Fail-closed behavior is unchanged — the report
adds information to denials, never capability. All synthetic; no network.
"""
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch

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

import gitea_config
import mcp_server

CONFIG = {
    "version": 2,
    "contexts": {
        "ctx": {
            "enabled": True,
            "gitea": {
                "enabled": True,
                "base_url": "https://gitea.example.com"
            }
        }
    },
    "profiles": {
        "author-profile": {
            "enabled": True,
            "context": "ctx",
            "role": "author",
            "username": "author-user",
            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
            "allowed_operations": [
                "gitea.read", "gitea.pr.create", "gitea.branch.push"],
            "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
            "execution_profile": "author-profile"
        },
        "reviewer-profile": {
            "enabled": True,
            "context": "ctx",
            "role": "reviewer",
            "username": "reviewer-user",
            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
            "allowed_operations": [
                "gitea.read", "gitea.pr.review", "gitea.pr.approve",
                "gitea.pr.merge", "gitea.issue.comment"],
            "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
            "execution_profile": "reviewer-profile"
        }
    },
    "rules": {
        "allow_runtime_switching": False
    }
}

CONFIG_SWITCHING = {**CONFIG, "rules": {"allow_runtime_switching": True}}

PR_PAYLOAD = {
    "user": {"login": "someone-else"},
    "state": "open",
    "head": {"sha": "abc123"},
    "mergeable": True,
}

REPORT_FIELDS = (
    "requested_operation",
    "missing_permission",
    "required_permission",
    "active_profile",
    "active_identity",
    "active_allowed_operations",
    "matching_configured_profiles",
    "runtime_switching_supported",
    "different_mcp_namespace_required",
    "exact_safe_next_action",
)


class PermissionReportBase(unittest.TestCase):
    def setUp(self):
        self._remotes = patch.dict(mcp_server.REMOTES, {
            "prgs": {"host": "gitea.example.com", "org": "Example-Org",
                     "repo": "Example-Repo"},
        })
        self._remotes.start()
        mcp_server._IDENTITY_CACHE.clear()
        gitea_config._active_profile_override = None
        self._dir = tempfile.TemporaryDirectory()
        self.config_path = os.path.join(self._dir.name, "profiles.json")
        self._write_config(CONFIG)

    def tearDown(self):
        self._remotes.stop()
        mcp_server._IDENTITY_CACHE.clear()
        mcp_server.review_workflow_load.clear_review_workflow_load()
        gitea_config._active_profile_override = None
        self._dir.cleanup()

    def _write_config(self, obj):
        with open(self.config_path, "w", encoding="utf-8") as fh:
            fh.write(json.dumps(obj))

    def _env(self, profile="author-profile"):
        return {
            "GITEA_MCP_CONFIG": self.config_path,
            "GITEA_MCP_PROFILE": profile,
            "GITEA_TOKEN_AUTHOR": "author-pass",
            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
        }

    def _assert_report_shape(self, report):
        for field in REPORT_FIELDS:
            self.assertIn(field, report, f"report missing {field!r}")

    def _assert_report_safe(self, report):
        blob = json.dumps(report)
        for banned in ("author-pass", "reviewer-pass", "https://", "http://",
                       "keychain", "Authorization"):
            self.assertNotIn(banned, blob, f"report leaks {banned!r}")


class TestIssueCommentDenialReport(PermissionReportBase):
    def test_missing_issue_comment_permission_gets_report(self):
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_create_issue_comment(
                issue_number=7, body="hello", remote="prgs")
        self.assertFalse(res["success"])
        self.assertFalse(res["performed"])
        self.assertTrue(res["reasons"])  # fail-closed reasons preserved
        report = res["permission_report"]
        self._assert_report_shape(report)
        self._assert_report_safe(report)
        self.assertEqual(report["missing_permission"], "gitea.issue.comment")
        self.assertEqual(report["active_profile"], "author-profile")
        self.assertIn("gitea.read", report["active_allowed_operations"])
        self.assertEqual(
            report["matching_configured_profiles"], ["reviewer-profile"])
        self.assertFalse(report["runtime_switching_supported"])
        self.assertTrue(report["different_mcp_namespace_required"])
        self.assertTrue(report["exact_safe_next_action"])

    def test_empty_body_alone_is_not_a_permission_failure(self):
        # reviewer-profile HAS gitea.issue.comment; an empty body must not
        # produce a permission report (it is an input error, not a denial).
        with patch.dict(os.environ, self._env("reviewer-profile")):
            res = mcp_server.gitea_create_issue_comment(
                issue_number=7, body="   ", remote="prgs")
        self.assertFalse(res["success"])
        self.assertNotIn("permission_report", res)

    def test_switching_enabled_changes_guidance(self):
        self._write_config(CONFIG_SWITCHING)
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_create_issue_comment(
                issue_number=7, body="hello", remote="prgs")
        report = res["permission_report"]
        self.assertTrue(report["runtime_switching_supported"])
        self.assertFalse(report["different_mcp_namespace_required"])
        self.assertIn("gitea_activate_profile",
                      report["exact_safe_next_action"])


class TestEligibilityDenialReport(PermissionReportBase):
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_author_attempting_review_gets_report(self, _auth, mock_api):
        def fake_api(method, url, auth, payload=None):
            if url.endswith("/user"):
                return {"login": "author-user"}
            return PR_PAYLOAD
        mock_api.side_effect = fake_api
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_check_pr_eligibility(
                pr_number=42, action="approve", remote="prgs")
        self.assertFalse(res["eligible"])
        report = res["permission_report"]
        self._assert_report_shape(report)
        self._assert_report_safe(report)
        self.assertEqual(report["missing_permission"], "gitea.pr.approve")
        self.assertEqual(report["active_profile"], "author-profile")
        # "active identity if resolved": the #164 fail-fast path performs no
        # identity lookup on a pure permission block, so None is acceptable.
        self.assertIn(report["active_identity"], (None, "author-user"))
        self.assertEqual(
            report["matching_configured_profiles"], ["reviewer-profile"])
        self.assertTrue(report["different_mcp_namespace_required"])

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_author_attempting_merge_gets_report(self, _auth, mock_api):
        def fake_api(method, url, auth, payload=None):
            if url.endswith("/user"):
                return {"login": "author-user"}
            return PR_PAYLOAD
        mock_api.side_effect = fake_api
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_check_pr_eligibility(
                pr_number=42, action="merge", remote="prgs")
        self.assertFalse(res["eligible"])
        report = res["permission_report"]
        self.assertEqual(report["missing_permission"], "gitea.pr.merge")
        self._assert_report_safe(report)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_eligible_call_has_no_report(self, _auth, mock_api):
        def fake_api(method, url, auth, payload=None):
            if url.endswith("/user"):
                return {"login": "reviewer-user"}
            return PR_PAYLOAD
        mock_api.side_effect = fake_api
        with patch.dict(os.environ, self._env("reviewer-profile")):
            res = mcp_server.gitea_check_pr_eligibility(
                pr_number=42, action="approve", remote="prgs")
        self.assertTrue(res["eligible"])
        self.assertNotIn("permission_report", res)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_review_tool_denial_propagates_report(self, _auth, mock_api):
        def fake_api(method, url, auth, payload=None):
            if url.endswith("/user"):
                return {"login": "author-user"}
            return PR_PAYLOAD
        mock_api.side_effect = fake_api
        mcp_server.init_review_decision_lock("prgs", "review_pr")
        from tests.test_mcp_server import _seed_ready_review_decision

        _seed_ready_review_decision(42, "approve", remote="prgs")
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_submit_pr_review(
                pr_number=42, action="approve", body="lgtm", remote="prgs",
                final_review_decision_ready=True)
        self.assertFalse(res["performed"])
        self.assertIn("permission_report", res)
        self.assertEqual(res["permission_report"]["missing_permission"],
                         "gitea.pr.approve")

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_merge_tool_denial_propagates_report(self, _auth, mock_api):
        def fake_api(method, url, auth, payload=None):
            if url.endswith("/user"):
                return {"login": "author-user"}
            return PR_PAYLOAD
        mock_api.side_effect = fake_api
        mcp_server.init_review_decision_lock("prgs", "review_pr")
        mcp_server.gitea_load_review_workflow()
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_merge_pr(
                pr_number=42, confirmation="MERGE PR 42",
                expected_head_sha="abc123", remote="prgs")
        self.assertFalse(res["performed"])
        self.assertIn("permission_report", res)
        self.assertEqual(res["permission_report"]["missing_permission"],
                         "gitea.pr.merge")


class TestReviewCommentPathUsesCanonicalOp(PermissionReportBase):
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_review_comment_denial_names_pr_review_op(self, _auth, mock_api):
        # gitea_submit_pr_review(action="comment") maps to the 'review'
        # eligibility action; the report must name the canonical
        # gitea.pr.review op (never a phantom 'gitea.review'), classify the
        # role as reviewer, and find the configured reviewer profile.
        def fake_api(method, url, auth, payload=None):
            if url.endswith("/user"):
                return {"login": "author-user"}
            return PR_PAYLOAD
        mock_api.side_effect = fake_api
        mcp_server.init_review_decision_lock("prgs", "review_pr")
        from tests.test_mcp_server import _seed_ready_review_decision

        _seed_ready_review_decision(42, "comment", remote="prgs")
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_submit_pr_review(
                pr_number=42, action="comment", body="finding", remote="prgs",
                final_review_decision_ready=True)
        self.assertFalse(res["performed"])
        report = res["permission_report"]
        self._assert_report_safe(report)
        self.assertEqual(report["missing_permission"], "gitea.pr.review")
        self.assertEqual(
            report["matching_configured_profiles"], ["reviewer-profile"])
        self.assertIn("reviewer", report["exact_safe_next_action"].lower())


class TestMatchingScanFailsClosed(PermissionReportBase):
    def test_profile_with_bad_forbidden_entry_is_not_reported_matching(self):
        # A profile whose forbidden list cannot be normalized would be
        # refused by the real gate (invalid-forbidden-entry, fail closed);
        # the report must not advertise it as a match.
        config = json.loads(json.dumps(CONFIG))
        config["profiles"]["reviewer-profile"]["forbidden_operations"] = [
            "definitely.not.a.real.op.entry"]
        self._write_config(config)
        with patch.dict(os.environ, self._env("author-profile")):
            report = mcp_server._permission_block_report("gitea.pr.approve")
        self.assertEqual(report["matching_configured_profiles"], [])


class TestReviewerAttemptingAuthoring(PermissionReportBase):
    def test_reviewer_blocked_from_authoring_op_names_author_namespace(self):
        with patch.dict(os.environ, self._env("reviewer-profile")):
            report = mcp_server._permission_block_report("gitea.branch.push")
        self._assert_report_shape(report)
        self._assert_report_safe(report)
        self.assertEqual(report["missing_permission"], "gitea.branch.push")
        self.assertEqual(report["active_profile"], "reviewer-profile")
        self.assertEqual(
            report["matching_configured_profiles"], ["author-profile"])
        self.assertTrue(report["different_mcp_namespace_required"])
        self.assertIn("author", report["exact_safe_next_action"].lower())


class TestUnknownProfileFailsClosed(PermissionReportBase):
    def test_unknown_profile_denial_still_fails_closed_with_safe_report(self):
        with patch.dict(os.environ, self._env("no-such-profile")):
            res = mcp_server.gitea_create_issue_comment(
                issue_number=7, body="hello", remote="prgs")
        self.assertFalse(res["success"])
        self.assertFalse(res["performed"])
        self.assertTrue(res["reasons"])
        report = res["permission_report"]
        self._assert_report_shape(report)
        self._assert_report_safe(report)
        self.assertIsNone(report["active_profile"])
        self.assertTrue(report["different_mcp_namespace_required"])
        self.assertIn("operator", report["exact_safe_next_action"].lower())


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