sysadminandClaude Opus 4.8
89582a0f2a
Add merge_approval_gate assessment so gitea_merge_pr fails closed when visible APPROVED reviews do not pin the live head SHA. Expose approval_at_current_head and stale_approval_block_reason in review feedback. Document re-review requirement in canonical merge workflow. Closes #471. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
475 lines
21 KiB
Python
475 lines
21 KiB
Python
"""Tests for issue #167: LLM role/task alignment and PR review feedback discovery.
|
|
|
|
Covers:
|
|
- gitea_get_pr_review_feedback: MCP-native discovery of formal PR review
|
|
verdicts (APPROVED / REQUEST_CHANGES / COMMENT) without reading issue
|
|
comments, including blocking-state summary, staleness against the current
|
|
head SHA, redaction, and fail-closed permission gating.
|
|
- gitea_resolve_task_capability role guards: a review/merge task under an
|
|
author profile returns explicit stop guidance; author-side remediation is
|
|
distinguished as its own task kind.
|
|
- build_validation_report: validation reporting distinguishes passed,
|
|
failed, skipped, and not-run commands and never hides failures behind
|
|
vague language.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
import gitea_config
|
|
import mcp_server
|
|
from mcp_server import (
|
|
build_validation_report,
|
|
gitea_get_pr_review_feedback,
|
|
gitea_resolve_task_capability,
|
|
)
|
|
|
|
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
|
|
|
|
|
|
def _pr_details(head_sha="headsha2", state="open", author="author-user"):
|
|
return {
|
|
"number": 5,
|
|
"title": "PR 5",
|
|
"state": state,
|
|
"head": {"ref": "feature", "sha": head_sha},
|
|
"base": {"ref": "master"},
|
|
"mergeable": True,
|
|
"user": {"login": author},
|
|
}
|
|
|
|
|
|
def _review(reviewer, state, commit_id="headsha2", body="", submitted_at="2026-07-05T10:00:00Z",
|
|
dismissed=False, review_id=1):
|
|
return {
|
|
"id": review_id,
|
|
"user": {"login": reviewer},
|
|
"state": state,
|
|
"body": body,
|
|
"submitted_at": submitted_at,
|
|
"commit_id": commit_id,
|
|
"dismissed": dismissed,
|
|
"stale": False,
|
|
"html_url": "https://internal.example/pr/5#review",
|
|
}
|
|
|
|
|
|
class TestPRReviewFeedbackDiscovery(unittest.TestCase):
|
|
"""Formal review feedback must be discoverable through MCP tools."""
|
|
|
|
def _profile(self, allowed=("gitea.read",)):
|
|
return {
|
|
"profile_name": "gitea-author",
|
|
"allowed_operations": list(allowed),
|
|
"forbidden_operations": [],
|
|
"base_url": None,
|
|
}
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
@patch("mcp_server.get_profile")
|
|
def test_review_feedback_discoverable_without_issue_comments(self, mock_get_profile, _auth, mock_api):
|
|
mock_get_profile.return_value = self._profile()
|
|
mock_api.side_effect = [
|
|
_pr_details(),
|
|
[_review("reviewer1", "REQUEST_CHANGES", body="Please add tests.")],
|
|
]
|
|
result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs")
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(result["pr_number"], 5)
|
|
self.assertEqual(len(result["reviews"]), 1)
|
|
review = result["reviews"][0]
|
|
self.assertEqual(review["reviewer"], "reviewer1")
|
|
self.assertEqual(review["verdict"], "REQUEST_CHANGES")
|
|
self.assertEqual(review["body"], "Please add tests.")
|
|
self.assertIn("submitted_at", review)
|
|
self.assertTrue(result["has_blocking_change_requests"])
|
|
self.assertFalse(result["approval_visible"])
|
|
self.assertEqual(
|
|
result["latest_review_state_by_reviewer"], {"reviewer1": "REQUEST_CHANGES"})
|
|
# only GET calls; discovery never mutates
|
|
for call in mock_api.call_args_list:
|
|
self.assertEqual(call.args[0], "GET")
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
@patch("mcp_server.get_profile")
|
|
def test_latest_state_per_reviewer_and_approval_visible(self, mock_get_profile, _auth, mock_api):
|
|
mock_get_profile.return_value = self._profile()
|
|
mock_api.side_effect = [
|
|
_pr_details(),
|
|
[
|
|
_review("reviewer1", "REQUEST_CHANGES",
|
|
submitted_at="2026-07-05T10:00:00Z", review_id=1),
|
|
_review("reviewer1", "APPROVED",
|
|
submitted_at="2026-07-05T11:00:00Z", review_id=2),
|
|
],
|
|
]
|
|
result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs")
|
|
self.assertEqual(
|
|
result["latest_review_state_by_reviewer"], {"reviewer1": "APPROVED"})
|
|
self.assertFalse(result["has_blocking_change_requests"])
|
|
self.assertTrue(result["approval_visible"])
|
|
self.assertTrue(result["approval_at_current_head"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
@patch("mcp_server.get_profile")
|
|
def test_stale_approval_not_at_current_head(self, mock_get_profile, _auth, mock_api):
|
|
mock_get_profile.return_value = self._profile()
|
|
mock_api.side_effect = [
|
|
_pr_details(head_sha="newhead3"),
|
|
[_review("reviewer1", "APPROVED", commit_id="oldhead1")],
|
|
]
|
|
result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs")
|
|
self.assertTrue(result["approval_visible"])
|
|
self.assertFalse(result["approval_at_current_head"])
|
|
self.assertEqual(result["latest_approved_head_sha"], "oldhead1")
|
|
self.assertIn("stale approval", result["stale_approval_block_reason"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
@patch("mcp_server.get_profile")
|
|
def test_change_request_staleness_after_new_commits(self, mock_get_profile, _auth, mock_api):
|
|
mock_get_profile.return_value = self._profile()
|
|
mock_api.side_effect = [
|
|
_pr_details(head_sha="newhead3"),
|
|
[_review("reviewer1", "REQUEST_CHANGES", commit_id="oldhead1")],
|
|
]
|
|
result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs")
|
|
self.assertEqual(result["current_head_sha"], "newhead3")
|
|
self.assertEqual(result["latest_reviewed_head_sha"], "oldhead1")
|
|
self.assertTrue(result["review_feedback_stale"])
|
|
self.assertTrue(result["author_pushed_after_request_changes"])
|
|
self.assertEqual(result["reviews"][0]["reviewed_head_sha"], "oldhead1")
|
|
self.assertTrue(result["reviews"][0]["stale"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
@patch("mcp_server.get_profile")
|
|
def test_feedback_current_when_review_matches_head(self, mock_get_profile, _auth, mock_api):
|
|
mock_get_profile.return_value = self._profile()
|
|
mock_api.side_effect = [
|
|
_pr_details(head_sha="headsha2"),
|
|
[_review("reviewer1", "REQUEST_CHANGES", commit_id="headsha2")],
|
|
]
|
|
result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs")
|
|
self.assertFalse(result["review_feedback_stale"])
|
|
self.assertFalse(result["author_pushed_after_request_changes"])
|
|
self.assertTrue(result["has_blocking_change_requests"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
@patch("mcp_server.get_profile")
|
|
def test_dismissed_change_requests_not_blocking(self, mock_get_profile, _auth, mock_api):
|
|
mock_get_profile.return_value = self._profile()
|
|
mock_api.side_effect = [
|
|
_pr_details(),
|
|
[_review("reviewer1", "REQUEST_CHANGES", dismissed=True)],
|
|
]
|
|
result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs")
|
|
self.assertFalse(result["has_blocking_change_requests"])
|
|
self.assertTrue(result["reviews"][0]["dismissed"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
@patch("mcp_server.get_profile")
|
|
def test_later_comment_does_not_mask_change_request_staleness(self, mock_get_profile, _auth, mock_api):
|
|
mock_get_profile.return_value = self._profile()
|
|
mock_api.side_effect = [
|
|
_pr_details(head_sha="newhead3"),
|
|
[
|
|
_review("reviewer1", "REQUEST_CHANGES", commit_id="oldhead1",
|
|
submitted_at="2026-07-05T10:00:00Z", review_id=1),
|
|
_review("reviewer2", "COMMENT", commit_id="newhead3",
|
|
submitted_at="2026-07-05T11:00:00Z", review_id=2),
|
|
],
|
|
]
|
|
result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs")
|
|
self.assertEqual(result["latest_reviewed_head_sha"], "oldhead1")
|
|
self.assertTrue(result["review_feedback_stale"])
|
|
self.assertTrue(result["author_pushed_after_request_changes"])
|
|
self.assertTrue(result["has_blocking_change_requests"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
@patch("mcp_server.get_profile")
|
|
def test_dismissed_approval_not_visible(self, mock_get_profile, _auth, mock_api):
|
|
mock_get_profile.return_value = self._profile()
|
|
mock_api.side_effect = [
|
|
_pr_details(),
|
|
[_review("reviewer1", "APPROVED", dismissed=True)],
|
|
]
|
|
result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs")
|
|
self.assertFalse(result["approval_visible"])
|
|
self.assertTrue(result["reviews"][0]["dismissed"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
@patch("mcp_server.get_profile")
|
|
def test_pending_reviews_do_not_set_latest_state(self, mock_get_profile, _auth, mock_api):
|
|
mock_get_profile.return_value = self._profile()
|
|
mock_api.side_effect = [
|
|
_pr_details(),
|
|
[
|
|
_review("reviewer1", "APPROVED",
|
|
submitted_at="2026-07-05T10:00:00Z", review_id=1),
|
|
_review("reviewer1", "PENDING",
|
|
submitted_at="2026-07-05T11:00:00Z", review_id=2),
|
|
],
|
|
]
|
|
result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs")
|
|
self.assertEqual(
|
|
result["latest_review_state_by_reviewer"], {"reviewer1": "APPROVED"})
|
|
self.assertTrue(result["approval_visible"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
@patch("mcp_server.get_profile")
|
|
def test_review_bodies_are_redacted(self, mock_get_profile, _auth, mock_api):
|
|
mock_get_profile.return_value = self._profile()
|
|
mock_api.side_effect = [
|
|
_pr_details(),
|
|
[_review("reviewer1", "COMMENT",
|
|
body="failure mentions token supersecret123 in log")],
|
|
]
|
|
result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs")
|
|
dumped = json.dumps(result)
|
|
self.assertNotIn("supersecret123", dumped)
|
|
self.assertIn("[REDACTED]", result["reviews"][0]["body"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
@patch("mcp_server.get_profile")
|
|
def test_no_url_leak_without_reveal_opt_in(self, mock_get_profile, _auth, mock_api):
|
|
mock_get_profile.return_value = self._profile()
|
|
mock_api.side_effect = [
|
|
_pr_details(),
|
|
[_review("reviewer1", "APPROVED")],
|
|
]
|
|
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": ""}):
|
|
result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs")
|
|
dumped = json.dumps(result)
|
|
self.assertNotIn("internal.example", dumped)
|
|
self.assertNotIn("url", result["reviews"][0])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
@patch("mcp_server.get_profile")
|
|
def test_permission_block_fails_closed_without_api_calls(self, mock_get_profile, _auth, mock_api):
|
|
mock_get_profile.return_value = self._profile(allowed=())
|
|
result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs")
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(result["feedback_not_attempted"])
|
|
self.assertNotIn("reviews", result)
|
|
self.assertIn("permission_report", result)
|
|
self.assertTrue(result["reasons"])
|
|
self.assertEqual(mock_api.call_count, 0)
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
@patch("mcp_server.get_profile")
|
|
def test_no_reviews_is_distinct_from_not_attempted(self, mock_get_profile, _auth, mock_api):
|
|
mock_get_profile.return_value = self._profile()
|
|
mock_api.side_effect = [_pr_details(), []]
|
|
result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs")
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(result["reviews"], [])
|
|
self.assertFalse(result.get("feedback_not_attempted", False))
|
|
self.assertFalse(result["has_blocking_change_requests"])
|
|
self.assertFalse(result["approval_visible"])
|
|
|
|
|
|
CONFIG_GUARD = {
|
|
"version": 2,
|
|
"contexts": {
|
|
"ctx": {
|
|
"enabled": True,
|
|
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
|
|
}
|
|
},
|
|
"profiles": {
|
|
"author-profile": {
|
|
"enabled": True,
|
|
"context": "ctx",
|
|
"role": "author",
|
|
"username": "author-user",
|
|
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
|
"allowed_operations": [
|
|
"gitea.read", "gitea.issue.create", "gitea.pr.create",
|
|
"gitea.branch.push", "gitea.issue.comment",
|
|
],
|
|
"forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
|
|
"execution_profile": "author-profile",
|
|
},
|
|
"reviewer-profile": {
|
|
"enabled": True,
|
|
"context": "ctx",
|
|
"role": "reviewer",
|
|
"username": "reviewer-user",
|
|
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
|
|
"allowed_operations": [
|
|
"gitea.read", "gitea.pr.review", "gitea.pr.approve",
|
|
"gitea.pr.merge",
|
|
],
|
|
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
|
|
"execution_profile": "reviewer-profile",
|
|
},
|
|
},
|
|
"rules": {"allow_runtime_switching": False},
|
|
}
|
|
|
|
|
|
class TestTaskRoleGuards(unittest.TestCase):
|
|
"""Review/merge tasks under an author profile must return stop guidance."""
|
|
|
|
def setUp(self):
|
|
self._remotes_patch = patch.dict(mcp_server.REMOTES, {
|
|
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
|
|
"repo": "Example-Repo"},
|
|
})
|
|
self._remotes_patch.start()
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
gitea_config._active_profile_override = None
|
|
self._dir = tempfile.TemporaryDirectory()
|
|
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
|
with open(self.config_path, "w", encoding="utf-8") as fh:
|
|
fh.write(json.dumps(CONFIG_GUARD))
|
|
|
|
def tearDown(self):
|
|
self._remotes_patch.stop()
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
gitea_config._active_profile_override = None
|
|
self._dir.cleanup()
|
|
|
|
def _env(self, profile):
|
|
return {
|
|
"GITEA_MCP_CONFIG": self.config_path,
|
|
"GITEA_MCP_PROFILE": profile,
|
|
"GITEA_TOKEN_AUTHOR": "author-pass",
|
|
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
|
|
}
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_review_task_under_author_profile_returns_stop_guidance(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = gitea_resolve_task_capability(task="review_pr", remote="prgs")
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertTrue(res["stop_required"])
|
|
guidance = " ".join(res["task_role_guidance"])
|
|
self.assertIn("STOP", guidance)
|
|
self.assertIn("author", guidance.lower())
|
|
# must forbid author-side mutations while the task is review/merge
|
|
self.assertIn("commit", guidance.lower())
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_merge_task_under_author_profile_returns_stop_guidance(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = gitea_resolve_task_capability(task="merge_pr", remote="prgs")
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertTrue(res["stop_required"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_author_fix_task_is_distinct_and_allowed_for_author(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("author-profile")):
|
|
res = gitea_resolve_task_capability(
|
|
task="address_pr_change_requests", remote="prgs")
|
|
self.assertEqual(res["required_role_kind"], "author")
|
|
self.assertTrue(res["allowed_in_current_session"])
|
|
self.assertFalse(res["stop_required"])
|
|
guidance = " ".join(res["task_role_guidance"])
|
|
self.assertIn("must not submit review verdicts", guidance.lower())
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
|
def test_review_task_under_reviewer_profile_allowed(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("reviewer-profile")):
|
|
res = gitea_resolve_task_capability(task="review_pr", remote="prgs")
|
|
self.assertTrue(res["allowed_in_current_session"])
|
|
self.assertFalse(res["stop_required"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
|
|
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
|
def test_author_fix_task_under_reviewer_profile_blocked(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("reviewer-profile")):
|
|
res = gitea_resolve_task_capability(
|
|
task="address_pr_change_requests", remote="prgs")
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertTrue(res["stop_required"])
|
|
|
|
|
|
class TestValidationReport(unittest.TestCase):
|
|
"""Validation reporting must not hide failed or skipped checks."""
|
|
|
|
def test_distinguishes_passed_failed_skipped_not_run(self):
|
|
report = build_validation_report([
|
|
{"command": "python3 -m py_compile mcp_server.py", "status": "passed"},
|
|
{"command": "python3 -m unittest discover -s tests", "status": "failed",
|
|
"output": "FAILED (failures=1)"},
|
|
{"command": "git diff --check", "status": "skipped",
|
|
"reason": "replaced by targeted check"},
|
|
{"command": "secret sweep", "status": "not-run"},
|
|
])
|
|
self.assertFalse(report["all_passed"])
|
|
statuses = {e["command"]: e["status"] for e in report["entries"]}
|
|
self.assertEqual(statuses["python3 -m py_compile mcp_server.py"], "passed")
|
|
self.assertEqual(statuses["python3 -m unittest discover -s tests"], "failed")
|
|
self.assertEqual(statuses["git diff --check"], "skipped")
|
|
self.assertEqual(statuses["secret sweep"], "not-run")
|
|
summary = report["summary"]
|
|
self.assertIn("FAILED", summary)
|
|
self.assertIn("FAILED (failures=1)", summary)
|
|
self.assertIn("SKIPPED", summary)
|
|
self.assertIn("NOT-RUN", summary)
|
|
|
|
def test_failed_entry_requires_exact_output(self):
|
|
with self.assertRaises(ValueError):
|
|
build_validation_report([
|
|
{"command": "unittest", "status": "failed"},
|
|
])
|
|
|
|
def test_unknown_status_fails_closed(self):
|
|
with self.assertRaises(ValueError):
|
|
build_validation_report([
|
|
{"command": "unittest", "status": "shell-invocation quirks"},
|
|
])
|
|
|
|
def test_missing_command_fails_closed(self):
|
|
with self.assertRaises(ValueError):
|
|
build_validation_report([{"status": "passed"}])
|
|
|
|
def test_full_suite_success_not_implied(self):
|
|
report = build_validation_report([
|
|
{"command": "python3 -m unittest discover -s tests",
|
|
"status": "not-run", "full_suite": True},
|
|
{"command": "targeted tests", "status": "passed"},
|
|
])
|
|
self.assertFalse(report["all_passed"])
|
|
self.assertIs(report["full_suite_passed"], False)
|
|
|
|
def test_full_suite_passed_when_run_and_green(self):
|
|
report = build_validation_report([
|
|
{"command": "python3 -m unittest discover -s tests",
|
|
"status": "passed", "full_suite": True},
|
|
])
|
|
self.assertTrue(report["all_passed"])
|
|
self.assertIs(report["full_suite_passed"], True)
|
|
|
|
def test_full_suite_none_when_no_full_suite_entry(self):
|
|
report = build_validation_report([
|
|
{"command": "targeted tests", "status": "passed"},
|
|
])
|
|
self.assertIsNone(report["full_suite_passed"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|