sysadmin
dfb5ebd0af
This fixes #723 Defect B where attempting to acquire a reviewer lease from a merger session caused the session role to be incorrectly stamped as 'reviewer', leading to downstream workspace binding exceptions. This commit moves the preflight capability recording to only apply when the task is allowed, and adds a try/except downstream to correctly fail-closed instead of throwing RuntimeError.
465 lines
25 KiB
Python
465 lines
25 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.issue.comment"],
|
|
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.pr.merge"],
|
|
"execution_profile": "reviewer-profile"
|
|
},
|
|
"merger-profile": {
|
|
"enabled": True,
|
|
"context": "ctx",
|
|
"role": "merger",
|
|
"username": "merger-user",
|
|
"auth": {"type": "env", "name": "GITEA_TOKEN_MERGER"},
|
|
"allowed_operations": ["gitea.read", "gitea.pr.merge", "gitea.issue.comment"],
|
|
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.pr.approve"],
|
|
"execution_profile": "merger-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",
|
|
"GITEA_TOKEN_MERGER": "merger-pass",
|
|
}
|
|
|
|
def test_diagnose_terminal_success_has_no_error(self):
|
|
res = mcp_server.gitea_diagnose_terminal(
|
|
command=[sys.executable, "-c", "raise SystemExit(0)"]
|
|
)
|
|
self.assertTrue(res["healthy"])
|
|
self.assertIsNone(res["error_type"])
|
|
self.assertEqual(res["error_msg"], "")
|
|
|
|
@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.native_mcp_preference.probe_terminal_spawn")
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_resolve_mutation_task_blocks_when_terminal_launcher_unhealthy(
|
|
self,
|
|
_auth,
|
|
_api,
|
|
mock_probe,
|
|
):
|
|
mock_probe.return_value = {
|
|
"healthy": False,
|
|
"error_type": "missing cwd",
|
|
"error_msg": "Current working directory does not exist: /missing",
|
|
"cwd": "/missing",
|
|
"command": ["git", "--version"],
|
|
"diagnostics": {"cwd_exists": False},
|
|
}
|
|
env = self._env("author-profile")
|
|
# Probe is skipped under pytest unless forced (mirrors production path).
|
|
env["GITEA_FORCE_TERMINAL_PROBE"] = "1"
|
|
with patch.dict(os.environ, env):
|
|
res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
|
|
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertTrue(res["stop_required"])
|
|
self.assertTrue(res["infra_stop"])
|
|
self.assertTrue(res["terminal_launcher_unhealthy"])
|
|
self.assertTrue(res["blocked"])
|
|
self.assertEqual(res["blocked_reason"], "BLOCKED + DIAGNOSE")
|
|
self.assertEqual(res["terminal_launcher_diagnostics"]["error_type"], "missing cwd")
|
|
self.assertIn("terminal-launcher-failure", res["exact_safe_next_action"])
|
|
self.assertIn("BLOCKED + DIAGNOSE", 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_work_issue_author_profile_allowed(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
for task in ("work_issue", "work-issue"):
|
|
res = mcp_server.gitea_resolve_task_capability(
|
|
task=task, remote="prgs"
|
|
)
|
|
self.assertEqual(res["requested_task"], task)
|
|
self.assertEqual(
|
|
res["required_operation_permission"], "gitea.pr.create"
|
|
)
|
|
self.assertEqual(res["required_role_kind"], "author")
|
|
self.assertTrue(res["allowed_in_current_session"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_resolve_work_issue_reviewer_profile_blocked(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("reviewer-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(
|
|
task="work_issue", remote="prgs"
|
|
)
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertEqual(res["required_role_kind"], "author")
|
|
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_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("merger", res["exact_safe_next_action"].lower())
|
|
self.assertEqual(res["required_operation_permission"], "gitea.pr.merge")
|
|
self.assertEqual(res["required_role_kind"], "merger")
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
|
def test_reviewer_role_with_merge_permission_still_cannot_merge(self, _auth, _api):
|
|
# #539: exact permission alone is insufficient. The active profile
|
|
# role must also be merger for merge_pr.
|
|
config = json.loads(json.dumps(CONFIG_RESOLVER))
|
|
config["profiles"]["reviewer-with-merge"] = {
|
|
"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",
|
|
],
|
|
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
|
|
"execution_profile": "reviewer-with-merge",
|
|
}
|
|
self._write_config(config)
|
|
with patch.dict(os.environ, self._env("reviewer-with-merge")):
|
|
res = mcp_server.gitea_resolve_task_capability(
|
|
task="merge_pr", remote="prgs"
|
|
)
|
|
self.assertEqual(res["required_role_kind"], "merger")
|
|
self.assertEqual(res["active_role_kind"], "reviewer")
|
|
self.assertTrue(res["active_profile_permission_allowed"])
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertTrue(res["stop_required"])
|
|
self.assertIn("cannot perform merger task", " ".join(res["task_role_guidance"]))
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "merger-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token merger-pass")
|
|
def test_merger_profile_can_resolve_merge_task(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("merger-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(
|
|
task="merge_pr", remote="prgs"
|
|
)
|
|
self.assertEqual(res["required_role_kind"], "merger")
|
|
self.assertEqual(res["active_role_kind"], "merger")
|
|
self.assertTrue(res["allowed_in_current_session"])
|
|
self.assertFalse(res["stop_required"])
|
|
|
|
@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_reconcile_already_landed_pr_requires_close(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = mcp_server.gitea_resolve_task_capability(
|
|
task="reconcile_already_landed_pr", remote="prgs"
|
|
)
|
|
self.assertEqual(res["requested_task"], "reconcile_already_landed_pr")
|
|
self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
|
|
self.assertEqual(res["required_role_kind"], "reconciler")
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
|
|
@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")
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_resolve_task_capability_does_not_poison_role_stamp_on_denial(self, _auth, _api):
|
|
# Issue #723: attempting to acquire a reviewer lease from a merger profile
|
|
# should fail without poisoning the session role stamp as 'reviewer'.
|
|
with patch.dict(os.environ, self._env("merger-profile")):
|
|
# Initially no preflight check is recorded
|
|
self.assertEqual(mcp_server._preflight_resolved_role, None)
|
|
|
|
# Request reviewer lease task which should be denied for merger
|
|
res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
|
|
|
|
# The task is denied
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
|
|
# The session role stamp should NOT be poisoned
|
|
self.assertNotEqual(mcp_server._preflight_resolved_role, "reviewer")
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|