import os import sys import unittest import re from unittest.mock import patch, MagicMock sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) from mcp_server import ( gitea_list_prs, gitea_review_pr, gitea_check_pr_eligibility, ) import gitea_config FAKE_AUTH = "Basic dGVzdDp0ZXN0" class TestPRQueueInventory(unittest.TestCase): @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") def test_author_profile_can_list_open_prs(self, mock_get_profile, _auth, mock_get_all): mock_get_profile.return_value = { "profile_name": "gitea-author", "allowed_operations": ["read", "gitea.pr.comment"], "forbidden_operations": [], "base_url": None, } mock_get_all.return_value = [ {"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True} ] result = gitea_list_prs(state="open", remote="prgs") self.assertEqual(len(result), 1) self.assertEqual(result[0]["number"], 1) @patch("mcp_server.api_get_all") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") @patch("gitea_config.is_runtime_switching_enabled", return_value=True) def test_author_profile_can_produce_pending_reviewer_queue_report(self, mock_switch, mock_get_profile, _auth, mock_api, mock_get_all): mock_get_profile.return_value = { "profile_name": "gitea-author", "allowed_operations": ["read"], "forbidden_operations": [], "base_url": None, } mock_get_all.return_value = [ { "number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}, "labels": [{"name": "bug"}], "body": "Fixes #10" }, { "number": 2, "title": "PR 2", "state": "open", "head": {"ref": "branch2", "sha": "abc2"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "other_user"}, "labels": [], "body": "" } ] mock_api.return_value = {"login": "jcwalker3"} # whoami result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs") self.assertFalse(result["success"]) self.assertIn("=== PR Queue Inventory ===", result["message"]) self.assertIn("Repository:", result["message"]) self.assertIn("Open PRs found: 2", result["message"]) self.assertIn("Review/merge blocked (Self-author). Requires a genuine non-author reviewer", result["message"]) self.assertIn("Review/merge blocked (Current profile is Author)", result["message"]) self.assertIn("gitea_activate_profile", result["message"]) @patch("mcp_server.api_get_all") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") def test_author_profile_never_calls_mutation_tools_during_inventory(self, mock_get_profile, _auth, mock_api, mock_get_all): mock_get_profile.return_value = { "profile_name": "gitea-author", "allowed_operations": ["read"], "forbidden_operations": [], "base_url": None, } mock_get_all.return_value = [] mock_api.return_value = {"login": "jcwalker3"} gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs") for call in mock_api.call_args_list: method = call.args[0] self.assertEqual(method, "GET") @patch("mcp_server.api_get_all") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") def test_reviewer_profile_can_proceed_from_inventory_to_review_gates(self, mock_get_profile, _auth, mock_api, mock_get_all): mock_get_profile.return_value = { "profile_name": "gitea-reviewer", "allowed_operations": ["read", "approve", "review"], "forbidden_operations": [], "base_url": None, } mock_get_all.return_value = [ {"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "other_user"}} ] # mock_api: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility), 4) /pulls/1/reviews (POST review) mock_api.side_effect = [ {"login": "reviewer1"}, # inventory whoami {"login": "reviewer1"}, # eligibility whoami {"state": "open", "head": {"sha": "abc1"}, "mergeable": True, "user": {"login": "other_user"}}, # eligibility PR details {"id": 100} # POST review ] result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs") self.assertTrue(result["success"]) self.assertIn("=== PR Queue Inventory ===", result["message"]) self.assertIn("Repository:", result["message"]) self.assertIn("Successfully submitted review", result["message"]) @patch("mcp_server.api_get_all") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") @patch("gitea_config.is_runtime_switching_enabled", return_value=False) def test_static_runtime_rejects_or_hides_profile_activation(self, mock_switch, mock_get_profile, _auth, mock_api, mock_get_all): mock_get_profile.return_value = { "profile_name": "gitea-author", "allowed_operations": ["read"], "forbidden_operations": [], "base_url": None, } mock_get_all.return_value = [] mock_api.return_value = {"login": "jcwalker3"} result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs") self.assertFalse(result["success"]) self.assertIn("Repository:", result["message"]) self.assertNotIn("gitea_activate_profile", result["message"]) self.assertIn("Switch to the reviewer MCP session", result["message"]) @patch("mcp_server.get_profile") def test_stopped_before_listing_is_not_reported_as_no_prs(self, mock_get_profile): mock_get_profile.return_value = { "profile_name": "gitea-restricted", "allowed_operations": [], "forbidden_operations": [], "base_url": None, } result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs") self.assertFalse(result["success"]) self.assertIn("PR inventory not attempted", result["message"]) self.assertNotIn("Open PRs found: 0", result["message"]) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") def test_reviewer_eligibility_check_not_run_for_author_profile(self, mock_get_profile, _auth, mock_api): mock_get_profile.return_value = { "profile_name": "gitea-author", "allowed_operations": ["read"], "forbidden_operations": [], "base_url": None, } result = gitea_check_pr_eligibility(pr_number=1, action="approve", remote="prgs") self.assertFalse(result["eligible"]) self.assertEqual(mock_api.call_count, 0) @patch("mcp_server.api_get_all") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") def test_inventory_failure_output_is_redacted_no_raw_exception_leak(self, mock_get_profile, _auth, mock_api, mock_get_all): """Inventory failure path must use redaction and not leak raw exception text.""" mock_get_profile.return_value = { "profile_name": "gitea-author", "allowed_operations": ["read"], "forbidden_operations": [], "base_url": None, } # Exception contains something that should be redacted or not leaked raw mock_get_all.side_effect = Exception("connection error token supersecrettoken123 at https://internal.example/repo") mock_api.return_value = {"login": "jcwalker3"} result = gitea_review_pr(pr_number=42, event="APPROVE", remote="prgs") self.assertFalse(result["success"]) msg = result.get("message", "") self.assertIn("=== PR Queue Inventory ===", msg) self.assertIn("Repository:", msg) self.assertIn("PR inventory failed:", msg) # raw secret and potentially sensitive details must not appear self.assertNotIn("supersecrettoken123", msg) self.assertNotIn("token=supersecrettoken123", msg) # uses project's redaction pattern (token prefix -> [REDACTED]) # even if url leaks, the exception is redacted per pattern @patch("mcp_server.api_get_all") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") def test_author_profiles_can_perform_read_only_pr_inventory(self, mock_get_profile, _auth, mock_api, mock_get_all): """Author profiles with read can perform read-only PR inventory (report produced, no mutations).""" mock_get_profile.return_value = { "profile_name": "gitea-author", "allowed_operations": ["read"], "forbidden_operations": [], "base_url": None, } mock_get_all.return_value = [ {"number": 7, "title": "Some PR", "state": "open", "head": {"ref": "f", "sha": "def"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "someone"}, "labels": [], "body": ""} ] mock_api.return_value = {"login": "jcwalker3"} result = gitea_review_pr(pr_number=7, event="APPROVE", remote="prgs") self.assertFalse(result["success"]) msg = result["message"] self.assertIn("=== PR Queue Inventory ===", msg) self.assertIn("Repository:", msg) self.assertIn("Open PRs found: 1", msg) # no mutation calls (covered by other test but reconfirm here) for call in mock_api.call_args_list: self.assertEqual(call.args[0], "GET") @patch("mcp_server.api_get_all") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all): """Author profiles still cannot approve, request_changes, merge, or bypass gates even with inventory.""" for event in ["APPROVE", "REQUEST_CHANGES"]: mock_get_profile.return_value = { "profile_name": "gitea-author", "allowed_operations": ["read"], "forbidden_operations": [], "base_url": None, } mock_get_all.return_value = [] mock_api.return_value = {"login": "jcwalker3"} result = gitea_review_pr(pr_number=1, event=event, remote="prgs") self.assertFalse(result["success"]) self.assertIn("Review/Merge Blocked", result["message"]) self.assertIn("does not have review or merge permissions", result["message"]) # ensure no mutation attempted mutation_calls = [c for c in mock_api.call_args_list if c.args[0] != "GET"] self.assertEqual(len(mutation_calls), 0) # quick check merge path is still hard-gated (via eligibility in submit, but author profile blocks early) # we already block before calling submit for non-reviewer