feat: add gated gitea_submit_pr_review review actions (#15)
Add gitea_submit_pr_review, the only tool that submits a Gitea PR review. It performs a review mutation (comment / approve / request_changes) only after every safety gate passes, and never merges. Gates (fail-closed at each step): 1. Validate action is comment | approve | request_changes. 2. Reuse gitea_check_pr_eligibility (#14) for authenticated-user lookup, active-profile lookup, PR-author lookup, self-approval block, and the profile-allowed-operation check. approve requires 'approve' eligibility, request_changes requires 'request_changes', comment requires 'review'. 3. Redundant self-approval block (auth user == PR author). 4. Optional expected_head_sha: refuse if the PR head has moved. 5. Only then POST /repos/{owner}/{repo}/pulls/{n}/reviews (formal review endpoint, so approvals/change-requests carry real review state). Output reports action, whether performed, authenticated user, profile name, PR author, PR number, head SHA checked, and reasons — never a token, auth header, or credential. Error text is scrubbed via _redact as defence in depth. Merge is intentionally not implemented (belongs to #16). Tests cover: self-author approve blocked, approve/request_changes/comment succeed only when eligible, unknown identity fail-closed, disallowed profile op blocked, head-SHA mismatch blocked, no mutation when gates fail, invalid action rejected, and secret redaction in output and error paths. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -29,6 +29,7 @@ from mcp_server import ( # noqa: E402
|
||||
gitea_whoami,
|
||||
gitea_get_profile,
|
||||
gitea_check_pr_eligibility,
|
||||
gitea_submit_pr_review,
|
||||
)
|
||||
from gitea_auth import get_profile # noqa: E402
|
||||
|
||||
@@ -785,5 +786,229 @@ class TestPrEligibility(unittest.TestCase):
|
||||
self.assertNotIn(secret, blob)
|
||||
|
||||
|
||||
class TestSubmitPrReview(unittest.TestCase):
|
||||
"""Gated review-mutation tool (#15)."""
|
||||
|
||||
def _pr(self, author, state="open", sha="abc123", mergeable=True):
|
||||
return {
|
||||
"user": {"login": author},
|
||||
"state": state,
|
||||
"head": {"sha": sha},
|
||||
"mergeable": mergeable,
|
||||
}
|
||||
|
||||
def _methods(self, mock_api):
|
||||
return [c.args[0] for c in mock_api.call_args_list]
|
||||
|
||||
def _assert_no_mutation(self, mock_api):
|
||||
# A review mutation is POST .../reviews; eligibility only ever GETs.
|
||||
for c in mock_api.call_args_list:
|
||||
method, url = c.args[0], c.args[1]
|
||||
self.assertFalse(
|
||||
method == "POST" and url.endswith("/reviews"),
|
||||
f"unexpected review mutation: {method} {url}",
|
||||
)
|
||||
|
||||
# -- approve --------------------------------------------------------------
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_approve_blocked_when_author(self, _auth, mock_api):
|
||||
mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")]
|
||||
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", remote="prgs")
|
||||
self.assertFalse(r["performed"])
|
||||
self.assertIn("authenticated user is PR author", r["reasons"])
|
||||
self._assert_no_mutation(mock_api)
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@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},
|
||||
]
|
||||
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")
|
||||
self.assertTrue(r["performed"])
|
||||
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")
|
||||
self.assertEqual(payload["commit_id"], "abc123")
|
||||
|
||||
# -- request_changes ------------------------------------------------------
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_request_changes_succeeds_when_eligible(self, _auth, mock_api):
|
||||
mock_api.side_effect = [
|
||||
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 9},
|
||||
]
|
||||
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
|
||||
"GITEA_ALLOWED_OPERATIONS": "read,review,request_changes"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
r = gitea_submit_pr_review(
|
||||
pr_number=8, action="request_changes",
|
||||
body="needs work", remote="prgs")
|
||||
self.assertTrue(r["performed"])
|
||||
self.assertEqual(mock_api.call_args.args[3]["event"], "REQUEST_CHANGES")
|
||||
|
||||
def test_request_changes_blocked_without_eligibility(self):
|
||||
with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) as _a, \
|
||||
patch("mcp_server.api_request") as mock_api:
|
||||
mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
|
||||
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
|
||||
"GITEA_ALLOWED_OPERATIONS": "read,review"} # no request_changes
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
r = gitea_submit_pr_review(
|
||||
pr_number=8, action="request_changes", remote="prgs")
|
||||
self.assertFalse(r["performed"])
|
||||
self.assertIn("profile is not allowed to request_changes", r["reasons"])
|
||||
self._assert_no_mutation(mock_api)
|
||||
|
||||
# -- comment --------------------------------------------------------------
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_comment_succeeds_when_review_eligible(self, _auth, mock_api):
|
||||
mock_api.side_effect = [
|
||||
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 3},
|
||||
]
|
||||
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
|
||||
"GITEA_ALLOWED_OPERATIONS": "read,review"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
r = gitea_submit_pr_review(
|
||||
pr_number=8, action="comment", body="finding", remote="prgs")
|
||||
self.assertTrue(r["performed"])
|
||||
self.assertEqual(mock_api.call_args.args[3]["event"], "COMMENT")
|
||||
|
||||
def test_comment_by_author_allowed(self):
|
||||
# Commenting on your own PR is fine — only approve is self-blocked.
|
||||
with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
|
||||
patch("mcp_server.api_request") as mock_api:
|
||||
mock_api.side_effect = [
|
||||
{"login": "jcwalker3"}, self._pr("jcwalker3"), {"id": 4},
|
||||
]
|
||||
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
|
||||
"GITEA_ALLOWED_OPERATIONS": "read,review"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
r = gitea_submit_pr_review(
|
||||
pr_number=8, action="comment", body="note", remote="prgs")
|
||||
self.assertTrue(r["performed"])
|
||||
|
||||
# -- identity / profile fail-closed ---------------------------------------
|
||||
|
||||
@patch("mcp_server.get_auth_header", return_value=None)
|
||||
def test_unknown_identity_blocks(self, _auth):
|
||||
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", remote="prgs")
|
||||
self.assertFalse(r["performed"])
|
||||
self.assertIsNone(r["authenticated_user"])
|
||||
self.assertIn("authenticated identity could not be determined", r["reasons"])
|
||||
|
||||
def test_disallowed_profile_operation_blocks(self):
|
||||
with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
|
||||
patch("mcp_server.api_request") as mock_api:
|
||||
mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
|
||||
env = {"GITEA_PROFILE_NAME": "gitea-author",
|
||||
"GITEA_ALLOWED_OPERATIONS": "read,pr.create"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
r = gitea_submit_pr_review(
|
||||
pr_number=8, action="approve", remote="prgs")
|
||||
self.assertFalse(r["performed"])
|
||||
self.assertIn("profile is not allowed to approve", r["reasons"])
|
||||
self._assert_no_mutation(mock_api)
|
||||
|
||||
# -- head SHA guard -------------------------------------------------------
|
||||
|
||||
def test_head_sha_mismatch_blocks(self):
|
||||
with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
|
||||
patch("mcp_server.api_request") as mock_api:
|
||||
mock_api.side_effect = [
|
||||
{"login": "reviewer-bot"}, self._pr("author-bot", sha="abc123"),
|
||||
]
|
||||
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",
|
||||
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._assert_no_mutation(mock_api)
|
||||
|
||||
def test_head_sha_match_allows(self):
|
||||
with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
|
||||
patch("mcp_server.api_request") as mock_api:
|
||||
mock_api.side_effect = [
|
||||
{"login": "reviewer-bot"}, self._pr("author-bot", sha="abc123"),
|
||||
{"id": 5},
|
||||
]
|
||||
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",
|
||||
expected_head_sha="abc123", remote="prgs")
|
||||
self.assertTrue(r["performed"])
|
||||
|
||||
# -- invalid action -------------------------------------------------------
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
def test_invalid_review_action_rejected(self, mock_api):
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
r = gitea_submit_pr_review(pr_number=8, action="delete", remote="prgs")
|
||||
self.assertFalse(r["performed"])
|
||||
self.assertTrue(any("unknown review action" in x for x in r["reasons"]))
|
||||
mock_api.assert_not_called() # never touches the API on a bad action
|
||||
|
||||
# -- redaction ------------------------------------------------------------
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_output_redacts_secrets(self, _auth, mock_api):
|
||||
mock_api.side_effect = [
|
||||
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 1},
|
||||
]
|
||||
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
|
||||
"GITEA_ALLOWED_OPERATIONS": "read,review,approve",
|
||||
"GITEA_TOKEN": "super-secret-token"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
r = gitea_submit_pr_review(pr_number=5, action="approve", remote="prgs")
|
||||
blob = repr(r).lower()
|
||||
for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
|
||||
self.assertNotIn(secret, blob)
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_error_message_redacts_credential(self, _auth, mock_api):
|
||||
mock_api.side_effect = [
|
||||
{"login": "reviewer-bot"},
|
||||
self._pr("author-bot"),
|
||||
RuntimeError("HTTP 500: token abc-secret-xyz rejected"),
|
||||
]
|
||||
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=5, action="approve", remote="prgs")
|
||||
self.assertFalse(r["performed"])
|
||||
blob = repr(r)
|
||||
self.assertIn("[REDACTED]", blob)
|
||||
self.assertNotIn("abc-secret-xyz", blob)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user