Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 76 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

fix(mcp): complete merger lease acquire + structured capability resolve (#718, #723)

Browse Source 
Harden gitea_acquire_merger_pr_lease with required candidate_head, live-head
match, and fail-closed POST handling. Add capability-map aliases for reviewer
and merger lease acquire/adopt. Make unknown tasks return structured
unknown_task (no ValueError escape). Expand tests for roles, head scoping,
aliases, and unknown_task.

Closes #718. Addresses #723 resolver unknown_task and lease task mapping.
This commit is contained in:
sysadmin
2026-07-17 10:09:36 -04:00
parent 1421fa8568
commit 4f894009f6
4 changed files with 451 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_merger_lease_acquire.py
+323 -64
View File
@@ -1,15 +1,22 @@
"""Merger lease acquisition tests (#718)."""
"""Merger lease acquisition + capability resolution tests (#718, #723)."""
from __future__ import annotations
import os
import sys
import unittest
from datetime import datetime, timezone
from unittest.mock import patch
from unittest.mock import MagicMock, patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import gitea_mcp_server as mcp_server
import reviewer_pr_lease as leases
import task_capability_map as tcm
HEAD = "a" * 40
LIVE_HEAD = HEAD
class TestMergerLeaseAcquireTool(unittest.TestCase):
@@ -17,62 +24,80 @@ class TestMergerLeaseAcquireTool(unittest.TestCase):
mcp_server._preflight_whoami_called = True
mcp_server._preflight_capability_called = True
mcp_server._preflight_resolved_role = "merger"
mcp_server._preflight_resolved_task = "acquire_merger_pr_lease"
leases.clear_session_lease()
def tearDown(self):
leases.clear_session_lease()
def _merger_profile(self, **extra):
p = {
"username": "merger-user",
"profile_name": "prgs-merger",
"role": "merger",
"allowed_operations": [
"gitea.read",
"gitea.pr.comment",
"gitea.pr.merge",
],
"forbidden_operations": [
"gitea.pr.approve",
"gitea.pr.review",
"gitea.pr.request_changes",
],
}
p.update(extra)
return p
@patch("gitea_mcp_server.api_request")
@patch("gitea_mcp_server._auth", return_value="token pass")
@patch("gitea_mcp_server._resolve", return_value=("dadeschools", "org", "repo"))
@patch("gitea_mcp_server.get_profile", return_value={"username": "merger-user", "profile_name": "prgs-merger", "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"], "forbidden_operations": []})
@patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
@patch("gitea_mcp_server.get_profile")
@patch("gitea_mcp_server._fetch_pr_comments", return_value=[])
@patch("gitea_mcp_server._verify_role_mutation_workspace")
def test_acquire_merger_lease_success(self, _verify, _comments, _profile, _resolve, _auth, mock_api):
"""Merger lease acquisition succeeds when no reviewer lease exists."""
# api_request is called for PR state, then for POSTing comment
def test_acquire_merger_lease_success(
self, _verify, _comments, mock_profile, _resolve, _auth, mock_api
):
mock_profile.return_value = self._merger_profile()
mock_api.side_effect = [
{"state": "open"}, # PR is open
{"id": 456}, # Posted comment ID
{"state": "open", "head": {"sha": LIVE_HEAD}},
{"id": 456},
]
with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
res = mcp_server.gitea_acquire_merger_pr_lease(
pr_number=718,
worktree="branches/issue-718",
session_id="merger-session",
candidate_head=HEAD,
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
self.assertTrue(res["success"])
self.assertTrue(res["acquired"])
self.assertEqual(res["comment_id"], 456)
self.assertEqual(res["session_id"], "merger-session")
# Verify lease body includes merger identity
args, kwargs = mock_api.call_args_list[-1]
self.assertEqual(args[0], "POST")
body = args[3]["body"]
self.assertIn("reviewer_identity: merger-user", body)
self.assertIn("profile: prgs-merger", body)
self.assertIn("phase: claimed", body)
# Post-mutation readback: lease is recorded in session
session_lease = leases._SESSION_LEASE
self.assertIsNotNone(session_lease)
self.assertEqual(session_lease["phase"], "claimed")
self.assertEqual(session_lease["session_id"], "merger-session")
# Verify lease provenance uses SOURCE_ACQUIRE_MERGER
provenance = session_lease.get("lease_provenance") or {}
self.assertEqual(provenance.get("source"), "gitea_acquire_merger_pr_lease")
self.assertTrue(res["success"], res)
self.assertTrue(res["acquired"])
self.assertEqual(res["comment_id"], 456)
self.assertEqual(res["session_id"], "merger-session")
self.assertEqual(res.get("repo"), "Scaled-Tech-Consulting/Gitea-Tools")
body = mock_api.call_args_list[-1][0][3]["body"]
self.assertIn("reviewer_identity: merger-user", body)
self.assertIn("profile: prgs-merger", body)
session_lease = leases._SESSION_LEASE
self.assertIsNotNone(session_lease)
provenance = session_lease.get("lease_provenance") or {}
self.assertEqual(provenance.get("source"), "gitea_acquire_merger_pr_lease")
@patch("gitea_mcp_server.api_request")
@patch("gitea_mcp_server._auth", return_value="token pass")
@patch("gitea_mcp_server._resolve", return_value=("dadeschools", "org", "repo"))
@patch("gitea_mcp_server.get_profile", return_value={"username": "merger-user", "profile_name": "prgs-merger", "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"], "forbidden_operations": []})
@patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
@patch("gitea_mcp_server.get_profile")
@patch("gitea_mcp_server._fetch_pr_comments")
@patch("gitea_mcp_server._verify_role_mutation_workspace")
def test_acquire_merger_lease_fails_if_active_exists(self, _verify, _comments, _profile, _resolve, _auth, mock_api):
"""Acquisition fails closed if another session holds an active lease."""
def test_acquire_merger_lease_fails_if_active_exists(
self, _verify, _comments, mock_profile, _resolve, _auth, mock_api
):
mock_profile.return_value = self._merger_profile()
active_body = leases.format_lease_body(
repo="org/repo",
repo="Scaled-Tech-Consulting/Gitea-Tools",
pr_number=718,
issue_number=None,
reviewer_identity="other-user",
@@ -80,51 +105,285 @@ class TestMergerLeaseAcquireTool(unittest.TestCase):
session_id="other-session",
worktree="branches/review-1",
phase="claimed",
candidate_head="a" * 40,
candidate_head=HEAD,
target_branch="master",
target_branch_sha="b" * 40,
last_activity=datetime.now(timezone.utc),
)
_comments.return_value = [{"id": 1, "body": active_body, "user": {"login": "other-user"}}]
# PR is open
mock_api.return_value = {"state": "open"}
_comments.return_value = [
{"id": 1, "body": active_body, "user": {"login": "other-user"}}
]
mock_api.return_value = {"state": "open", "head": {"sha": LIVE_HEAD}}
with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
res = mcp_server.gitea_acquire_merger_pr_lease(
pr_number=718,
worktree="branches/issue-718",
session_id="merger-session",
candidate_head=HEAD,
)
self.assertFalse(res["success"])
self.assertFalse(res["acquired"])
self.assertIn("already has active reviewer lease", " ".join(res["reasons"]))
# No POST made
post_calls = [c for c in mock_api.call_args_list if c[0][0] == "POST"]
self.assertEqual(len(post_calls), 0)
# No partial state recorded
self.assertIsNone(leases._SESSION_LEASE)
self.assertFalse(res["success"])
self.assertFalse(res["acquired"])
self.assertIn("already has active reviewer lease", " ".join(res["reasons"]))
self.assertEqual(
[c for c in mock_api.call_args_list if c[0][0] == "POST"], []
)
self.assertIsNone(leases._SESSION_LEASE)
@patch("gitea_mcp_server.get_profile", return_value={"username": "merger-user", "profile_name": "prgs-merger", "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"], "forbidden_operations": []})
@patch("gitea_mcp_server.get_profile")
@patch("gitea_mcp_server._verify_role_mutation_workspace")
def test_acquire_merger_lease_fails_workspace_binding(self, _verify, _profile):
"""Fails closed if workspace verification throws RuntimeError."""
def test_acquire_merger_lease_fails_workspace_binding(self, _verify, mock_profile):
mock_profile.return_value = self._merger_profile()
_verify.side_effect = RuntimeError("Branches-only mutation guard")
with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
res = mcp_server.gitea_acquire_merger_pr_lease(
pr_number=718,
worktree="branches/issue-718",
session_id="merger-session",
candidate_head=HEAD,
)
self.assertFalse(res["success"])
self.assertFalse(res["acquired"])
self.assertIn("Branches-only mutation guard", res["reasons"][0])
self.assertIsNone(leases._SESSION_LEASE)
self.assertFalse(res["success"])
self.assertIn("Branches-only mutation guard", res["reasons"][0])
self.assertIsNone(leases._SESSION_LEASE)
@patch("gitea_mcp_server.get_profile")
def test_acquire_merger_requires_candidate_head(self, mock_profile):
mock_profile.return_value = self._merger_profile()
res = mcp_server.gitea_acquire_merger_pr_lease(
pr_number=718,
worktree="branches/issue-718",
candidate_head=None,
)
self.assertFalse(res["success"])
self.assertTrue(any("candidate_head is required" in r for r in res["reasons"]))
@patch("gitea_mcp_server.api_request")
@patch("gitea_mcp_server._auth", return_value="token pass")
@patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"))
@patch("gitea_mcp_server.get_profile")
@patch("gitea_mcp_server._fetch_pr_comments", return_value=[])
@patch("gitea_mcp_server._verify_role_mutation_workspace")
def test_acquire_merger_head_mismatch(
self, _verify, _comments, mock_profile, _resolve, _auth, mock_api
):
mock_profile.return_value = self._merger_profile()
mock_api.return_value = {"state": "open", "head": {"sha": "b" * 40}}
res = mcp_server.gitea_acquire_merger_pr_lease(
pr_number=718,
worktree="branches/issue-718",
candidate_head=HEAD,
)
self.assertFalse(res["success"])
self.assertTrue(any("does not match live PR head" in r for r in res["reasons"]))
self.assertIsNone(leases._SESSION_LEASE)
@patch("gitea_mcp_server._profile_operation_gate")
def test_author_denied_merge_permission(self, mock_gate):
"""Author profile lacks gitea.pr.merge → fail closed before mutation."""
def gate(op):
if op == "gitea.pr.merge":
return [f"profile lacks {op}"]
return None
mock_gate.side_effect = gate
res = mcp_server.gitea_acquire_merger_pr_lease(
pr_number=718,
worktree="branches/issue-718",
candidate_head=HEAD,
)
self.assertFalse(res["success"])
self.assertTrue(any("gitea.pr.merge" in r for r in res["reasons"]))
@patch("gitea_mcp_server._profile_operation_gate")
def test_reviewer_denied_merge_permission(self, mock_gate):
def gate(op):
if op == "gitea.pr.merge":
return [f"profile lacks {op}"]
return None
mock_gate.side_effect = gate
res = mcp_server.gitea_acquire_merger_pr_lease(
pr_number=718,
worktree="branches/issue-718",
candidate_head=HEAD,
)
self.assertFalse(res["success"])
self.assertTrue(any("gitea.pr.merge" in r for r in res["reasons"]))
class TestCapabilityMapLeaseTasks(unittest.TestCase):
def test_merger_acquire_map_entries(self):
for task in ("acquire_merger_pr_lease", "gitea_acquire_merger_pr_lease"):
self.assertEqual(tcm.required_permission(task), "gitea.pr.comment")
self.assertEqual(tcm.required_role(task), "merger")
def test_reviewer_acquire_map_entries(self):
for task in ("acquire_reviewer_pr_lease", "gitea_acquire_reviewer_pr_lease"):
self.assertEqual(tcm.required_permission(task), "gitea.pr.comment")
self.assertEqual(tcm.required_role(task), "reviewer")
def test_adopt_merger_aliases(self):
for task in ("adopt_merger_pr_lease", "gitea_adopt_merger_pr_lease"):
self.assertEqual(tcm.required_role(task), "merger")
class TestResolveTaskCapabilityLeaseAndUnknown(unittest.TestCase):
def setUp(self):
mcp_server._process_boot_head_sha = None
if hasattr(mcp_server, "capability_stop_terminal"):
mcp_server.capability_stop_terminal.clear()
@patch.object(mcp_server, "record_mutation_authority", return_value=None)
@patch.object(mcp_server, "init_review_decision_lock", return_value=None)
@patch.object(mcp_server, "record_preflight_check", return_value=None)
@patch.object(mcp_server, "_ensure_matching_profile", return_value=None)
@patch.object(mcp_server, "_authenticated_username", return_value="sysadmin")
@patch.object(mcp_server, "get_profile")
@patch.object(mcp_server.gitea_config, "load_config")
def test_resolve_acquire_reviewer_authorized(
self, mock_cfg, mock_profile, *_mocks
):
mock_profile.return_value = {
"profile_name": "prgs-reviewer",
"role": "reviewer",
"allowed_operations": ["gitea.pr.comment", "gitea.read", "gitea.pr.review"],
"forbidden_operations": [],
}
mock_cfg.return_value = {
"profiles": {
"prgs-reviewer": {
"role": "reviewer",
"allowed_operations": [
"gitea.pr.comment",
"gitea.read",
"gitea.pr.review",
],
"forbidden_operations": [],
}
}
}
with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-reviewer"}, clear=False):
for task in ("acquire_reviewer_pr_lease", "gitea_acquire_reviewer_pr_lease"):
res = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs")
self.assertEqual(res["required_role_kind"], "reviewer", res)
self.assertEqual(
res["required_operation_permission"], "gitea.pr.comment", res
)
self.assertTrue(res.get("allowed_in_current_session"), res)
@patch.object(mcp_server, "record_mutation_authority", return_value=None)
@patch.object(mcp_server, "init_review_decision_lock", return_value=None)
@patch.object(mcp_server, "record_preflight_check", return_value=None)
@patch.object(mcp_server, "_ensure_matching_profile", return_value=None)
@patch.object(mcp_server, "_authenticated_username", return_value="jcwalker3")
@patch.object(mcp_server, "get_profile")
@patch.object(mcp_server.gitea_config, "load_config")
def test_resolve_acquire_reviewer_author_denied(
self, mock_cfg, mock_profile, *_mocks
):
mock_profile.return_value = {
"profile_name": "prgs-author",
"role": "author",
"allowed_operations": ["gitea.pr.comment", "gitea.branch.push", "gitea.read"],
"forbidden_operations": [],
}
mock_cfg.return_value = {
"profiles": {
"prgs-author": {
"role": "author",
"allowed_operations": [
"gitea.pr.comment",
"gitea.branch.push",
"gitea.read",
],
"forbidden_operations": [],
},
"prgs-reviewer": {
"role": "reviewer",
"allowed_operations": ["gitea.pr.comment", "gitea.pr.review"],
"forbidden_operations": [],
},
}
}
with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-author"}, clear=False):
res = mcp_server.gitea_resolve_task_capability(
task="acquire_reviewer_pr_lease", remote="prgs"
)
self.assertFalse(res.get("allowed_in_current_session"), res)
self.assertEqual(res["required_role_kind"], "reviewer")
guidance = " ".join(res.get("task_role_guidance") or [])
self.assertTrue(
"cannot perform reviewer" in guidance.lower()
or "reviewer" in guidance.lower()
or res.get("stop_required"),
res,
)
@patch.object(mcp_server, "record_mutation_authority", return_value=None)
@patch.object(mcp_server, "init_review_decision_lock", return_value=None)
@patch.object(mcp_server, "record_preflight_check", return_value=None)
@patch.object(mcp_server, "_ensure_matching_profile", return_value=None)
@patch.object(mcp_server, "_authenticated_username", return_value="sysadmin")
@patch.object(mcp_server, "get_profile")
def test_resolve_unknown_task_structured_no_raise(self, mock_profile, *_mocks):
mock_profile.return_value = {
"profile_name": "prgs-reviewer",
"role": "reviewer",
"allowed_operations": ["gitea.pr.comment"],
"forbidden_operations": [],
}
# Must not raise ValueError into the tool boundary.
res = mcp_server.gitea_resolve_task_capability(
task="some_made_up_task", remote="prgs"
)
self.assertIsInstance(res, dict)
self.assertEqual(res.get("reason_code"), "unknown_task", res)
self.assertFalse(res.get("allowed_in_current_session"))
self.assertTrue(res.get("stop_required"))
self.assertIs(res.get("mutation_performed"), False)
self.assertNotIn("token", str(res).lower())
self.assertNotIn("password", str(res).lower())
guidance = " ".join(res.get("task_role_guidance") or [])
self.assertIn("Unknown task/action", guidance)
@patch.object(mcp_server, "record_mutation_authority", return_value=None)
@patch.object(mcp_server, "init_review_decision_lock", return_value=None)
@patch.object(mcp_server, "record_preflight_check", return_value=None)
@patch.object(mcp_server, "_ensure_matching_profile", return_value=None)
@patch.object(mcp_server, "_authenticated_username", return_value="sysadmin")
@patch.object(mcp_server, "get_profile")
@patch.object(mcp_server.gitea_config, "load_config")
def test_resolve_merger_acquire_aliases(
self, mock_cfg, mock_profile, *_mocks
):
mock_profile.return_value = {
"profile_name": "prgs-merger",
"role": "merger",
"allowed_operations": [
"gitea.read",
"gitea.pr.comment",
"gitea.pr.merge",
],
"forbidden_operations": [],
}
mock_cfg.return_value = {
"profiles": {
"prgs-merger": {
"role": "merger",
"allowed_operations": [
"gitea.read",
"gitea.pr.comment",
"gitea.pr.merge",
],
"forbidden_operations": [],
}
}
}
with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-merger"}, clear=False):
for task in ("acquire_merger_pr_lease", "gitea_acquire_merger_pr_lease"):
res = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs")
self.assertEqual(res["required_role_kind"], "merger", res)
self.assertEqual(
res["required_operation_permission"], "gitea.pr.comment", res
)
if __name__ == "__main__":
unittest.main()
tests/test_resolve_task_capability.py
+14 -4
View File
@@ -231,9 +231,16 @@ class TestResolveTaskCapability(unittest.TestCase):
self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
def test_resolve_unknown_task_fails_closed(self):
# #723: unknown tasks return structured unknown_task (no ValueError escape).
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")
res = mcp_server.gitea_resolve_task_capability(
task="invalid_task_xyz", remote="prgs"
)
self.assertIsInstance(res, dict)
self.assertEqual(res.get("reason_code"), "unknown_task", res)
self.assertFalse(res.get("allowed_in_current_session"))
self.assertTrue(res.get("stop_required"))
self.assertIs(res.get("mutation_performed"), False)
# Additional regression tests per #145 for permission boundaries and structured guidance
def test_issue_comment_does_not_imply_close(self):
@@ -439,8 +446,11 @@ class TestResolveTaskCapability(unittest.TestCase):
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")
unk = mcp_server.gitea_resolve_task_capability(
task=unknown, remote="prgs"
)
self.assertEqual(unk.get("reason_code"), "unknown_task", unk)
self.assertFalse(unk.get("allowed_in_current_session"))
if __name__ == "__main__":
unittest.main()