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

fix(mcp): require visible APPROVED reviews before merge (closes #244)

Browse Source 
Gitea expects review event APPROVED, not APPROVE; the wrong event left
submissions as PENDING drafts. Verify terminal verdicts after submit,
submit pending drafts when needed, and block merge when approval_visible
is false.
This commit is contained in:
sysadmin
2026-07-06 11:51:29 -04:00
parent bab803ff3d
commit a1ba69eebb
4 changed files with 262 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_audit.py
+8 -2
View File
@@ -293,9 +293,12 @@ class TestGatedToolAudit(_AuditWiringBase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_merge_success_audited(self, _auth, mock_api):
# user, pr, merge POST, readback — no extra identity call (uses result).
# user, pr, feedback pr+reviews, merge POST, readback.
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
self._pr("author-bot"),
[{"id": 1, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
"submitted_at": "2026-07-06T10:00:00Z", "dismissed": False}],
{}, {"merged_commit_sha": "c1"},
]
env = self._env(GITEA_PROFILE_NAME="gitea-merger",
@@ -332,7 +335,10 @@ class TestGatedToolAudit(_AuditWiringBase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_submit_review_success_audited(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 7},
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 7, "state": "APPROVED"},
[{"id": 7, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
"submitted_at": "2026-07-06T10:00:00Z", "dismissed": False}],
]
env = self._env(GITEA_PROFILE_NAME="gitea-reviewer",
GITEA_ALLOWED_OPERATIONS="read,review,approve")
tests/test_mcp_server.py
+149 -17
View File
@@ -47,6 +47,22 @@ import mcp_server
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
def _formal_review(reviewer, verdict, sha="abc123", review_id=1):
return {
"id": review_id,
"user": {"login": reviewer},
"state": verdict,
"commit_id": sha,
"submitted_at": "2026-07-06T10:00:00Z",
"dismissed": False,
}
def _visible_approval_reviews(reviewer="reviewer-bot", sha="abc123"):
return [_formal_review(reviewer, "APPROVED", sha=sha)]
# Issue-write tools are profile-gated (#69).
ISSUE_WRITE_ENV = {
"GITEA_ALLOWED_OPERATIONS": (
@@ -488,6 +504,10 @@ class TestMergePR(unittest.TestCase):
f"unexpected merge mutation: {method} {url}",
)
def _feedback_reads(self, author="author-bot", sha="abc123"):
"""PR + reviews GETs for gitea_get_pr_review_feedback during merge."""
return [self._pr(author, sha=sha), _visible_approval_reviews(sha=sha)]
# -- success --------------------------------------------------------------
@patch("mcp_server.api_request")
@@ -495,6 +515,7 @@ class TestMergePR(unittest.TestCase):
def test_merge_succeeds_when_all_gates_pass(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
*self._feedback_reads(),
{}, # merge POST
{"merged_commit_sha": "mergecommit99"}, # read-back
]
@@ -511,8 +532,8 @@ class TestMergePR(unittest.TestCase):
self.assertEqual(r["head_sha"], "abc123")
self.assertEqual(r["merge_method"], "squash")
self.assertEqual(r["merge_commit"], "mergecommit99")
# 3rd call is the merge POST with the requested method/title/message.
merge_call = mock_api.call_args_list[2]
# 5th call is the merge POST with the requested method/title/message.
merge_call = mock_api.call_args_list[4]
self.assertEqual(merge_call.args[0], "POST")
self.assertTrue(merge_call.args[1].endswith("/pulls/8/merge"))
payload = merge_call.args[3]
@@ -527,6 +548,7 @@ class TestMergePR(unittest.TestCase):
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
[{"filename": "a.py"}, {"filename": "b.py"}], # files
*self._feedback_reads(),
{}, # merge POST
{"merged_commit_sha": "c1"}, # read-back
]
@@ -546,6 +568,7 @@ class TestMergePR(unittest.TestCase):
"""Merge OK + read-back GET failure => explicit cleanup skip, not silence."""
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
*self._feedback_reads(),
{}, # merge POST
RuntimeError("HTTP 502: Gitea upstream unavailable"), # read-back fails
]
@@ -561,8 +584,8 @@ class TestMergePR(unittest.TestCase):
# The skip is explicit, not silent.
self.assertEqual(r["cleanup_status"], "skipped (merge read-back failed)")
# No tracker-cleanup API traffic after the failed read-back:
# user, PR (eligibility), merge POST, read-back — and nothing more.
self.assertEqual(mock_api.call_count, 4)
# user, PR (eligibility), feedback PR+reviews, merge POST, read-back.
self.assertEqual(mock_api.call_count, 6)
for c in mock_api.call_args_list:
self.assertNotEqual(c.args[0], "DELETE")
@@ -574,6 +597,7 @@ class TestMergePR(unittest.TestCase):
"""Unexpected cleanup exception => merge still succeeds; error surfaced redacted."""
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
*self._feedback_reads(),
{}, # merge POST
{"merged_commit_sha": "c9"}, # read-back OK
]
@@ -761,6 +785,7 @@ class TestMergePR(unittest.TestCase):
def test_output_redacts_secrets(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
*self._feedback_reads(),
{}, {"merged_commit_sha": "c1"},
]
env = {"GITEA_PROFILE_NAME": "gitea-merger",
@@ -778,6 +803,7 @@ class TestMergePR(unittest.TestCase):
def test_merge_error_message_redacts_credential(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
*self._feedback_reads(),
RuntimeError("HTTP 500: token abc-secret-xyz rejected"),
]
env = {"GITEA_PROFILE_NAME": "gitea-merger",
@@ -790,6 +816,45 @@ class TestMergePR(unittest.TestCase):
self.assertIn("[REDACTED]", blob)
self.assertNotIn("abc-secret-xyz", blob)
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_merge_blocked_without_visible_approval(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
self._pr("author-bot"),
[_formal_review("sysadmin", "PENDING")],
]
env = {"GITEA_PROFILE_NAME": "gitea-merger",
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8), remote="prgs")
self.assertFalse(r["performed"])
self.assertFalse(r.get("approval_visible"))
self.assertTrue(any("no visible APPROVED review" in x for x in r["reasons"]))
self._assert_no_merge_call(mock_api)
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_merge_blocked_on_request_changes(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
self._pr("author-bot"),
[
_formal_review("reviewer-bot", "APPROVED"),
_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=2),
],
]
env = {"GITEA_PROFILE_NAME": "gitea-merger",
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8), remote="prgs")
self.assertFalse(r["performed"])
self.assertTrue(r.get("has_blocking_change_requests"))
self.assertTrue(any("REQUEST_CHANGES" in x for x in r["reasons"]))
self._assert_no_merge_call(mock_api)
class TestNoUngatedMergePath(unittest.TestCase):
"""Prove no other exposed tool can merge (#16 surface audit)."""
@@ -1545,7 +1610,9 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_duplicate_terminal_decision_blocked(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 1},
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 1, "state": "APPROVED"},
[_formal_review("reviewer-bot", "APPROVED", review_id=1)],
{"login": "reviewer-bot"}, self._pr("author-bot"),
]
gitea_mark_final_review_decision(
@@ -1628,7 +1695,9 @@ class TestSubmitPrReview(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_approve_succeeds_when_eligible(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 7},
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 7, "state": "APPROVED"},
[_formal_review("reviewer-bot", "APPROVED", review_id=7)],
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
@@ -1638,16 +1707,63 @@ class TestSubmitPrReview(unittest.TestCase):
final_review_decision_ready=True,
)
self.assertTrue(r["performed"])
self.assertTrue(r.get("review_verdict_visible"))
self.assertEqual(r["authenticated_user"], "reviewer-bot")
self.assertEqual(r["pr_author"], "author-bot")
self.assertEqual(r["head_sha"], "abc123")
method, url = mock_api.call_args.args[0], mock_api.call_args.args[1]
self.assertEqual(method, "POST")
self.assertTrue(url.endswith("/pulls/8/reviews"))
payload = mock_api.call_args.args[3]
self.assertEqual(payload["event"], "APPROVE")
post_calls = [
c for c in mock_api.call_args_list
if c.args[0] == "POST" and c.args[1].endswith("/pulls/8/reviews")
]
self.assertEqual(len(post_calls), 1)
payload = post_calls[0].args[3]
self.assertEqual(payload["event"], "APPROVED")
self.assertEqual(payload["commit_id"], "abc123")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_approve_fails_when_verdict_stays_pending(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 7, "state": "PENDING"},
{"id": 7, "state": "PENDING"},
[_formal_review("reviewer-bot", "PENDING", review_id=7)],
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(
pr_number=8, action="approve", body="LGTM", remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertFalse(r.get("review_verdict_visible"))
self.assertTrue(any("expected visible 'APPROVED'" in x for x in r["reasons"]))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_approve_submits_pending_draft_when_api_returns_pending(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 7, "state": "PENDING"},
{"id": 7, "state": "APPROVED"},
[_formal_review("reviewer-bot", "APPROVED", review_id=7)],
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(
pr_number=8, action="approve", body="LGTM", remote="prgs",
final_review_decision_ready=True,
)
self.assertTrue(r["performed"])
submit_calls = [
c for c in mock_api.call_args_list
if c.args[0] == "POST" and "/reviews/7" in c.args[1]
]
self.assertEqual(len(submit_calls), 1)
self.assertEqual(submit_calls[0].args[3]["event"], "APPROVED")
# -- request_changes ------------------------------------------------------
@patch("mcp_server.api_request")
@@ -1655,7 +1771,9 @@ class TestSubmitPrReview(unittest.TestCase):
def test_request_changes_succeeds_when_eligible(self, _auth, mock_api):
gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 9},
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 9, "state": "REQUEST_CHANGES"},
[_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=9)],
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,request_changes"}
@@ -1666,7 +1784,12 @@ class TestSubmitPrReview(unittest.TestCase):
final_review_decision_ready=True,
)
self.assertTrue(r["performed"])
self.assertEqual(mock_api.call_args.args[3]["event"], "REQUEST_CHANGES")
post_calls = [
c for c in mock_api.call_args_list
if c.args[0] == "POST" and c.args[1].endswith("/pulls/8/reviews")
]
self.assertEqual(len(post_calls), 1)
self.assertEqual(post_calls[0].args[3]["event"], "REQUEST_CHANGES")
def test_request_changes_blocked_without_eligibility(self):
gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
@@ -1777,7 +1900,8 @@ class TestSubmitPrReview(unittest.TestCase):
patch("mcp_server.api_request") as mock_api:
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot", sha="abc123"),
{"id": 5},
{"id": 5, "state": "APPROVED"},
[_formal_review("reviewer-bot", "APPROVED", review_id=5)],
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
@@ -1929,8 +2053,12 @@ class TestSubmitPrReview(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_correction_flow_allows_second_terminal_review(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 42},
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 43},
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 42, "state": "APPROVED"},
[_formal_review("reviewer-bot", "APPROVED", review_id=42)],
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 43, "state": "REQUEST_CHANGES"},
[_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=43)],
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes"}
@@ -2003,7 +2131,7 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
patch("gitea_audit.audit_enabled", return_value=True).start()
self.mock_audit = patch("gitea_audit.write_event").start()
# gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216).
patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()
patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["read", "merge", "edit", "close", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()
def tearDown(self):
patch.stopall()
@@ -2051,6 +2179,8 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and "/user" in url:
return {"login": "merger"}
if method == "GET" and url.endswith("/reviews"):
return [_formal_review("reviewer", "APPROVED", sha="sha123")]
if method == "GET" and "pulls/1" in url and "/files" not in url:
return {
"user": {"login": "author"},
@@ -2083,6 +2213,8 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and "/user" in url:
return {"login": "merger"}
if method == "GET" and url.endswith("/reviews"):
return [_formal_review("reviewer", "APPROVED", sha="sha123")]
if method == "GET" and "pulls/1" in url and "/files" not in url:
return {
"user": {"login": "author"},