Merge branch 'master' into feat/issue-727-pr-sync-status
Resolve task_capability_map conflict by keeping PR-sync entries and #725 merger-lease map entries. Fix author ownership for PR #728 (issue #727): prove lock via linked issue/session/branch instead of issue_number==pr_number. Add regression tests for 727/728 ownership and unrelated issue rejection.
This commit is contained in:
@@ -0,0 +1,466 @@
|
||||
"""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 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):
|
||||
def setUp(self):
|
||||
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=("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, mock_profile, _resolve, _auth, mock_api
|
||||
):
|
||||
mock_profile.return_value = self._merger_profile()
|
||||
mock_api.side_effect = [
|
||||
{"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"], 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=("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, mock_profile, _resolve, _auth, mock_api
|
||||
):
|
||||
mock_profile.return_value = self._merger_profile()
|
||||
active_body = leases.format_lease_body(
|
||||
repo="Scaled-Tech-Consulting/Gitea-Tools",
|
||||
pr_number=718,
|
||||
issue_number=None,
|
||||
reviewer_identity="other-user",
|
||||
profile="prgs-reviewer",
|
||||
session_id="other-session",
|
||||
worktree="branches/review-1",
|
||||
phase="claimed",
|
||||
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"}}
|
||||
]
|
||||
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"]))
|
||||
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")
|
||||
@patch("gitea_mcp_server._verify_role_mutation_workspace")
|
||||
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.assertIn("Branches-only mutation guard", res["reasons"][0])
|
||||
self.assertIsNone(leases._SESSION_LEASE)
|
||||
|
||||
@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_forwards_explicit_org_repo_to_workspace_verify(
|
||||
self, mock_verify, _comments, mock_profile, _resolve, _auth, mock_api
|
||||
):
|
||||
"""Explicit org/repo must reach anti-stomp (prgs bare default is Timesheet)."""
|
||||
mock_profile.return_value = self._merger_profile()
|
||||
mock_api.side_effect = [
|
||||
{"state": "open", "head": {"sha": LIVE_HEAD}},
|
||||
{"id": 789},
|
||||
]
|
||||
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"], res)
|
||||
kwargs = mock_verify.call_args.kwargs
|
||||
self.assertEqual(kwargs.get("org"), "Scaled-Tech-Consulting")
|
||||
self.assertEqual(kwargs.get("repo"), "Gitea-Tools")
|
||||
self.assertEqual(kwargs.get("task"), "acquire_merger_pr_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.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_fails_closed_when_live_head_unresolvable(
|
||||
self, _verify, _comments, mock_profile, _resolve, _auth, mock_api
|
||||
):
|
||||
"""Empty/missing live head must not silently proceed past exact-head gate."""
|
||||
mock_profile.return_value = self._merger_profile()
|
||||
# PR payload present but head.sha missing / empty / non-dict.
|
||||
mock_api.return_value = {"state": "open", "head": {}}
|
||||
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"], res)
|
||||
self.assertFalse(res.get("acquired"), res)
|
||||
self.assertTrue(
|
||||
any("live PR head could not be resolved" in r for r in res["reasons"]),
|
||||
res,
|
||||
)
|
||||
self.assertIsNone(res.get("live_head"))
|
||||
self.assertEqual(res.get("candidate_head"), HEAD)
|
||||
self.assertEqual(
|
||||
[c for c in mock_api.call_args_list if c[0][0] == "POST"], []
|
||||
)
|
||||
self.assertIsNone(leases._SESSION_LEASE)
|
||||
|
||||
# Completely empty GET /pulls body also fails closed.
|
||||
mock_api.return_value = {}
|
||||
res2 = mcp_server.gitea_acquire_merger_pr_lease(
|
||||
pr_number=718,
|
||||
worktree="branches/issue-718",
|
||||
candidate_head=HEAD,
|
||||
)
|
||||
self.assertFalse(res2["success"], res2)
|
||||
self.assertTrue(
|
||||
any("live PR head could not be resolved" in r for r in res2["reasons"]),
|
||||
res2,
|
||||
)
|
||||
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()
|
||||
@@ -0,0 +1,196 @@
|
||||
"""#727 / PR #728: author ownership when tracking issue number != PR number.
|
||||
|
||||
Regression for REQUEST_CHANGES: gitea_update_pr_branch_by_merge must not require
|
||||
issue_number == pr_number. Ownership is proven via linked issue + lock/branch
|
||||
context and fails closed for unrelated issues.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from unittest.mock import patch
|
||||
|
||||
import issue_lock_store
|
||||
import gitea_mcp_server as mcp
|
||||
|
||||
|
||||
def _live_lock(
|
||||
*,
|
||||
issue_number: int,
|
||||
branch_name: str = "feat/issue-727-pr-sync-status",
|
||||
worktree_path: str = "/tmp/branches/issue-727-pr-sync-status",
|
||||
remote: str = "prgs",
|
||||
org: str = "Scaled-Tech-Consulting",
|
||||
repo: str = "Gitea-Tools",
|
||||
) -> dict:
|
||||
now = datetime.now(timezone.utc)
|
||||
return {
|
||||
"issue_number": issue_number,
|
||||
"branch_name": branch_name,
|
||||
"worktree_path": worktree_path,
|
||||
"remote": remote,
|
||||
"org": org,
|
||||
"repo": repo,
|
||||
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
|
||||
"acquired_at": now.isoformat(),
|
||||
"expires_at": (now + timedelta(hours=2)).isoformat(),
|
||||
"owner_pid": os.getpid(),
|
||||
"status": "active",
|
||||
}
|
||||
|
||||
|
||||
class TestAuthorOwnershipIssuePrMismatch(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._tmp = tempfile.TemporaryDirectory(prefix="gitea-issue-locks-")
|
||||
self.lock_dir = self._tmp.name
|
||||
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir
|
||||
|
||||
def tearDown(self):
|
||||
# Clear session pointer for this process.
|
||||
try:
|
||||
ptr = issue_lock_store.session_pointer_path(self.lock_dir)
|
||||
if os.path.isfile(ptr):
|
||||
os.unlink(ptr)
|
||||
except Exception:
|
||||
pass
|
||||
os.environ.pop("GITEA_ISSUE_LOCK_DIR", None)
|
||||
self._tmp.cleanup()
|
||||
|
||||
def test_issue_727_pr_728_session_lock_accepted(self):
|
||||
"""Tracking issue #727 lock is valid ownership for PR #728."""
|
||||
lock = _live_lock(issue_number=727)
|
||||
issue_lock_store.bind_session_lock(lock)
|
||||
result = mcp._prove_author_ownership_for_pr(
|
||||
pr_number=728,
|
||||
pr_title="feat: pr sync",
|
||||
pr_body="Closes #727\n\nNative PR sync lifecycle.",
|
||||
source_branch="feat/issue-727-pr-sync-status",
|
||||
remote="prgs",
|
||||
host=None,
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
worktree_path=lock["worktree_path"],
|
||||
)
|
||||
self.assertTrue(result["proven"], result)
|
||||
self.assertTrue(result["has_author_lock"])
|
||||
self.assertEqual(result["matched_issue"], 727)
|
||||
self.assertEqual(result["matched_via"], "session_lock")
|
||||
self.assertIn(727, result["linked_issues"])
|
||||
|
||||
def test_issue_727_pr_728_durable_lock_accepted(self):
|
||||
lock = _live_lock(issue_number=727)
|
||||
path = issue_lock_store.lock_file_path(
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
issue_number=727,
|
||||
lock_dir=self.lock_dir,
|
||||
)
|
||||
issue_lock_store.save_lock_file(path, lock)
|
||||
result = mcp._prove_author_ownership_for_pr(
|
||||
pr_number=728,
|
||||
pr_title="feat: pr sync",
|
||||
pr_body="Fixes #727",
|
||||
source_branch="feat/issue-727-pr-sync-status",
|
||||
remote="prgs",
|
||||
host=None,
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
)
|
||||
self.assertTrue(result["proven"], result)
|
||||
self.assertEqual(result["matched_issue"], 727)
|
||||
self.assertEqual(result["matched_via"], "durable_lock")
|
||||
|
||||
def test_legacy_same_number_still_accepted(self):
|
||||
lock = _live_lock(
|
||||
issue_number=100,
|
||||
branch_name="fix/issue-100",
|
||||
worktree_path="/tmp/branches/issue-100",
|
||||
)
|
||||
issue_lock_store.bind_session_lock(lock)
|
||||
result = mcp._prove_author_ownership_for_pr(
|
||||
pr_number=100,
|
||||
pr_title="fix something",
|
||||
pr_body="Closes #100",
|
||||
source_branch="fix/issue-100",
|
||||
remote="prgs",
|
||||
host=None,
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
worktree_path=lock["worktree_path"],
|
||||
)
|
||||
self.assertTrue(result["proven"], result)
|
||||
self.assertEqual(result["matched_issue"], 100)
|
||||
|
||||
def test_unrelated_issue_lock_rejected(self):
|
||||
"""Lock for issue #999 must not authorize PR #728 linked only to #727."""
|
||||
lock = _live_lock(
|
||||
issue_number=999,
|
||||
branch_name="feat/issue-999",
|
||||
worktree_path="/tmp/branches/issue-999",
|
||||
)
|
||||
issue_lock_store.bind_session_lock(lock)
|
||||
path = issue_lock_store.lock_file_path(
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
issue_number=999,
|
||||
lock_dir=self.lock_dir,
|
||||
)
|
||||
issue_lock_store.save_lock_file(path, lock)
|
||||
result = mcp._prove_author_ownership_for_pr(
|
||||
pr_number=728,
|
||||
pr_title="feat: pr sync",
|
||||
pr_body="Closes #727",
|
||||
source_branch="feat/issue-727-pr-sync-status",
|
||||
remote="prgs",
|
||||
host=None,
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
worktree_path=lock["worktree_path"],
|
||||
)
|
||||
self.assertFalse(result["proven"], result)
|
||||
self.assertFalse(result["has_author_lock"])
|
||||
self.assertIsNone(result["matched_issue"])
|
||||
self.assertTrue(result["reasons"])
|
||||
|
||||
def test_branch_mismatch_on_session_lock_fail_closed(self):
|
||||
lock = _live_lock(
|
||||
issue_number=727,
|
||||
branch_name="feat/other-branch",
|
||||
)
|
||||
issue_lock_store.bind_session_lock(lock)
|
||||
result = mcp._prove_author_ownership_for_pr(
|
||||
pr_number=728,
|
||||
pr_title="feat: pr sync",
|
||||
pr_body="Closes #727",
|
||||
source_branch="feat/issue-727-pr-sync-status",
|
||||
remote="prgs",
|
||||
host=None,
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
worktree_path=lock["worktree_path"],
|
||||
)
|
||||
self.assertFalse(result["proven"], result)
|
||||
self.assertTrue(any("branch" in r for r in result["reasons"]))
|
||||
|
||||
def test_no_lock_fail_closed(self):
|
||||
result = mcp._prove_author_ownership_for_pr(
|
||||
pr_number=728,
|
||||
pr_title="feat: pr sync",
|
||||
pr_body="Closes #727",
|
||||
source_branch="feat/issue-727-pr-sync-status",
|
||||
remote="prgs",
|
||||
host=None,
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
)
|
||||
self.assertFalse(result["proven"], result)
|
||||
self.assertTrue(any("no live author issue lock" in r for r in result["reasons"]))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user