Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c8dcce4c15 |
+89
-3
@@ -438,9 +438,6 @@ def _ensure_matching_profile(required_permission: str, required_role: str, remot
|
||||
for p_name, p_data in config["profiles"].items():
|
||||
p_allowed = p_data.get("allowed_operations") or []
|
||||
p_forbidden = p_data.get("forbidden_operations") or []
|
||||
p_role = p_data.get("role") or _role_kind(p_allowed, p_forbidden)
|
||||
if required_role and p_role != required_role:
|
||||
continue
|
||||
p_allowed_n = []
|
||||
for op in p_allowed:
|
||||
try:
|
||||
@@ -3883,6 +3880,90 @@ def gitea_get_profile(
|
||||
return result
|
||||
|
||||
|
||||
_RUNTIME_CAPABILITY_TASKS = (
|
||||
"create_issue",
|
||||
"comment_issue",
|
||||
"create_pr",
|
||||
"review_pr",
|
||||
"merge_pr",
|
||||
"close_issue",
|
||||
)
|
||||
|
||||
|
||||
def _matching_configured_profiles(
|
||||
config: dict | None,
|
||||
required_permission: str,
|
||||
) -> list[str]:
|
||||
"""Profile names that allow *required_permission* (redacted metadata only)."""
|
||||
if not config or "profiles" not in config:
|
||||
return []
|
||||
matches: list[str] = []
|
||||
for p_name, p_data in config["profiles"].items():
|
||||
if not p_data.get("enabled", True):
|
||||
continue
|
||||
p_allowed = p_data.get("allowed_operations") or []
|
||||
p_forbidden = p_data.get("forbidden_operations") or []
|
||||
p_allowed_n = []
|
||||
for op in p_allowed:
|
||||
try:
|
||||
p_allowed_n.append(gitea_config.normalize_operation(op))
|
||||
except Exception:
|
||||
pass
|
||||
p_forbidden_n = []
|
||||
for op in p_forbidden:
|
||||
try:
|
||||
p_forbidden_n.append(gitea_config.normalize_operation(op))
|
||||
except Exception:
|
||||
pass
|
||||
ok, _ = gitea_config.check_operation(
|
||||
required_permission, p_allowed_n, p_forbidden_n
|
||||
)
|
||||
if ok:
|
||||
matches.append(p_name)
|
||||
return sorted(matches)
|
||||
|
||||
|
||||
def _build_runtime_task_capabilities(
|
||||
allowed: list[str],
|
||||
forbidden: list[str],
|
||||
config: dict | None,
|
||||
) -> dict:
|
||||
"""Per-task capability summary for role-aware runtime context (#139)."""
|
||||
task_entries = []
|
||||
flags: dict[str, bool] = {}
|
||||
flag_keys = {
|
||||
"create_issue": "can_create_issues",
|
||||
"comment_issue": "can_comment_on_issues",
|
||||
"create_pr": "can_author_prs",
|
||||
"review_pr": "can_review_prs",
|
||||
"merge_pr": "can_merge_prs",
|
||||
"close_issue": "can_close_issues",
|
||||
}
|
||||
for task in _RUNTIME_CAPABILITY_TASKS:
|
||||
permission = task_capability_map.required_permission(task)
|
||||
allowed_here, _ = gitea_config.check_operation(
|
||||
permission, allowed, forbidden
|
||||
)
|
||||
entry = {
|
||||
"task": task,
|
||||
"required_permission": permission,
|
||||
"required_role_kind": task_capability_map.required_role(task),
|
||||
"allowed_in_current_session": allowed_here,
|
||||
"matching_configured_profiles": _matching_configured_profiles(
|
||||
config, permission
|
||||
),
|
||||
}
|
||||
task_entries.append(entry)
|
||||
flag_name = flag_keys.get(task)
|
||||
if flag_name:
|
||||
flags[flag_name] = allowed_here
|
||||
return {
|
||||
**flags,
|
||||
"issue_comment_not_implied_by_pr_comment": True,
|
||||
"task_capabilities": task_entries,
|
||||
}
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_get_runtime_context(
|
||||
remote: str = "dadeschools",
|
||||
@@ -3970,6 +4051,10 @@ def gitea_get_runtime_context(
|
||||
"or ask the operator to update GITEA_MCP_PROFILE to a reviewer profile."
|
||||
)
|
||||
|
||||
session_capabilities = _build_runtime_task_capabilities(
|
||||
allowed, forbidden, config
|
||||
)
|
||||
|
||||
result = {
|
||||
"active_profile": profile["profile_name"],
|
||||
"authenticated_username": username,
|
||||
@@ -3984,6 +4069,7 @@ def gitea_get_runtime_context(
|
||||
"review_merge_blocked_reasons": blocked_reasons,
|
||||
"suggested_fix": suggested_fix,
|
||||
"safe_next_action": safe_next_action,
|
||||
"session_capabilities": session_capabilities,
|
||||
}
|
||||
|
||||
if reveal and h:
|
||||
|
||||
@@ -168,74 +168,6 @@ class TestRoleSessionRouter(unittest.TestCase):
|
||||
self.assertFalse(result["success"])
|
||||
self.assertIn("switching is disabled", result["message"])
|
||||
|
||||
def _switching_env(self, profile):
|
||||
switching_config = dict(CONFIG)
|
||||
switching_config["rules"] = {"allow_runtime_switching": True}
|
||||
switching_config_path = os.path.join(self._dir.name, "profiles_switching.json")
|
||||
with open(switching_config_path, "w", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(switching_config))
|
||||
return {
|
||||
"GITEA_MCP_CONFIG": switching_config_path,
|
||||
"GITEA_MCP_PROFILE": profile,
|
||||
"GITEA_TOKEN_AUTHOR": "author-pass",
|
||||
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
|
||||
}
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
||||
def test_dynamic_switch_to_reviewer_profile(self, _auth, mock_api):
|
||||
mock_api.side_effect = lambda method, url, header: (
|
||||
{"login": "sysadmin"} if "reviewer-pass" in str(header) else {"login": "jcwalker3"}
|
||||
)
|
||||
with patch.dict(os.environ, self._switching_env("prgs-author")):
|
||||
self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
|
||||
route = mcp_server.gitea_route_task_session(
|
||||
task_type="review_pr", remote="prgs"
|
||||
)
|
||||
self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
|
||||
self.assertTrue(route["downstream_allowed"])
|
||||
self.assertEqual(gitea_config.selected_profile_name(), "prgs-reviewer")
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_dynamic_switch_to_author_profile(self, _auth, mock_api):
|
||||
mock_api.side_effect = lambda method, url, header: (
|
||||
{"login": "jcwalker3"} if "author-pass" in str(header) else {"login": "sysadmin"}
|
||||
)
|
||||
with patch.dict(os.environ, self._switching_env("prgs-reviewer")):
|
||||
self.assertEqual(gitea_config.selected_profile_name(), "prgs-reviewer")
|
||||
route = mcp_server.gitea_route_task_session(
|
||||
task_type="create_issue", remote="prgs"
|
||||
)
|
||||
self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
|
||||
self.assertTrue(route["downstream_allowed"])
|
||||
self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
|
||||
|
||||
def test_dynamic_switch_blocked_for_wrong_role(self):
|
||||
with patch.dict(os.environ, self._switching_env("prgs-author")):
|
||||
config_data = dict(CONFIG)
|
||||
config_data["rules"] = {"allow_runtime_switching": True}
|
||||
config_data["profiles"]["prgs-author"]["allowed_operations"] = [
|
||||
"gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push"
|
||||
]
|
||||
custom_path = os.path.join(self._dir.name, "profiles_wrong_role.json")
|
||||
with open(custom_path, "w", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(config_data))
|
||||
|
||||
env = {
|
||||
"GITEA_MCP_CONFIG": custom_path,
|
||||
"GITEA_MCP_PROFILE": "prgs-author",
|
||||
"GITEA_TOKEN_AUTHOR": "author-pass",
|
||||
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
|
||||
}
|
||||
with patch.dict(os.environ, env):
|
||||
route = mcp_server.gitea_route_task_session(
|
||||
task_type="comment_issue", remote="prgs"
|
||||
)
|
||||
self.assertEqual(route["route_result"], role_session_router.ROUTE_TO_AUTHOR)
|
||||
self.assertFalse(route["downstream_allowed"])
|
||||
self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
|
||||
|
||||
def test_handoff_requires_route_fields(self):
|
||||
incomplete = assess_role_route_handoff("Task done.")
|
||||
self.assertFalse(incomplete["complete"])
|
||||
|
||||
@@ -110,6 +110,18 @@ class TestRuntimeClarity(unittest.TestCase):
|
||||
self.assertEqual(ctx["suggested_fix"], "reviewer namespace")
|
||||
self.assertIn("does not permit review or merge", ctx["review_merge_blocked_reasons"][0])
|
||||
self.assertIn("Switch to the reviewer MCP session", ctx["safe_next_action"])
|
||||
caps = ctx["session_capabilities"]
|
||||
self.assertTrue(caps["can_author_prs"])
|
||||
self.assertFalse(caps["can_create_issues"])
|
||||
self.assertFalse(caps["can_comment_on_issues"])
|
||||
self.assertFalse(caps["can_review_prs"])
|
||||
self.assertFalse(caps["can_merge_prs"])
|
||||
self.assertTrue(caps["issue_comment_not_implied_by_pr_comment"])
|
||||
merge_entry = next(
|
||||
t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
|
||||
)
|
||||
self.assertFalse(merge_entry["allowed_in_current_session"])
|
||||
self.assertIn("reviewer-profile", merge_entry["matching_configured_profiles"])
|
||||
|
||||
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
|
||||
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
||||
@@ -121,6 +133,15 @@ class TestRuntimeClarity(unittest.TestCase):
|
||||
self.assertTrue(ctx["review_merge_allowed"])
|
||||
self.assertEqual(ctx["suggested_fix"], "none")
|
||||
self.assertEqual(ctx["safe_next_action"], "None; ready for operations.")
|
||||
caps = ctx["session_capabilities"]
|
||||
self.assertFalse(caps["can_author_prs"])
|
||||
self.assertFalse(caps["can_create_issues"])
|
||||
self.assertFalse(caps["can_review_prs"])
|
||||
self.assertTrue(caps["can_merge_prs"])
|
||||
merge_entry = next(
|
||||
t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
|
||||
)
|
||||
self.assertTrue(merge_entry["allowed_in_current_session"])
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# gitea_list_profiles
|
||||
|
||||
Reference in New Issue
Block a user