test: fix reviewer lease gate regressions in MCP integration tests (#407)
Align merge/review/audit mocks with the per-PR reviewer lease gate introduced in #407: install owned session leases after init_review_decision_lock, stub _fetch_pr_comments and _authenticated_username to preserve mock ordering, and update head-SHA mismatch expectations for lease-first fail-closed paths. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Claude Opus 4.8
+29
-3
@@ -286,6 +286,21 @@ class TestSimpleToolAudit(_AuditWiringBase):
|
||||
|
||||
class TestGatedToolAudit(_AuditWiringBase):
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
from tests.test_mcp_server import _install_owned_reviewer_lease
|
||||
import reviewer_pr_lease
|
||||
|
||||
self._lease_patch = _install_owned_reviewer_lease(8)
|
||||
self._lease_patch.start()
|
||||
self._auth_identity_patch = patch(
|
||||
"mcp_server._authenticated_username", return_value="reviewer-bot"
|
||||
)
|
||||
self._auth_identity_patch.start()
|
||||
self.addCleanup(self._auth_identity_patch.stop)
|
||||
self.addCleanup(self._lease_patch.stop)
|
||||
self.addCleanup(reviewer_pr_lease.clear_session_lease)
|
||||
|
||||
def _pr(self, author, state="open", sha="abc123", mergeable=True):
|
||||
return {"user": {"login": author}, "state": state,
|
||||
"head": {"sha": sha}, "mergeable": mergeable}
|
||||
@@ -344,11 +359,22 @@ class TestGatedToolAudit(_AuditWiringBase):
|
||||
GITEA_ALLOWED_OPERATIONS="read,review,approve")
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
|
||||
from tests.test_mcp_server import _install_owned_reviewer_lease
|
||||
import reviewer_pr_lease
|
||||
|
||||
init_review_decision_lock("prgs", "review_pr")
|
||||
gitea_mark_final_review_decision(8, "approve", remote="prgs")
|
||||
r = gitea_submit_pr_review(pr_number=8, action="approve",
|
||||
body="LGTM", remote="prgs",
|
||||
final_review_decision_ready=True)
|
||||
lease_patch = _install_owned_reviewer_lease(8)
|
||||
lease_patch.start()
|
||||
try:
|
||||
r = gitea_submit_pr_review(
|
||||
pr_number=8, action="approve",
|
||||
body="LGTM", remote="prgs",
|
||||
final_review_decision_ready=True,
|
||||
)
|
||||
finally:
|
||||
lease_patch.stop()
|
||||
reviewer_pr_lease.clear_session_lease()
|
||||
self.assertTrue(r["performed"])
|
||||
recs = self._records()
|
||||
self.assertEqual(len(recs), 1)
|
||||
|
||||
+127
-12
@@ -79,6 +79,64 @@ def _visible_approval_reviews(reviewer="reviewer-bot", sha="abc123"):
|
||||
return [_formal_review(reviewer, "APPROVED", sha=sha)]
|
||||
|
||||
|
||||
_DEFAULT_LEASE_SESSION = "mcp-test-reviewer-lease"
|
||||
|
||||
|
||||
def _reviewer_lease_comment(
|
||||
pr_number,
|
||||
*,
|
||||
session_id=_DEFAULT_LEASE_SESSION,
|
||||
head_sha="abc123",
|
||||
reviewer="reviewer-bot",
|
||||
):
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import reviewer_pr_lease
|
||||
|
||||
body = reviewer_pr_lease.format_lease_body(
|
||||
repo="Scaled-Tech-Consulting/Gitea-Tools",
|
||||
pr_number=pr_number,
|
||||
issue_number=407,
|
||||
reviewer_identity=reviewer,
|
||||
profile="gitea-reviewer",
|
||||
session_id=session_id,
|
||||
worktree="branches/review-test",
|
||||
phase="claimed",
|
||||
candidate_head=head_sha,
|
||||
target_branch="master",
|
||||
target_branch_sha="b" * 40,
|
||||
last_activity=datetime.now(timezone.utc),
|
||||
)
|
||||
return {"id": 9001, "body": body, "user": {"login": reviewer}}
|
||||
|
||||
|
||||
def _install_owned_reviewer_lease(
|
||||
pr_number,
|
||||
*,
|
||||
session_id=_DEFAULT_LEASE_SESSION,
|
||||
head_sha="abc123",
|
||||
):
|
||||
import reviewer_pr_lease
|
||||
|
||||
reviewer_pr_lease.clear_session_lease()
|
||||
reviewer_pr_lease.record_session_lease({
|
||||
"pr_number": pr_number,
|
||||
"session_id": session_id,
|
||||
"candidate_head": head_sha,
|
||||
"target_branch": "master",
|
||||
})
|
||||
return patch(
|
||||
"mcp_server._fetch_pr_comments",
|
||||
return_value=[
|
||||
_reviewer_lease_comment(
|
||||
pr_number,
|
||||
session_id=session_id,
|
||||
head_sha=head_sha,
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
# Issue-write tools are profile-gated (#69).
|
||||
ISSUE_WRITE_ENV = {
|
||||
"GITEA_ALLOWED_OPERATIONS": (
|
||||
@@ -512,6 +570,19 @@ class TestViewPR(unittest.TestCase):
|
||||
class TestMergePR(unittest.TestCase):
|
||||
"""Gated merge workflow (#16). gitea_merge_pr is the only merge path."""
|
||||
|
||||
def setUp(self):
|
||||
import reviewer_pr_lease
|
||||
|
||||
self._lease_patch = _install_owned_reviewer_lease(8)
|
||||
self._lease_patch.start()
|
||||
self._auth_identity_patch = patch(
|
||||
"mcp_server._authenticated_username", return_value="reviewer-bot"
|
||||
)
|
||||
self._auth_identity_patch.start()
|
||||
self.addCleanup(self._auth_identity_patch.stop)
|
||||
self.addCleanup(self._lease_patch.stop)
|
||||
self.addCleanup(reviewer_pr_lease.clear_session_lease)
|
||||
|
||||
def _pr(self, author, state="open", sha="abc123", mergeable=True):
|
||||
return {
|
||||
"user": {"login": author},
|
||||
@@ -772,9 +843,11 @@ class TestMergePR(unittest.TestCase):
|
||||
pr_number=8, confirmation=self._confirm(8),
|
||||
expected_head_sha="deadbeef", remote="prgs")
|
||||
self.assertFalse(r["performed"])
|
||||
self.assertIn(
|
||||
"expected head SHA does not match current PR head (fail closed)",
|
||||
r["reasons"])
|
||||
self.assertTrue(any(
|
||||
"expected head SHA does not match current PR head (fail closed)" in reason
|
||||
or "PR head changed during lease" in reason
|
||||
for reason in r["reasons"]
|
||||
))
|
||||
self._assert_no_merge_call(mock_api)
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@@ -1653,7 +1726,20 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
|
||||
}
|
||||
|
||||
def setUp(self):
|
||||
import reviewer_pr_lease
|
||||
|
||||
init_review_decision_lock("prgs", "review_pr")
|
||||
self._lease_patch = _install_owned_reviewer_lease(
|
||||
self.PR, head_sha=self.SHA,
|
||||
)
|
||||
self._lease_patch.start()
|
||||
self._auth_identity_patch = patch(
|
||||
"mcp_server._authenticated_username", return_value="reviewer-bot"
|
||||
)
|
||||
self._auth_identity_patch.start()
|
||||
self.addCleanup(self._auth_identity_patch.stop)
|
||||
self.addCleanup(self._lease_patch.stop)
|
||||
self.addCleanup(reviewer_pr_lease.clear_session_lease)
|
||||
|
||||
def _env(self):
|
||||
return patch.dict(os.environ, {
|
||||
@@ -1748,8 +1834,19 @@ class TestSubmitPrReview(unittest.TestCase):
|
||||
"""Gated review-mutation tool (#15)."""
|
||||
|
||||
def setUp(self):
|
||||
import reviewer_pr_lease
|
||||
|
||||
init_review_decision_lock("prgs", "review_pr")
|
||||
gitea_mark_final_review_decision(8, "approve", remote="prgs")
|
||||
self._lease_patch = _install_owned_reviewer_lease(8)
|
||||
self._lease_patch.start()
|
||||
self._auth_identity_patch = patch(
|
||||
"mcp_server._authenticated_username", return_value="reviewer-bot"
|
||||
)
|
||||
self._auth_identity_patch.start()
|
||||
self.addCleanup(self._auth_identity_patch.stop)
|
||||
self.addCleanup(self._lease_patch.stop)
|
||||
self.addCleanup(reviewer_pr_lease.clear_session_lease)
|
||||
|
||||
def _pr(self, author, state="open", sha="abc123", mergeable=True):
|
||||
return {
|
||||
@@ -1987,9 +2084,11 @@ class TestSubmitPrReview(unittest.TestCase):
|
||||
final_review_decision_ready=True,
|
||||
)
|
||||
self.assertFalse(r["performed"])
|
||||
self.assertIn(
|
||||
"expected head SHA does not match current PR head (fail closed)",
|
||||
r["reasons"])
|
||||
self.assertTrue(any(
|
||||
"expected head SHA does not match current PR head (fail closed)" in reason
|
||||
or "PR head changed during lease" in reason
|
||||
for reason in r["reasons"]
|
||||
))
|
||||
self._assert_no_mutation(mock_api)
|
||||
|
||||
def test_head_sha_match_allows(self):
|
||||
@@ -2052,9 +2151,9 @@ class TestSubmitPrReview(unittest.TestCase):
|
||||
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
|
||||
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
gitea_mark_final_review_decision(5, "approve", remote="prgs")
|
||||
gitea_mark_final_review_decision(8, "approve", remote="prgs")
|
||||
r = gitea_submit_pr_review(
|
||||
pr_number=5, action="approve", remote="prgs",
|
||||
pr_number=8, action="approve", remote="prgs",
|
||||
final_review_decision_ready=True,
|
||||
)
|
||||
self.assertFalse(r["performed"])
|
||||
@@ -2273,6 +2372,13 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
|
||||
self.assertEqual(res["cleanup_status"].get(1), "not present")
|
||||
|
||||
def test_merge_pr_with_closes_removes_label(self):
|
||||
import reviewer_pr_lease
|
||||
|
||||
lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123")
|
||||
lease_patch.start()
|
||||
self.addCleanup(lease_patch.stop)
|
||||
self.addCleanup(reviewer_pr_lease.clear_session_lease)
|
||||
|
||||
def api_side_effect(method, url, auth, payload=None):
|
||||
if method == "GET" and "/user" in url:
|
||||
return {"login": "merger"}
|
||||
@@ -2307,6 +2413,13 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
|
||||
self.assertEqual(res["cleanup_status"].get(123), "released")
|
||||
|
||||
def test_merge_pr_with_branch_name_removes_label(self):
|
||||
import reviewer_pr_lease
|
||||
|
||||
lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123")
|
||||
lease_patch.start()
|
||||
self.addCleanup(lease_patch.stop)
|
||||
self.addCleanup(reviewer_pr_lease.clear_session_lease)
|
||||
|
||||
def api_side_effect(method, url, auth, payload=None):
|
||||
if method == "GET" and "/user" in url:
|
||||
return {"login": "merger"}
|
||||
@@ -2988,10 +3101,12 @@ class TestVerifyMutationAuthority(unittest.TestCase):
|
||||
# profile; the active profile resolves as reviewer — side-channel
|
||||
# override rejected even with a matching in-process authority.
|
||||
self._authority()
|
||||
with patch.dict(os.environ,
|
||||
{"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_mutation_authority("prgs")
|
||||
with patch("mcp_server.gitea_config.is_runtime_switching_enabled",
|
||||
return_value=False):
|
||||
with patch.dict(os.environ,
|
||||
{"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_mutation_authority("prgs")
|
||||
self.assertIn("side-channel override rejected", str(ctx.exception))
|
||||
|
||||
def test_foreign_pid_authority_is_not_trusted(self):
|
||||
|
||||
@@ -156,9 +156,22 @@ class TestPRQueueInventory(unittest.TestCase):
|
||||
]
|
||||
|
||||
from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
|
||||
init_review_decision_lock("prgs", "review_pr")
|
||||
gitea_mark_final_review_decision(1, "approve", remote="prgs")
|
||||
result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs", final_review_decision_ready=True)
|
||||
from tests.test_mcp_server import _install_owned_reviewer_lease
|
||||
import reviewer_pr_lease
|
||||
|
||||
with patch("mcp_server._authenticated_username", return_value="reviewer1"):
|
||||
init_review_decision_lock("prgs", "review_pr")
|
||||
gitea_mark_final_review_decision(1, "approve", remote="prgs")
|
||||
lease_patch = _install_owned_reviewer_lease(
|
||||
1, head_sha="abc1", session_id="inventory-review-lease",
|
||||
)
|
||||
lease_patch.start()
|
||||
self.addCleanup(lease_patch.stop)
|
||||
self.addCleanup(reviewer_pr_lease.clear_session_lease)
|
||||
result = gitea_review_pr(
|
||||
pr_number=1, event="APPROVE", remote="prgs",
|
||||
final_review_decision_ready=True,
|
||||
)
|
||||
self.assertTrue(result["success"])
|
||||
self.assertIn("=== PR Queue Inventory ===", result["message"])
|
||||
self.assertIn("Repository:", result["message"])
|
||||
|
||||
Reference in New Issue
Block a user