sysadminandClaude Opus 4.8
91c67930ec
Adds canonical pr-queue-cleanup workflow, report verifier, reviewer-only task routing, and runbook guidance so operators can drain open PRs one canonical review at a time with fail-closed boundaries on batching and unauthorized merges. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
314 lines
17 KiB
Python
314 lines
17 KiB
Python
"""Tests for gitea_resolve_task_capability tool.
|
|
|
|
Covers Issue #141 requirements and #145 regression tests for permission boundaries,
|
|
self blocks, and structured guidance.
|
|
"""
|
|
import os
|
|
import sys
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
import gitea_config
|
|
import gitea_auth
|
|
import mcp_server
|
|
|
|
CONFIG_RESOLVER = {
|
|
"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.issue.create", "gitea.pr.create", "gitea.branch.push", "gitea.issue.comment"],
|
|
"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
|
|
}
|
|
}
|
|
|
|
class TestResolveTaskCapability(unittest.TestCase):
|
|
def setUp(self):
|
|
self._remotes_patch = patch.dict(mcp_server.REMOTES, {
|
|
"dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
|
|
"prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}
|
|
})
|
|
self._remotes_patch.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_RESOLVER)
|
|
|
|
def tearDown(self):
|
|
self._remotes_patch.stop()
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
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",
|
|
}
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_resolve_author_capable_task(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
|
|
self.assertEqual(res["requested_task"], "create_issue")
|
|
self.assertEqual(res["required_operation_permission"], "gitea.issue.create")
|
|
self.assertEqual(res["required_role_kind"], "author")
|
|
self.assertEqual(res["active_profile"], "author-profile")
|
|
self.assertTrue(res["allowed_in_current_session"])
|
|
self.assertIn("author-profile", res["matching_configured_profile"])
|
|
self.assertFalse(res["different_mcp_namespace_required"])
|
|
self.assertIn("ready for operations", res["exact_safe_next_action"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_resolve_reviewer_capable_task_blocked(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
|
|
self.assertEqual(res["requested_task"], "review_pr")
|
|
self.assertEqual(res["required_operation_permission"], "gitea.pr.review")
|
|
self.assertEqual(res["required_role_kind"], "reviewer")
|
|
self.assertEqual(res["active_profile"], "author-profile")
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertIn("reviewer-profile", res["matching_configured_profile"])
|
|
self.assertTrue(res["different_mcp_namespace_required"])
|
|
self.assertIn("gitea-reviewer", res["exact_safe_next_action"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
|
def test_resolve_reviewer_capable_task_allowed(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("reviewer-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
|
|
self.assertTrue(res["allowed_in_current_session"])
|
|
self.assertFalse(res["different_mcp_namespace_required"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_resolve_issue_comment_task(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(task="comment_issue", remote="prgs")
|
|
self.assertTrue(res["allowed_in_current_session"])
|
|
self.assertIn("author-profile", res["matching_configured_profile"])
|
|
self.assertIn("reviewer-profile", res["matching_configured_profile"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_resolve_mark_issue_author_profile_allowed(self, _auth, _api):
|
|
# #183: mark_issue must map to gitea.issue.comment (prgs-author allowed op).
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(task="mark_issue", remote="prgs")
|
|
self.assertEqual(res["requested_task"], "mark_issue")
|
|
self.assertEqual(res["required_operation_permission"], "gitea.issue.comment")
|
|
self.assertTrue(res["allowed_in_current_session"])
|
|
self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_resolve_lock_issue_author_profile_allowed(self, _auth, _api):
|
|
# #275: lock_issue must be an exact resolver task for claim/lock gates.
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(task="lock_issue", remote="prgs")
|
|
self.assertEqual(res["requested_task"], "lock_issue")
|
|
self.assertEqual(res["required_operation_permission"], "gitea.issue.comment")
|
|
self.assertTrue(res["allowed_in_current_session"])
|
|
self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
|
|
|
|
def test_resolve_unknown_task_fails_closed(self):
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
with self.assertRaises(ValueError):
|
|
mcp_server.gitea_resolve_task_capability(task="invalid_task_xyz", remote="prgs")
|
|
|
|
# Additional regression tests per #145 for permission boundaries and structured guidance
|
|
def test_issue_comment_does_not_imply_close(self):
|
|
# Author profile has issue.comment but not issue.close
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res_comment = mcp_server.gitea_resolve_task_capability(task="comment_issue", remote="prgs")
|
|
res_close = mcp_server.gitea_resolve_task_capability(task="close_issue", remote="prgs")
|
|
self.assertTrue(res_comment["allowed_in_current_session"])
|
|
self.assertIn("gitea.issue.comment", res_comment["active_profile_allowed_operations"])
|
|
self.assertFalse(res_close["allowed_in_current_session"])
|
|
self.assertIn("gitea.issue.close", res_close.get("required_operation_permission", ""))
|
|
|
|
def test_pr_comment_does_not_imply_issue_comment(self):
|
|
# Create a profile that has pr.comment but not issue.comment
|
|
config = dict(CONFIG_RESOLVER)
|
|
config["profiles"]["pr-only"] = {
|
|
"enabled": True,
|
|
"context": "ctx",
|
|
"role": "author",
|
|
"username": "pr-user",
|
|
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
|
"allowed_operations": ["gitea.read", "gitea.pr.comment"],
|
|
"forbidden_operations": ["gitea.issue.comment"],
|
|
"execution_profile": "pr-only"
|
|
}
|
|
self._write_config(config)
|
|
with patch.dict(os.environ, {**self._env("pr-only"), "GITEA_MCP_PROFILE": "pr-only"}):
|
|
res_pr = mcp_server.gitea_resolve_task_capability(task="comment_pr", remote="prgs")
|
|
res_issue = mcp_server.gitea_resolve_task_capability(task="comment_issue", remote="prgs")
|
|
self.assertTrue(res_pr["allowed_in_current_session"])
|
|
self.assertFalse(res_issue["allowed_in_current_session"])
|
|
self.assertIn("gitea.pr.comment", res_pr["active_profile_allowed_operations"])
|
|
self.assertNotIn("gitea.issue.comment", res_pr["active_profile_allowed_operations"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
|
def test_reviewer_cannot_push_branch(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("reviewer-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(task="push_branch", remote="prgs")
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertIn("author-profile", res["matching_configured_profile"])
|
|
self.assertTrue(res["different_mcp_namespace_required"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_structured_guidance_for_missing_permission(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(task="merge_pr", remote="prgs")
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertIn("reviewer", res["exact_safe_next_action"].lower())
|
|
self.assertEqual(res["required_operation_permission"], "gitea.pr.merge")
|
|
self.assertEqual(res["required_role_kind"], "reviewer")
|
|
|
|
@patch("mcp_server.api_request")
|
|
def test_self_review_blocked_structured(self, mock_api):
|
|
# Test that self-approve (self-review gate) is blocked and returns structured reasons
|
|
def side_api(method, url, auth):
|
|
if "/user" in url:
|
|
return {"login": "author-user"}
|
|
if "/pulls/123" in url:
|
|
return {"user": {"login": "author-user"}, "state": "open", "head": {"sha": "abc"}, "mergeable": True}
|
|
return {}
|
|
mock_api.side_effect = side_api
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_check_pr_eligibility(pr_number=123, action="approve", remote="prgs")
|
|
self.assertFalse(res.get("eligible", True))
|
|
reasons = res.get("reasons", [])
|
|
self.assertTrue(any("author" in str(r).lower() for r in reasons) or len(reasons) > 0)
|
|
|
|
# Issue #216: close_pr is a first-class resolver task gated on gitea.pr.close.
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_resolve_close_pr_author_profile_without_close_blocked(self, _auth, _api):
|
|
# Default author profile has PR create/comment but not close: the
|
|
# close capability must never be implied by broader author ops.
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
|
|
self.assertEqual(res["requested_task"], "close_pr")
|
|
self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
|
|
self.assertEqual(res["required_role_kind"], "author")
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertTrue(res["stop_required"])
|
|
self.assertEqual(res["matching_configured_profile"], [])
|
|
self.assertTrue(res["different_mcp_namespace_required"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_resolve_close_pr_author_profile_with_close_allowed(self, _auth, _api):
|
|
# Operator-granted close capability resolves cleanly under an author profile.
|
|
config = json.loads(json.dumps(CONFIG_RESOLVER))
|
|
config["profiles"]["author-closer"] = {
|
|
"enabled": True,
|
|
"context": "ctx",
|
|
"role": "author",
|
|
"username": "author-user",
|
|
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
|
"allowed_operations": ["gitea.read", "gitea.pr.close"],
|
|
"forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
|
|
"execution_profile": "author-closer",
|
|
}
|
|
self._write_config(config)
|
|
with patch.dict(os.environ, self._env("author-closer")):
|
|
res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
|
|
self.assertTrue(res["allowed_in_current_session"])
|
|
self.assertFalse(res["stop_required"])
|
|
self.assertIn("author-closer", res["matching_configured_profile"])
|
|
self.assertIn("ready for operations", res["exact_safe_next_action"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
|
def test_resolve_close_pr_reviewer_profile_blocked(self, _auth, _api):
|
|
# close_pr is author-side: a reviewer profile must be told to stop.
|
|
with patch.dict(os.environ, self._env("reviewer-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertTrue(res["stop_required"])
|
|
self.assertEqual(res["required_role_kind"], "author")
|
|
self.assertIn("author", res["exact_safe_next_action"].lower())
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_resolve_pr_queue_cleanup_author_profile_blocked(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(
|
|
task="pr_queue_cleanup", remote="prgs"
|
|
)
|
|
self.assertEqual(res["required_role_kind"], "reviewer")
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertTrue(res["stop_required"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
|
def test_resolve_pr_queue_cleanup_reviewer_profile_allowed(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("reviewer-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(
|
|
task="pr_queue_cleanup", remote="prgs"
|
|
)
|
|
self.assertEqual(res["required_operation_permission"], "gitea.pr.review")
|
|
self.assertTrue(res["allowed_in_current_session"])
|
|
self.assertFalse(res["stop_required"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_close_pr_known_and_lookalike_tasks_still_fail_closed(self, _auth, _api):
|
|
# close_pr resolves (no Unknown-task error); near-miss spellings keep
|
|
# failing closed so no untracked close fallback can be rationalized.
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
|
|
self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
|
|
for unknown in ("close_pull_request", "close", "pr_close"):
|
|
with self.assertRaises(ValueError):
|
|
mcp_server.gitea_resolve_task_capability(task=unknown, remote="prgs")
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|