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

feat(review-workflow): enforce single terminal review decision per review run (#211)

Browse Source
This commit is contained in:
sysadmin
2026-07-05 17:40:55 -04:00
parent a1a7c5b30e
commit 6e9b95bb96
8 changed files with 219 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+1
View File
@@ -25,4 +25,5 @@ def _reset_mutation_authority(monkeypatch):
return
monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
yield
tests/test_mcp_server.py
+65 -1
View File
@@ -812,11 +812,16 @@ class TestReviewPR(unittest.TestCase):
{"login": "jcwalker3"}, # /api/v1/user (eligibility)
{"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}}, # /pulls/1
]
from mcp_server import init_review_decision_lock
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",
body="Self approve",
merge=False
merge=False,
remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(result["success"])
self.assertIn("Review submission failed eligibility gates", result["message"])
@@ -1764,6 +1769,65 @@ class TestSubmitPrReview(unittest.TestCase):
self.assertIn("[REDACTED]", blob)
self.assertNotIn("abc-secret-xyz", blob)
def test_authorize_review_correction_success(self):
from mcp_server import _load_review_decision_lock, _save_review_decision_lock
lock = _load_review_decision_lock() or {}
lock["live_mutations"] = [{"pr_number": 8, "action": "approve", "review_id": 42, "review_state": "approve"}]
_save_review_decision_lock(lock)
r = gitea_authorize_review_correction(prior_review_id=42, prior_review_state="approve", reason="typo")
self.assertTrue(r["authorized"])
lock = _load_review_decision_lock()
self.assertTrue(lock["correction_authorized"])
def test_authorize_review_correction_mismatch(self):
from mcp_server import _load_review_decision_lock, _save_review_decision_lock
lock = _load_review_decision_lock() or {}
lock["live_mutations"] = [{"pr_number": 8, "action": "approve", "review_id": 42, "review_state": "approve"}]
_save_review_decision_lock(lock)
# Mismatched ID
r = gitea_authorize_review_correction(prior_review_id=99, prior_review_state="approve", reason="typo")
self.assertFalse(r["authorized"])
self.assertTrue(any("prior review ID" in x for x in r["reasons"]))
# Mismatched State
r = gitea_authorize_review_correction(prior_review_id=42, prior_review_state="request_changes", reason="typo")
self.assertFalse(r["authorized"])
self.assertTrue(any("prior review state" in x for x in r["reasons"]))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_remote_org_repo_validation_mismatch(self, _auth, mock_api):
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
# Mark decision for specific remote, org, repo
gitea_mark_final_review_decision(8, "approve", remote="prgs", org="MyOrg", repo="MyRepo")
# Mismatched remote
r = gitea_submit_pr_review(
pr_number=8, action="approve", remote="dadeschools", org="MyOrg", repo="MyRepo",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertTrue(any("ready remote" in x for x in r["reasons"]))
# Mismatched org
r = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs", org="OtherOrg", repo="MyRepo",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertTrue(any("ready org" in x for x in r["reasons"]))
# Mismatched repo
r = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs", org="MyOrg", repo="OtherRepo",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertTrue(any("ready repo" in x for x in r["reasons"]))
if __name__ == "__main__":
unittest.main()
tests/test_pr_queue_inventory.py
+4 -1
View File
@@ -127,7 +127,10 @@ class TestPRQueueInventory(unittest.TestCase):
{"id": 100} # POST review
]
result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs")
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)
self.assertTrue(result["success"])
self.assertIn("=== PR Queue Inventory ===", result["message"])
self.assertIn("Repository:", result["message"])
tests/test_review_pr.py
+5 -23
View File
@@ -112,15 +112,10 @@ class TestAPIPayload(unittest.TestCase):
class TestMutationAuthorityLock(unittest.TestCase):
"""#199 (refs #194): the CLI refuses to run under a profile that differs
from the session profile lock exported by the launching MCP session."""
"""#199 (refs #194): the CLI refuses to run under active MCP sessions."""
@patch("review_pr.get_profile")
def test_cli_blocked_on_session_lock_mismatch(self, mock_get_profile):
# An author-bound session exported the lock; the CLI resolves a
# reviewer profile (GITEA_MCP_PROFILE side-channel override) — reject
# before any API call.
mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
def test_cli_disabled_on_session_lock(self):
# When running inside an MCP session, direct review submission via CLI is disabled entirely.
import io
buf = io.StringIO()
with patch.dict(os.environ,
@@ -129,22 +124,9 @@ class TestMutationAuthorityLock(unittest.TestCase):
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE",
])
self.assertEqual(rc, 3)
self.assertEqual(rc, 2)
msg = buf.getvalue().lower()
self.assertIn("cli override rejected", msg)
@patch("review_pr.get_profile")
@patch("review_pr.api_request")
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
def test_cli_allowed_on_profile_match(self, _auth, mock_api, mock_get_profile):
mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
mock_api.side_effect = [FAKE_PR_DATA, {}]
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE",
])
self.assertEqual(rc, 0)
self.assertIn("disabled within mcp sessions", msg)
@patch("review_pr.api_request")
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
tests/test_review_proofs.py
+22
View File
@@ -167,6 +167,15 @@ def _good_live_state(**overrides):
return assess_live_state_recheck(recheck)
def _good_review_mutation():
return {
"complete": True,
"downgraded": False,
"missing_fields": [],
"reasons": [],
}
def _good_role_boundary_179(**overrides):
kwargs = {
"task_role": "reviewer",
@@ -531,6 +540,7 @@ class TestFinalReport(unittest.TestCase):
"sweep": _good_sweep(),
"live_state": _good_live_state(),
"role_boundary": _good_role_boundary(),
"review_mutation": _good_review_mutation(),
}
kwargs.update(overrides)
return build_final_report(**kwargs)
@@ -625,6 +635,7 @@ class TestFinalReport(unittest.TestCase):
"identity_eligible": True,
"merge_performed": False,
"issue_status_verified": True,
"review_mutation": _good_review_mutation(),
}
report = build_final_report(**kwargs)
self.assertNotEqual(report["grade"], "A")
@@ -1131,6 +1142,7 @@ class TestFinalReport179Bar(unittest.TestCase):
"sweep": _good_sweep(),
"live_state": _good_live_state(),
"role_boundary": _good_role_boundary(),
"review_mutation": _good_review_mutation(),
}
kwargs.update(overrides)
return build_final_report(**kwargs)
@@ -1195,6 +1207,16 @@ class TestFinalReport179Bar(unittest.TestCase):
self.assertTrue(report["inventory_complete"])
self.assertTrue(report["validated_on_pinned_head"])
def test_missing_review_mutation_downgrades(self):
report = self._report(review_mutation=None)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["review_mutation_complete"])
def test_incomplete_review_mutation_downgrades(self):
report = self._report(review_mutation={"complete": False, "downgraded": True, "reasons": []})
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["review_mutation_complete"])
if __name__ == "__main__":
unittest.main()