diff --git a/mcp_server.py b/mcp_server.py index 7e61555..7fc5cc4 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -924,7 +924,8 @@ _REVIEW_ACTIONS = { _TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"}) -REVIEW_DECISION_FILE = "/tmp/gitea_review_decision.lock" +# In-process only (#211): never persist to /tmp — host-global files are +# spoofable by any local process and go stale across sessions. _REVIEW_DECISION_LOCK: dict | None = None @@ -938,13 +939,41 @@ def _save_review_decision_lock(data): _REVIEW_DECISION_LOCK = data +def _review_decision_session_reasons(lock: dict | None) -> list[str]: + """Reject locks that do not belong to this MCP process/session.""" + if lock is None: + return [] + reasons = [] + if lock.get("session_pid") != os.getpid(): + reasons.append( + "review decision lock was created in a different process " + "(fail closed)" + ) + env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip() + stored_lock = (lock.get("session_profile_lock") or "").strip() + if env_lock and stored_lock and env_lock != stored_lock: + reasons.append( + "review decision lock session profile lock mismatch (fail closed)" + ) + return reasons + + def init_review_decision_lock(remote: str | None, task: str | None): """Seed read-only-until-ready state for reviewer PR review tasks.""" if task != "review_pr": return + profile = get_profile() + profile_name = (profile.get("profile_name") or "").strip() + session_lock = ( + (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip() + or profile_name + ) _save_review_decision_lock({ "task": task, "remote": remote, + "session_pid": os.getpid(), + "session_profile": profile_name, + "session_profile_lock": session_lock, "final_review_decision_ready": False, "ready_pr_number": None, "ready_action": None, @@ -977,6 +1006,13 @@ def check_review_decision_gate( ) return reasons + reasons.extend(_review_decision_session_reasons(lock)) + if reasons: + return reasons + + if remote in REMOTES: + _, org, repo = _resolve(remote, None, org, repo) + if not final_review_decision_ready: reasons.append( "final_review_decision_ready must be true; validation-phase live " @@ -1367,12 +1403,46 @@ def gitea_mark_final_review_decision( "review decision lock missing; resolve review_pr capability first" ], } - if lock.get("live_mutations"): + session_reasons = _review_decision_session_reasons(lock) + if session_reasons: + return {"marked_ready": False, "reasons": session_reasons} + if remote != lock.get("remote"): + return { + "marked_ready": False, + "reasons": [ + f"requested remote '{remote}' does not match locked remote " + f"'{lock.get('remote')}' (fail closed)" + ], + } + try: + _, resolved_org, resolved_repo = _resolve(remote, None, org, repo) + except ValueError as exc: + return {"marked_ready": False, "reasons": [str(exc)]} + if org is not None and org != resolved_org: + return { + "marked_ready": False, + "reasons": [ + f"requested org '{org}' does not match resolved org " + f"'{resolved_org}' (fail closed)" + ], + } + if repo is not None and repo != resolved_repo: + return { + "marked_ready": False, + "reasons": [ + f"requested repo '{repo}' does not match resolved repo " + f"'{resolved_repo}' (fail closed)" + ], + } + org = resolved_org + repo = resolved_repo + if lock.get("live_mutations") and not lock.get("correction_authorized"): return { "marked_ready": False, "reasons": [ "cannot mark final decision after a live review mutation was " - "already recorded in this run" + "already recorded in this run unless an operator-approved " + "correction was authorized" ], } if action not in _REVIEW_ACTIONS: @@ -1409,6 +1479,7 @@ def gitea_authorize_review_correction( prior_review_id: int, prior_review_state: str, reason: str, + operator_authorized: bool = False, ) -> dict: """Authorize one operator-approved correction after a mistaken live review.""" reason = (reason or "").strip() @@ -1416,6 +1487,17 @@ def gitea_authorize_review_correction( lock = _load_review_decision_lock() if lock is None: return {"authorized": False, "reasons": ["review decision lock missing"]} + session_reasons = _review_decision_session_reasons(lock) + if session_reasons: + return {"authorized": False, "reasons": session_reasons} + if not operator_authorized: + return { + "authorized": False, + "reasons": [ + "operator authorization is required for review correction " + "(fail closed)" + ], + } prior = list(lock.get("live_mutations") or []) if not prior: return { diff --git a/review_pr.py b/review_pr.py index a72d7e8..48b6168 100755 --- a/review_pr.py +++ b/review_pr.py @@ -7,16 +7,16 @@ making any API call. Merge is handled solely by the gated `gitea_merge_pr` MCP workflow (#16), which enforces identity/profile/eligibility, explicit confirmation, expected head SHA checking, and self-merge protection. -Usage (review only): +Live review submission is also disabled (#211): use the gated +``gitea_submit_pr_review`` MCP workflow, which enforces validation-phase +dry-run, final decision marking, and single-terminal review mutation rules. + +Usage (review only — disabled): review_pr.py --pr-number 12 --event APPROVE --body "Approved and signed off" """ import os import sys -import json -import base64 import argparse -import urllib.request -import urllib.error # Auto-execute using the project's local virtual environment Python PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) @@ -24,18 +24,10 @@ venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") if os.path.exists(venv_python) and sys.executable != venv_python: os.execv(venv_python, [venv_python] + sys.argv) -from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, get_profile +from gitea_auth import add_remote_args def main(argv=None): - if os.environ.get("GITEA_SESSION_PROFILE_LOCK"): - print( - "Direct CLI review submission is disabled within MCP sessions. " - "Use MCP tools instead.", - file=sys.stderr, - ) - return 2 - parser = argparse.ArgumentParser(description="Review and sign-off on a Gitea pull request.") add_remote_args(parser) parser.add_argument("--pr-number", type=int, required=True, help="PR number/index to review.") @@ -50,87 +42,26 @@ def main(argv=None): help="Ignored — CLI merge is disabled (see --merge).") args = parser.parse_args(argv) - # Fail closed: direct CLI merge is disabled (#16). LLM automations were - # using this flag as an ungated merge bypass. Merge is only available via - # the gated `gitea_merge_pr` MCP workflow, which enforces - # identity/profile/eligibility, explicit confirmation, expected head SHA, - # and self-merge protection. No API call is made here. if args.merge: print( "Direct CLI merge is disabled. Merge is only available through the " "gated #16 workflow (MCP tool 'gitea_merge_pr'), which enforces " "identity/profile/eligibility, explicit confirmation, expected head " "SHA checking, and self-merge protection. Re-run without --merge to " - "submit a review only.", + "see the review-submission guard message.", file=sys.stderr, ) return 2 - host, org, repo = resolve_remote(args) - - # ── Reviewer mutation side-channel wall (#199, refs #194) ── - # The launching MCP session exports GITEA_SESSION_PROFILE_LOCK with the - # profile it was started with; child processes inherit it. If this CLI - # resolves a different profile — e.g. an ad-hoc GITEA_MCP_PROFILE - # override escalating an author-bound session to reviewer — refuse - # before any API call. No lock in the environment means no session - # context (direct operator CLI use), which stays allowed. Unlike a /tmp - # lock file, the environment is per-process-tree: other sessions cannot - # spoof it and it cannot go stale across sessions. - session_lock = (os.environ.get("GITEA_SESSION_PROFILE_LOCK") or "").strip() - if session_lock: - print( - "Direct CLI review submission is disabled within MCP sessions. " - "Use the gated 'gitea_submit_pr_review' MCP tool instead.", - file=sys.stderr, - ) - return 2 - - body = args.body - if args.body_file: - if args.body_file == "-": - body = sys.stdin.read() - else: - with open(args.body_file, "r", encoding="utf-8") as fh: - body = fh.read() - - auth = get_auth_header(host) - if not auth: - print(f"Could not get credentials or token for {host}.", file=sys.stderr) - return 1 - - # 1. Fetch PR to get the latest head commit SHA (required for review validation) - pr_url = f"{repo_api_url(host, org, repo)}/pulls/{args.pr_number}" - try: - pr_data = api_request("GET", pr_url, auth) - except Exception as e: - print(f"Error fetching PR #{args.pr_number}: {e}", file=sys.stderr) - return 1 - - commit_sha = pr_data.get("head", {}).get("sha") - if not commit_sha: - print(f"Could not find head commit SHA for PR #{args.pr_number}.", file=sys.stderr) - return 1 - - # 2. Submit the PR review - review_url = f"{repo_api_url(host, org, repo)}/pulls/{args.pr_number}/reviews" - payload = { - "body": body, - "event": args.event, - "commit_id": commit_sha - } - - try: - api_request("POST", review_url, auth, payload) - print(f"Successfully submitted review for PR #{args.pr_number}: event={args.event}") - except Exception as e: - print(f"Error submitting review: {e}", file=sys.stderr) - return 1 - - # Merge is intentionally not performed here — see the fail-closed guard - # above. Use the gated `gitea_merge_pr` MCP workflow (#16) to merge. - return 0 + print( + "Direct CLI review submission is disabled (#211). Use the gated " + "'gitea_submit_pr_review' MCP workflow, which enforces validation-phase " + "dry-run, gitea_mark_final_review_decision, and single-terminal review " + "mutation rules.", + file=sys.stderr, + ) + return 2 if __name__ == "__main__": - sys.exit(main()) + sys.exit(main()) \ No newline at end of file diff --git a/review_proofs.py b/review_proofs.py index 4ed6fc4..cdc31c4 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -634,7 +634,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination, identity_eligible, merge_performed, issue_status_verified, capability_evidence=None, sweep=None, live_state=None, - role_boundary=None, review_mutation=None): + role_boundary=None, review_mutation=None, + report_text=None, review_decision_lock=None): """Required behavior 6 + acceptance criteria: one report, distinct proofs. Combines the individual proof verdicts into the final-report fields the @@ -655,6 +656,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, #211: the report must also carry the review mutation proof (``assess_review_mutation_final_report``). """ + if review_mutation is None and report_text is not None: + review_mutation = assess_review_mutation_final_report( + report_text, review_decision_lock + ) + contamination_status = contamination.get("status", "unknown") checkout_proven = bool(checkout_proof.get("proven")) validation_claimable = bool(validation.get("claimable")) diff --git a/tests/test_audit.py b/tests/test_audit.py index 70cae76..6594739 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -330,12 +330,12 @@ class TestGatedToolAudit(_AuditWiringBase): mock_api.side_effect = [ {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 7}, ] - 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(8, "approve", remote="prgs") env = self._env(GITEA_PROFILE_NAME="gitea-reviewer", GITEA_ALLOWED_OPERATIONS="read,review,approve") with patch.dict(os.environ, env, clear=True): + 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(8, "approve", remote="prgs") r = gitea_submit_pr_review(pr_number=8, action="approve", body="LGTM", remote="prgs", final_review_decision_ready=True) diff --git a/tests/test_llm_agent_sha.py b/tests/test_llm_agent_sha.py index c021374..1a6ecff 100644 --- a/tests/test_llm_agent_sha.py +++ b/tests/test_llm_agent_sha.py @@ -125,8 +125,8 @@ class TestShaCannotBypassSelfReview(unittest.TestCase): mock_get_all.return_value = [{"number": 9, "title": "PR 9", "state": "open", "head": {"ref": "branch9", "sha": "abc1234"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}] mock_api.side_effect = [ {"login": "jcwalker3"}, # /user (inventory) - {"login": "jcwalker3"}, # /user (eligibility) - {"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}} # /pulls/9 (eligibility) + {"login": "jcwalker3"}, # /user (submit eligibility) + {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/9 ] from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision init_review_decision_lock("prgs", "review_pr") diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index eacf845..263f248 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -35,7 +35,7 @@ from mcp_server import ( # noqa: E402 gitea_mark_final_review_decision, gitea_authorize_review_correction, init_review_decision_lock, - REVIEW_DECISION_FILE, + gitea_list_issue_comments, gitea_create_issue_comment, ) @@ -818,8 +818,8 @@ class TestReviewPR(unittest.TestCase): # mock_api responses: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility) mock_api.side_effect = [ {"login": "jcwalker3"}, # /api/v1/user (inventory) - {"login": "jcwalker3"}, # /api/v1/user (eligibility) - {"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}}, # /pulls/1 + {"login": "jcwalker3"}, # /api/v1/user (submit eligibility) + {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/1 ] from mcp_server import init_review_decision_lock init_review_decision_lock("prgs", "review_pr") @@ -1422,10 +1422,6 @@ class TestReviewDecisionValidationGate(unittest.TestCase): def setUp(self): init_review_decision_lock("prgs", "review_pr") - def tearDown(self): - if os.path.exists(REVIEW_DECISION_FILE): - os.remove(REVIEW_DECISION_FILE) - def _env(self): return patch.dict(os.environ, { "GITEA_PROFILE_NAME": "gitea-reviewer", @@ -1520,10 +1516,6 @@ class TestSubmitPrReview(unittest.TestCase): init_review_decision_lock("prgs", "review_pr") gitea_mark_final_review_decision(8, "approve", remote="prgs") - def tearDown(self): - if os.path.exists(REVIEW_DECISION_FILE): - os.remove(REVIEW_DECISION_FILE) - def _pr(self, author, state="open", sha="abc123", mergeable=True): return { "user": {"login": author}, @@ -1783,7 +1775,12 @@ class TestSubmitPrReview(unittest.TestCase): 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") + r = gitea_authorize_review_correction( + prior_review_id=42, + prior_review_state="approve", + reason="typo", + operator_authorized=True, + ) self.assertTrue(r["authorized"]) lock = _load_review_decision_lock() self.assertTrue(lock["correction_authorized"]) @@ -1795,15 +1792,98 @@ class TestSubmitPrReview(unittest.TestCase): _save_review_decision_lock(lock) # Mismatched ID - r = gitea_authorize_review_correction(prior_review_id=99, prior_review_state="approve", reason="typo") + r = gitea_authorize_review_correction( + prior_review_id=99, + prior_review_state="approve", + reason="typo", + operator_authorized=True, + ) 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") + r = gitea_authorize_review_correction( + prior_review_id=42, + prior_review_state="request_changes", + reason="typo", + operator_authorized=True, + ) self.assertFalse(r["authorized"]) self.assertTrue(any("prior review state" in x for x in r["reasons"])) + def test_authorize_review_correction_requires_operator_auth(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", + operator_authorized=False, + ) + self.assertFalse(r["authorized"]) + self.assertTrue(any("operator authorization" in x for x in r["reasons"])) + + def test_spoofed_tmp_lock_file_does_not_bypass_gate(self): + import json + import mcp_server + mcp_server._REVIEW_DECISION_LOCK = None + spoof_path = "/tmp/gitea_review_decision.lock" + with open(spoof_path, "w", encoding="utf-8") as fh: + json.dump({"final_review_decision_ready": True}, fh) + try: + 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", + final_review_decision_ready=True, + ) + self.assertFalse(r["performed"]) + self.assertTrue(any("review decision lock missing" in x for x in r["reasons"])) + finally: + if os.path.exists(spoof_path): + os.remove(spoof_path) + + def test_mark_final_decision_rejects_remote_mismatch(self): + init_review_decision_lock("prgs", "review_pr") + r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools") + self.assertFalse(r["marked_ready"]) + self.assertTrue(any("does not match locked remote" in x for x in r["reasons"])) + + @patch("mcp_server.api_request") + @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}, + ] + env = {"GITEA_PROFILE_NAME": "gitea-reviewer", + "GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes"} + gitea_mark_final_review_decision(8, "approve", remote="prgs") + with patch.dict(os.environ, env, clear=True): + first = gitea_submit_pr_review( + pr_number=8, action="approve", remote="prgs", + final_review_decision_ready=True, + ) + auth = gitea_authorize_review_correction( + prior_review_id=42, + prior_review_state="approve", + reason="operator approved correcting mistaken approve", + operator_authorized=True, + ) + gitea_mark_final_review_decision(8, "request_changes", remote="prgs") + second = gitea_submit_pr_review( + pr_number=8, action="request_changes", remote="prgs", + final_review_decision_ready=True, + ) + self.assertTrue(first["performed"]) + self.assertTrue(auth["authorized"]) + self.assertTrue(second["performed"]) + @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): diff --git a/tests/test_pr_queue_inventory.py b/tests/test_pr_queue_inventory.py index a5d948d..4866f99 100644 --- a/tests/test_pr_queue_inventory.py +++ b/tests/test_pr_queue_inventory.py @@ -122,9 +122,9 @@ class TestPRQueueInventory(unittest.TestCase): # mock_api: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility), 4) /pulls/1/reviews (POST review) mock_api.side_effect = [ {"login": "reviewer1"}, # inventory whoami - {"login": "reviewer1"}, # eligibility whoami - {"state": "open", "head": {"sha": "abc1"}, "mergeable": True, "user": {"login": "other_user"}}, # eligibility PR details - {"id": 100} # POST review + {"login": "reviewer1"}, # submit eligibility whoami + {"user": {"login": "other_user"}, "state": "open", "head": {"sha": "abc1"}, "mergeable": True}, # submit eligibility PR + {"id": 100}, # POST review ] from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision diff --git a/tests/test_review_pr.py b/tests/test_review_pr.py index 9124bb1..9db4a6f 100644 --- a/tests/test_review_pr.py +++ b/tests/test_review_pr.py @@ -34,8 +34,7 @@ class TestArgParsing(unittest.TestCase): self.exists_patcher.start() self.addCleanup(self.exists_patcher.stop) - @patch("review_pr.get_auth_header", return_value=FAKE_CREDS) - def test_missing_pr_number_exits(self, _auth): + def test_missing_pr_number_exits(self): with self.assertRaises(SystemExit): review_pr.main([]) @@ -47,37 +46,21 @@ class TestAPIPayload(unittest.TestCase): self.exists_patcher.start() self.addCleanup(self.exists_patcher.stop) - @patch("review_pr.api_request") - @patch("review_pr.get_auth_header", return_value=FAKE_CREDS) - def test_payload_fields_and_workflow(self, _auth, mock_api): - # Setup mock api_request to return PR details, then review response - mock_api.side_effect = [FAKE_PR_DATA, {}] + def test_review_submission_fails_closed_without_api_call(self): + # #211: live review submission must use gated MCP tools, not CLI POST. + import io + buf = io.StringIO() + with patch.object(sys, "stderr", buf): + rc = review_pr.main([ + "--pr-number", "81", + "--event", "APPROVE", + "--body", "Approved and ready to merge", + ]) + self.assertEqual(rc, 2) + self.assertIn("disabled", buf.getvalue().lower()) + self.assertIn("gitea_submit_pr_review", buf.getvalue()) - rc = review_pr.main([ - "--pr-number", "81", - "--event", "APPROVE", - "--body", "Approved and ready to merge", - ]) - self.assertEqual(rc, 0) - self.assertEqual(mock_api.call_count, 2) - - # Verify first call: GET PR - first_call_args = mock_api.call_args_list[0] - self.assertEqual(first_call_args[0][0], "GET") - self.assertEqual(first_call_args[0][1], "https://gitea.dadeschools.net/api/v1/repos/Contractor/Timesheet/pulls/81") - - # Verify second call: POST review - second_call_args = mock_api.call_args_list[1] - self.assertEqual(second_call_args[0][0], "POST") - self.assertEqual(second_call_args[0][1], "https://gitea.dadeschools.net/api/v1/repos/Contractor/Timesheet/pulls/81/reviews") - payload = second_call_args[0][3] - self.assertEqual(payload["event"], "APPROVE") - self.assertEqual(payload["body"], "Approved and ready to merge") - self.assertEqual(payload["commit_id"], "abcdef1234567890") - - @patch("review_pr.api_request") - @patch("review_pr.get_auth_header", return_value=FAKE_CREDS) - def test_merge_flag_fails_closed_without_api_call(self, _auth, mock_api): + def test_merge_flag_fails_closed_without_api_call(self): # --merge is an ungated bypass and is disabled (#16). It must fail # closed BEFORE any API call — no review, no merge. rc = review_pr.main([ @@ -88,24 +71,15 @@ class TestAPIPayload(unittest.TestCase): "--merge-method", "squash", ]) self.assertEqual(rc, 2) - self.assertEqual(mock_api.call_count, 0) def test_merge_flag_message_points_to_gated_workflow(self): - from _pytest.monkeypatch import MonkeyPatch import io - with patch("review_pr.get_auth_header", return_value=FAKE_CREDS), \ - patch("review_pr.api_request") as mock_api: - buf = io.StringIO() - monkeypatch = MonkeyPatch() - monkeypatch.setattr(sys, "stderr", buf) - try: - rc = review_pr.main([ - "--pr-number", "81", "--event", "APPROVE", "--merge", - ]) - finally: - monkeypatch.undo() + buf = io.StringIO() + with patch.object(sys, "stderr", buf): + rc = review_pr.main([ + "--pr-number", "81", "--event", "APPROVE", "--merge", + ]) self.assertEqual(rc, 2) - self.assertEqual(mock_api.call_count, 0) msg = buf.getvalue().lower() self.assertIn("disabled", msg) self.assertIn("gitea_merge_pr", msg) @@ -126,21 +100,18 @@ class TestMutationAuthorityLock(unittest.TestCase): ]) self.assertEqual(rc, 2) msg = buf.getvalue().lower() - self.assertIn("disabled within mcp sessions", msg) + self.assertIn("disabled", msg) + self.assertIn("gitea_submit_pr_review", msg) - @patch("review_pr.api_request") - @patch("review_pr.get_auth_header", return_value=FAKE_CREDS) - def test_cli_allowed_without_session_lock(self, _auth, mock_api): - # No lock in the environment = direct operator CLI use; the wall - # does not apply and the normal flow proceeds. - mock_api.side_effect = [FAKE_PR_DATA, {}] + def test_cli_disabled_even_without_session_lock(self): + # #211: CLI review submission is fail-closed regardless of session lock. env = {k: v for k, v in os.environ.items() if k != "GITEA_SESSION_PROFILE_LOCK"} with patch.dict(os.environ, env, clear=True): rc = review_pr.main([ "--pr-number", "81", "--event", "APPROVE", ]) - self.assertEqual(rc, 0) + self.assertEqual(rc, 2) if __name__ == "__main__": diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 4118f20..847de9d 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -168,12 +168,16 @@ def _good_live_state(**overrides): def _good_review_mutation(): - return { - "complete": True, - "downgraded": False, - "missing_fields": [], - "reasons": [], + report = ( + "Review complete on PR #203.\n" + "Review mutations: one request_changes on #203.\n" + "Review decision: request_changes." + ) + lock = { + "live_mutations": [{"pr_number": 203, "action": "request_changes"}], + "correction_authorized": False, } + return assess_review_mutation_final_report(report, lock) def _good_role_boundary_179(**overrides): @@ -920,6 +924,32 @@ class TestReviewMutationFinalReport(unittest.TestCase): ) self.assertTrue(complete["complete"]) + def test_build_final_report_wires_review_mutation_from_report_text(self): + report_text = ( + "Review complete on PR #203.\n" + "Review mutations: one request_changes on #203." + ) + lock = { + "live_mutations": [{"pr_number": 203, "action": "request_changes"}], + "correction_authorized": False, + } + final = build_final_report( + checkout_proof=_good_checkout(), + inventory=_good_inventory(), + validation=_good_validation(), + contamination=_good_contamination(), + identity_eligible=True, + merge_performed=False, + issue_status_verified=True, + capability_evidence=_good_capability_evidence(), + sweep=_good_sweep(), + live_state=_good_live_state(), + role_boundary=_good_role_boundary(), + report_text=report_text, + review_decision_lock=lock, + ) + self.assertTrue(final["review_mutation_complete"]) + class TestPRInventoryTrustGate(unittest.TestCase): """Issue #194: unit tests for the PR inventory trust gate."""