feat: harden LLM role/task alignment and PR review feedback discovery (#167)
- Add gitea_get_pr_review_feedback: read-only MCP-native discovery of formal PR review verdicts (APPROVED / REQUEST_CHANGES / COMMENT) with latest-state-per-reviewer, blocking change-request and approval summaries, reviewed-head vs current-head staleness, dismissed/PENDING handling, credential redaction of review bodies, no endpoint URLs without the reveal opt-in, and a fail-closed gitea.read gate that returns feedback_not_attempted with a structured #142 permission report and zero API calls. - Add task/role guards to gitea_resolve_task_capability: new stop_required and task_role_guidance outputs, and a distinct address_pr_change_requests author task, so a review/merge request under an author profile returns explicit STOP guidance instead of silently degrading into author-side mutations, and author-side remediation is explicitly barred from review verdicts and merges. - Add build_validation_report: validation reporting helper that distinguishes passed / failed / skipped / not-run, requires the exact output for failures, rejects vague statuses, and never implies full-suite success unless the full-suite command passed. - Document task/role alignment (review vs author-fix vs merge vs comment workflows with required identity, allowed/forbidden actions, and stop conditions), MCP-native review feedback discovery, and validation reporting discipline in docs/llm-workflow-runbooks.md. - Add tests/test_review_feedback.py (22 tests) covering discovery, staleness, redaction, URL hiding, fail-closed gating, role guards for author/reviewer profiles, and validation report semantics. Closes #167 Co-Authored-By: Claude Fable 5 <[email protected]>
This commit is contained in:
Claude Fable 5
@@ -379,6 +379,69 @@ touching anything.
|
||||
files, detected secret, or any production/deploy behavior — **stop, report the
|
||||
blocker, and take no mutating action.** Fail closed; never work around a gate.
|
||||
|
||||
## Task/role alignment (#167)
|
||||
|
||||
The **requested task** decides what a session may do — not the credential it
|
||||
happens to hold. Resolve the task first with
|
||||
`gitea_resolve_task_capability(task=...)`: it returns `stop_required` and
|
||||
`task_role_guidance` alongside the permission decision. An LLM asked to
|
||||
review must never degrade into author work just because it is connected as
|
||||
an author.
|
||||
|
||||
| Requested task | Required identity/profile | Allowed | Forbidden | Stop when |
|
||||
|---|---|---|---|---|
|
||||
| Review PR (`review_pr`) | reviewer (e.g. `sysadmin` / `prgs-reviewer`) | read, gated review verdicts | commits, pushes, file edits, author comments, merge without eligibility | active profile is an author profile — stop immediately; do **not** switch to author-side fixes unless the operator explicitly re-tasks |
|
||||
| Address PR change requests (`address_pr_change_requests`) | author (e.g. `jcwalker3` / `prgs-author`) | commit/push fixes to the PR branch, PR comment summarizing fixes | review verdicts, approve, request-changes, merge | active profile lacks branch push |
|
||||
| Merge PR (`merge_pr`) | reviewer/merger | gated merge after eligibility + approval | merging own PR, merging without pinned head match | active profile is an author profile, or any merge gate fails |
|
||||
| Comment on issue discussion (`comment_issue`) | any profile with `gitea.issue.comment` | issue thread comments | review verdicts, closing via comment | permission missing (`gitea.pr.comment` does **not** imply it) |
|
||||
| Comment on PR (`comment_pr`) | any profile with `gitea.pr.comment` | PR thread comments | review verdicts | permission missing |
|
||||
| Author implementation (`create_branch`/`push_branch`/`create_pr`) | author | branch, commit, push, open PR | self-review, self-merge | profile lacks the author permissions |
|
||||
|
||||
If the task is review/merge and the session is an author profile, the only
|
||||
correct outputs are: the read-only PR queue inventory (#164), the structured
|
||||
permission report (#142), and a stop. Ask the operator to reconnect to the
|
||||
reviewer namespace; a credential or profile swap in the same session never
|
||||
cures same-session authorship.
|
||||
|
||||
## Review feedback discovery (MCP-native)
|
||||
|
||||
Formal review verdicts (APPROVED / REQUEST_CHANGES / COMMENT) live on the
|
||||
review endpoints, **not** in the issue-comment thread. Never infer review
|
||||
state from issue comments — use `gitea_get_pr_review_feedback(pr_number=...)`
|
||||
(read-only, requires `gitea.read`). It reports:
|
||||
|
||||
- every submitted review: reviewer, verdict, redacted body, timestamp, and
|
||||
the head SHA it reviewed;
|
||||
- `latest_review_state_by_reviewer` (PENDING drafts never count);
|
||||
- `has_blocking_change_requests` / `approval_visible` (dismissed reviews do
|
||||
not block);
|
||||
- `current_head_sha` vs `latest_reviewed_head_sha`, `review_feedback_stale`,
|
||||
and `author_pushed_after_request_changes` — so a reviewer can see whether
|
||||
feedback predates new commits, and an author can see whether fixes have
|
||||
been pushed since the REQUEST_CHANGES.
|
||||
|
||||
A permission block returns `feedback_not_attempted: true` with a structured
|
||||
permission report — distinct from a successful "no reviews yet" result, so a
|
||||
blocked lookup is never misread as "no feedback exists".
|
||||
|
||||
## Validation reporting discipline (#167)
|
||||
|
||||
Validation results in handoffs and PR bodies must state exactly what ran and
|
||||
what happened. Build the validation section with
|
||||
`build_validation_report(...)` (in `mcp_server.py`) or follow its contract by
|
||||
hand — every command is one of:
|
||||
|
||||
- `passed` — the exact command ran and succeeded;
|
||||
- `failed` — must include the exact command **and its output**; never
|
||||
paraphrase ("shell-invocation quirks" is not a status);
|
||||
- `skipped` — deliberately not run; name the reason and any targeted check
|
||||
that replaced it;
|
||||
- `not-run` — was required but never executed; say so plainly.
|
||||
|
||||
Never imply full-suite success unless the full-suite command itself passed
|
||||
(`full_suite_passed: true`). A report that hides a failed or skipped check
|
||||
is worse than a failing report.
|
||||
|
||||
## Controller Handoff (required, every task)
|
||||
|
||||
Every task — implementation, review, merge, triage, documentation,
|
||||
|
||||
+236
@@ -776,6 +776,139 @@ def _redact(text: str) -> str:
|
||||
return out
|
||||
|
||||
|
||||
# Review states that carry a submitted verdict. Gitea also emits PENDING
|
||||
# (draft) and REQUEST_REVIEW (re-review ask); neither is a verdict and
|
||||
# neither may drive the blocking/approval summaries.
|
||||
_VERDICT_STATES = ("APPROVED", "REQUEST_CHANGES", "COMMENT")
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_get_pr_review_feedback(
|
||||
pr_number: int,
|
||||
remote: str = "dadeschools",
|
||||
host: str | None = None,
|
||||
org: str | None = None,
|
||||
repo: str | None = None,
|
||||
) -> dict:
|
||||
"""Read-only: discover formal PR review feedback through MCP (#167).
|
||||
|
||||
Formal review verdicts (APPROVED / REQUEST_CHANGES / COMMENT) live on
|
||||
Gitea's review endpoints, not in the issue-comment thread. This tool is
|
||||
the MCP-native way to read them, so an LLM never has to infer review
|
||||
state from issue comments. It never submits reviews and never mutates
|
||||
anything; the profile must allow ``gitea.read`` (fail closed otherwise,
|
||||
with a structured #142 permission report and no API call made).
|
||||
|
||||
Endpoints: ``GET /repos/{owner}/{repo}/pulls/{n}`` (current head SHA)
|
||||
and ``GET /repos/{owner}/{repo}/pulls/{n}/reviews``.
|
||||
|
||||
Normal output is LLM-safe: review bodies pass through credential
|
||||
redaction and no endpoint URLs appear. Set GITEA_MCP_REVEAL_ENDPOINTS=1
|
||||
(admin/debug opt-in) to include each review's web link.
|
||||
|
||||
Args:
|
||||
pr_number: The pull request index/number.
|
||||
remote: Known instance — 'dadeschools' or 'prgs'.
|
||||
host: Override the Gitea host.
|
||||
org: Override the owner/organization.
|
||||
repo: Override the repository name.
|
||||
|
||||
Returns:
|
||||
dict with 'success', 'pr_number', 'pr_state', 'current_head_sha',
|
||||
'reviews' (each: reviewer, verdict, body, submitted_at,
|
||||
reviewed_head_sha, dismissed, stale),
|
||||
'latest_review_state_by_reviewer' (latest non-COMMENT verdict per
|
||||
reviewer; PENDING drafts never count),
|
||||
'has_blocking_change_requests' (a reviewer's latest verdict is an
|
||||
undismissed REQUEST_CHANGES), 'approval_visible',
|
||||
'latest_reviewed_head_sha' (head at the most recent
|
||||
APPROVED/REQUEST_CHANGES review; COMMENT-only reviews never
|
||||
advance it), 'review_feedback_stale' (that verdict feedback
|
||||
predates the current head SHA), and
|
||||
'author_pushed_after_request_changes' (new commits landed after a
|
||||
blocking REQUEST_CHANGES). On a permission block: 'success' False,
|
||||
'feedback_not_attempted' True, 'reasons', and 'permission_report' —
|
||||
deliberately distinct from a successful "no reviews yet" result.
|
||||
"""
|
||||
reasons = _issue_comment_gate("gitea.read")
|
||||
if reasons:
|
||||
return {
|
||||
"success": False,
|
||||
"pr_number": pr_number,
|
||||
"feedback_not_attempted": True,
|
||||
"reasons": reasons,
|
||||
"permission_report": _permission_block_report("gitea.read"),
|
||||
}
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
base = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
|
||||
pr = api_request("GET", base, auth) or {}
|
||||
raw_reviews = api_request("GET", f"{base}/reviews", auth) or []
|
||||
current_head = (pr.get("head") or {}).get("sha")
|
||||
reveal = _reveal_endpoints()
|
||||
|
||||
ordered = sorted(
|
||||
raw_reviews,
|
||||
key=lambda rv: ((rv.get("submitted_at") or ""), rv.get("id") or 0),
|
||||
)
|
||||
reviews = []
|
||||
latest_by_reviewer = {}
|
||||
latest_reviewed_head = None
|
||||
for rv in ordered:
|
||||
state = (rv.get("state") or "").upper()
|
||||
reviewer = (rv.get("user") or {}).get("login", "")
|
||||
commit_id = rv.get("commit_id")
|
||||
entry = {
|
||||
"reviewer": reviewer,
|
||||
"verdict": state,
|
||||
"body": _redact(rv.get("body") or ""),
|
||||
"submitted_at": rv.get("submitted_at"),
|
||||
"reviewed_head_sha": commit_id,
|
||||
"dismissed": bool(rv.get("dismissed")),
|
||||
"stale": bool(rv.get("stale")) or bool(
|
||||
commit_id and current_head and commit_id != current_head),
|
||||
}
|
||||
if reveal:
|
||||
entry["url"] = rv.get("html_url")
|
||||
reviews.append(entry)
|
||||
# COMMENT reviews never advance the reviewed-head marker or the
|
||||
# per-reviewer verdict — otherwise a drive-by comment on the
|
||||
# current head would mask the staleness of an older undismissed
|
||||
# REQUEST_CHANGES.
|
||||
if state in _VERDICT_STATES and state != "COMMENT":
|
||||
latest_reviewed_head = commit_id or latest_reviewed_head
|
||||
if reviewer:
|
||||
latest_by_reviewer[reviewer] = entry
|
||||
|
||||
blocking = [
|
||||
e for e in latest_by_reviewer.values()
|
||||
if e["verdict"] == "REQUEST_CHANGES" and not e["dismissed"]
|
||||
]
|
||||
approvals = [
|
||||
e for e in latest_by_reviewer.values()
|
||||
if e["verdict"] == "APPROVED" and not e["dismissed"]
|
||||
]
|
||||
return {
|
||||
"success": True,
|
||||
"pr_number": pr_number,
|
||||
"pr_state": pr.get("state"),
|
||||
"current_head_sha": current_head,
|
||||
"reviews": reviews,
|
||||
"latest_review_state_by_reviewer": {
|
||||
login: e["verdict"] for login, e in latest_by_reviewer.items()},
|
||||
"has_blocking_change_requests": bool(blocking),
|
||||
"approval_visible": bool(approvals),
|
||||
"latest_reviewed_head_sha": latest_reviewed_head,
|
||||
"review_feedback_stale": bool(
|
||||
latest_reviewed_head and current_head
|
||||
and latest_reviewed_head != current_head),
|
||||
"author_pushed_after_request_changes": any(
|
||||
e["reviewed_head_sha"] and current_head
|
||||
and e["reviewed_head_sha"] != current_head
|
||||
for e in blocking),
|
||||
}
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
@_audit_pr_result("submit_pr_review")
|
||||
def gitea_submit_pr_review(
|
||||
@@ -3079,6 +3212,67 @@ def gitea_mirror_refs(
|
||||
"return_code": result.returncode,
|
||||
}
|
||||
|
||||
_VALIDATION_STATUSES = ("passed", "failed", "skipped", "not-run")
|
||||
|
||||
|
||||
def build_validation_report(commands: list[dict]) -> dict:
|
||||
"""Build an explicit validation report for LLM final summaries (#167).
|
||||
|
||||
Each entry must carry 'command' and a 'status' from 'passed' /
|
||||
'failed' / 'skipped' / 'not-run'. A failed entry must carry the exact
|
||||
'output'; vague language ("shell-invocation quirks") is rejected as an
|
||||
unknown status and a missing failure output raises, so a report can
|
||||
never hide what actually happened. Optional keys: 'reason' (for
|
||||
skipped/not-run, e.g. what targeted check replaced it) and
|
||||
'full_suite': True to mark the full-test-suite command —
|
||||
'full_suite_passed' is True only if that entry passed, False if it
|
||||
failed/was skipped/not run, and None when no entry is marked, so
|
||||
full-suite success is never implied.
|
||||
"""
|
||||
entries = []
|
||||
full_suite_passed = None
|
||||
for raw in commands:
|
||||
command = (raw.get("command") or "").strip()
|
||||
if not command:
|
||||
raise ValueError("validation entry missing 'command' (fail closed)")
|
||||
status = raw.get("status")
|
||||
if status not in _VALIDATION_STATUSES:
|
||||
raise ValueError(
|
||||
f"unknown validation status {status!r} for '{command}'; "
|
||||
f"use one of {_VALIDATION_STATUSES} (fail closed)")
|
||||
output = raw.get("output")
|
||||
if status == "failed" and not (output or "").strip():
|
||||
raise ValueError(
|
||||
f"failed validation '{command}' must include the exact "
|
||||
"command output (fail closed)")
|
||||
entry = {"command": command, "status": status}
|
||||
if output:
|
||||
entry["output"] = output
|
||||
if raw.get("reason"):
|
||||
entry["reason"] = raw["reason"]
|
||||
if raw.get("full_suite"):
|
||||
entry["full_suite"] = True
|
||||
full_suite_passed = status == "passed"
|
||||
entries.append(entry)
|
||||
label = {"passed": "PASSED", "failed": "FAILED",
|
||||
"skipped": "SKIPPED", "not-run": "NOT-RUN"}
|
||||
lines = []
|
||||
for e in entries:
|
||||
line = f"{label[e['status']]}: {e['command']}"
|
||||
if e.get("output"):
|
||||
line += f" — {e['output']}"
|
||||
if e.get("reason"):
|
||||
line += f" ({e['reason']})"
|
||||
lines.append(line)
|
||||
return {
|
||||
"entries": entries,
|
||||
"all_passed": bool(entries) and all(
|
||||
e["status"] == "passed" for e in entries),
|
||||
"full_suite_passed": full_suite_passed,
|
||||
"summary": "\n".join(lines),
|
||||
}
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_resolve_task_capability(
|
||||
task: str,
|
||||
@@ -3128,6 +3322,10 @@ def gitea_resolve_task_capability(
|
||||
"permission": "gitea.pr.comment",
|
||||
"role": "author",
|
||||
},
|
||||
"address_pr_change_requests": {
|
||||
"permission": "gitea.branch.push",
|
||||
"role": "author",
|
||||
},
|
||||
"review_pr": {
|
||||
"permission": "gitea.pr.review",
|
||||
"role": "reviewer",
|
||||
@@ -3211,6 +3409,42 @@ def gitea_resolve_task_capability(
|
||||
"or use the corresponding MCP namespace."
|
||||
)
|
||||
|
||||
# Task/role alignment guards (#167): the requested task, not the
|
||||
# available credential, decides what the session may do. A review/merge
|
||||
# task under a non-reviewer profile must stop — not silently degrade
|
||||
# into author-side mutations.
|
||||
stop_required = not allowed_in_current_session
|
||||
task_role_guidance = []
|
||||
if required_role == "reviewer":
|
||||
if allowed_in_current_session:
|
||||
task_role_guidance.append(
|
||||
"Review/merge task: proceed only through the gated "
|
||||
"review/merge tools and their eligibility checks; "
|
||||
"self-review and self-merge remain blocked.")
|
||||
else:
|
||||
task_role_guidance.append(
|
||||
"STOP: the requested task is review/merge but the active "
|
||||
"profile cannot perform it. Do not fall back to author-side "
|
||||
"work — do not commit, push, edit files, or comment as "
|
||||
"author unless the operator explicitly changes the task to "
|
||||
"author-side remediation (task "
|
||||
"'address_pr_change_requests').")
|
||||
elif task == "address_pr_change_requests":
|
||||
if allowed_in_current_session:
|
||||
task_role_guidance.append(
|
||||
"Author-side remediation: you may commit and push fixes to "
|
||||
"the PR branch, but you must not submit review verdicts and "
|
||||
"must not merge.")
|
||||
else:
|
||||
task_role_guidance.append(
|
||||
"STOP: author-side remediation requires an author profile "
|
||||
"with branch push permission; the active profile cannot "
|
||||
"perform it.")
|
||||
elif not allowed_in_current_session:
|
||||
task_role_guidance.append(
|
||||
"STOP: the active profile cannot perform the requested task; "
|
||||
"follow exact_safe_next_action instead of improvising.")
|
||||
|
||||
return {
|
||||
"requested_task": task,
|
||||
"required_operation_permission": required_permission,
|
||||
@@ -3219,6 +3453,8 @@ def gitea_resolve_task_capability(
|
||||
"active_identity": username,
|
||||
"active_profile_allowed_operations": active_allowed,
|
||||
"allowed_in_current_session": allowed_in_current_session,
|
||||
"stop_required": stop_required,
|
||||
"task_role_guidance": task_role_guidance,
|
||||
"matching_configured_profile": matching_profiles,
|
||||
"runtime_switching_supported": switching,
|
||||
"different_mcp_namespace_required": different_namespace_required,
|
||||
|
||||
@@ -0,0 +1,458 @@
|
||||
"""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"])
|
||||
|
||||
@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()
|
||||
Reference in New Issue
Block a user