From 10d2644790370a5624dbea232418178011e269ec Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Sun, 5 Jul 2026 16:12:44 -0400 Subject: [PATCH 1/2] feat(reviewer-workflow): add hard wall against reviewer mutations through alternate profile or CLI side-channel --- mcp_server.py | 92 ++++++++++++++++++++++++++++++++++++++++ review_pr.py | 25 ++++++++++- tests/test_mcp_server.py | 92 ++++++++++++++++++++++++++++++++++++++++ tests/test_review_pr.py | 77 +++++++++++++++++++++++++++++++++ 4 files changed, 285 insertions(+), 1 deletion(-) diff --git a/mcp_server.py b/mcp_server.py index 4a52b97..c21e306 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -16,10 +16,69 @@ Configuration (mcp_config.json): import os import re import sys +import json import functools import contextlib import subprocess + +LOCK_FILE = "/tmp/gitea_mutation_authority.lock" + +def record_mutation_authority(profile_name: str | None, identity: str | None, remote: str | None, task: str | None): + """Record the resolved capability context to fail-closed lock file.""" + data = { + "initial_profile": profile_name, + "initial_identity": identity, + "current_profile": profile_name, + "current_identity": identity, + "remote": remote, + "task": task, + "role_pivot_authorized": False, + "role_pivot_record": None, + } + try: + with open(LOCK_FILE, "w", encoding="utf-8") as f: + json.dump(data, f) + except Exception: + pass + +def verify_mutation_authority(remote: str | None, host: str | None = None, required_role: str = "reviewer"): + """Verify that the current mutation matches the locked capability context.""" + if not os.path.exists(LOCK_FILE): + raise RuntimeError("Mutation authority lock is missing (fail closed)") + + try: + with open(LOCK_FILE, "r", encoding="utf-8") as f: + data = json.load(f) + except Exception as e: + raise RuntimeError(f"Could not read mutation authority lock: {e} (fail closed)") + + if data.get("remote") != remote: + raise RuntimeError( + f"Mutation remote '{remote}' does not match locked remote '{data.get('remote')}' (fail closed)" + ) + + profile = get_profile() + active_profile = profile.get("profile_name") + h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None) + active_identity = _authenticated_username(h) if h else None + + locked_profile = data.get("current_profile") + locked_identity = data.get("current_identity") + + if active_profile != locked_profile or active_identity != locked_identity: + raise RuntimeError( + f"Mutation profile '{active_profile}' or identity '{active_identity}' " + f"does not match locked authority (profile: '{locked_profile}', identity: '{locked_identity}') (fail closed)" + ) + + # Check reviewer/author role pivot boundaries + if required_role == "reviewer" and "author" in str(data.get("initial_profile")).lower() and "reviewer" in str(active_profile).lower(): + if not data.get("role_pivot_authorized"): + raise RuntimeError( + "Attempted reviewer mutation from author session without authorized role pivot (fail closed)" + ) + # Resolve the project root. MCP clients must launch this script directly with # the venv interpreter (venv/bin/python3) — see the config example above. We do # NOT os.execv() to re-point the interpreter: replacing the process after the @@ -991,6 +1050,12 @@ def gitea_submit_pr_review( } reasons = result["reasons"] + try: + verify_mutation_authority(remote, host, required_role="reviewer") + except RuntimeError as e: + reasons.append(str(e)) + return result + # Gate 1 — valid review action (no mutation on unknown action). if action not in _REVIEW_ACTIONS: reasons.append( @@ -1311,6 +1376,12 @@ def gitea_merge_pr( } reasons = result["reasons"] + try: + verify_mutation_authority(remote, host, required_role="reviewer") + except RuntimeError as e: + reasons.append(str(e)) + return result + # Gate 1 — valid merge method (no API call on a bad method). if do not in _MERGE_METHODS: reasons.append( @@ -2987,6 +3058,25 @@ def gitea_activate_profile( after_profile = get_profile()["profile_name"] after_identity = _authenticated_username(h) if h else None + # 4.5 Record pivot in mutation authority lock + if os.path.exists(LOCK_FILE): + try: + with open(LOCK_FILE, "r", encoding="utf-8") as f: + lock_data = json.load(f) + lock_data["current_profile"] = after_profile + lock_data["current_identity"] = after_identity + lock_data["role_pivot_authorized"] = True + lock_data["role_pivot_record"] = { + "from_profile": before_profile, + "to_profile": after_profile, + "from_identity": before_identity, + "to_identity": after_identity + } + with open(LOCK_FILE, "w", encoding="utf-8") as f: + json.dump(lock_data, f) + except Exception: + pass + # 5. Audit the switch if auditing is on _audit( "activate_profile", @@ -3476,6 +3566,8 @@ def gitea_resolve_task_capability( "STOP: the active profile cannot perform the requested task; " "follow exact_safe_next_action instead of improvising.") + record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task) + return { "requested_task": task, "required_operation_permission": required_permission, diff --git a/review_pr.py b/review_pr.py index 464d348..1614720 100755 --- a/review_pr.py +++ b/review_pr.py @@ -24,7 +24,7 @@ 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 +from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, get_profile def main(argv=None): @@ -60,6 +60,29 @@ def main(argv=None): host, org, repo = resolve_remote(args) + # ── Mutation Authority context wall check (Issue #194) ── + LOCK_FILE = "/tmp/gitea_mutation_authority.lock" + + if os.path.exists(LOCK_FILE): + try: + with open(LOCK_FILE, "r", encoding="utf-8") as f: + lock_data = json.load(f) + + # Resolve current CLI profile + cli_profile = get_profile().get("profile_name") + locked_profile = lock_data.get("current_profile") + + if cli_profile != locked_profile: + print( + f"Mismatched active profile vs mutation profile (CLI override rejected): " + f"CLI profile '{cli_profile}' does not match locked active profile '{locked_profile}' (fail closed)", + file=sys.stderr + ) + return 3 + except Exception as e: + print(f"Mutation authority check failed: {e}", file=sys.stderr) + return 3 + body = args.body if args.body_file: if args.body_file == "-": diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index d39ff87..f1c1cce 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -37,6 +37,11 @@ from mcp_server import ( # noqa: E402 from gitea_auth import get_profile # noqa: E402 import gitea_config # noqa: E402 +import mcp_server +# Globally disable verification check for existing isolated unit tests +_real_verify = mcp_server.verify_mutation_authority +mcp_server.verify_mutation_authority = lambda *args, **kwargs: None + FAKE_AUTH = "Basic dGVzdDp0ZXN0" @@ -2296,3 +2301,90 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase): "gitea.issue.comment", reviewer["allowed_operations"], reviewer.get("forbidden_operations", [])) self.assertTrue(ok) + + +class TestVerifyMutationAuthority(unittest.TestCase): + """Test verification lock logic under various configurations.""" + + def setUp(self): + self.patch_profile = patch("mcp_server.get_profile") + self.mock_profile = self.patch_profile.start() + self.patch_username = patch("mcp_server._authenticated_username") + self.mock_username = self.patch_username.start() + + # Restore real function for these tests + self._old_verify = mcp_server.verify_mutation_authority + mcp_server.verify_mutation_authority = _real_verify + + def tearDown(self): + self.patch_profile.stop() + self.patch_username.stop() + mcp_server.verify_mutation_authority = self._old_verify + # Clean up lock file + if os.path.exists("/tmp/gitea_mutation_authority.lock"): + os.remove("/tmp/gitea_mutation_authority.lock") + + def test_missing_lock_fails_closed(self): + if os.path.exists("/tmp/gitea_mutation_authority.lock"): + os.remove("/tmp/gitea_mutation_authority.lock") + with self.assertRaises(RuntimeError) as ctx: + _real_verify("prgs") + self.assertIn("lock is missing", str(ctx.exception)) + + def test_mismatched_remote_fails(self): + with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f: + json.dump({ + "remote": "dadeschools", + "current_profile": "prgs-reviewer", + "current_identity": "sysadmin" + }, f) + with self.assertRaises(RuntimeError) as ctx: + _real_verify("prgs") + self.assertIn("does not match locked remote", str(ctx.exception)) + + def test_mismatched_profile_fails(self): + with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f: + json.dump({ + "remote": "prgs", + "current_profile": "prgs-author", + "current_identity": "jcwalker3" + }, f) + self.mock_profile.return_value = {"profile_name": "prgs-reviewer"} + self.mock_username.return_value = "sysadmin" + + with self.assertRaises(RuntimeError) as ctx: + _real_verify("prgs") + self.assertIn("does not match locked authority", str(ctx.exception)) + + def test_author_to_reviewer_pivot_blocked_without_authorization(self): + with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f: + json.dump({ + "remote": "prgs", + "initial_profile": "prgs-author", + "initial_identity": "jcwalker3", + "current_profile": "prgs-reviewer", + "current_identity": "sysadmin", + "role_pivot_authorized": False + }, f) + self.mock_profile.return_value = {"profile_name": "prgs-reviewer"} + self.mock_username.return_value = "sysadmin" + + with self.assertRaises(RuntimeError) as ctx: + _real_verify("prgs", required_role="reviewer") + self.assertIn("without authorized role pivot", str(ctx.exception)) + + def test_allowed_when_match(self): + with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f: + json.dump({ + "remote": "prgs", + "initial_profile": "prgs-reviewer", + "initial_identity": "sysadmin", + "current_profile": "prgs-reviewer", + "current_identity": "sysadmin", + "role_pivot_authorized": False + }, f) + self.mock_profile.return_value = {"profile_name": "prgs-reviewer"} + self.mock_username.return_value = "sysadmin" + + # Should pass without exception + _real_verify("prgs") diff --git a/tests/test_review_pr.py b/tests/test_review_pr.py index d28ff01..2575b8f 100644 --- a/tests/test_review_pr.py +++ b/tests/test_review_pr.py @@ -2,6 +2,8 @@ Mocks api_request and credentials. """ +import io +import os import sys import unittest from unittest.mock import patch @@ -27,6 +29,11 @@ FAKE_PR_DATA = { class TestArgParsing(unittest.TestCase): + def setUp(self): + self.exists_patcher = patch("os.path.exists", return_value=False) + 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): with self.assertRaises(SystemExit): @@ -35,6 +42,11 @@ class TestArgParsing(unittest.TestCase): class TestAPIPayload(unittest.TestCase): + def setUp(self): + self.exists_patcher = patch("os.path.exists", return_value=False) + 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): @@ -99,5 +111,70 @@ class TestAPIPayload(unittest.TestCase): self.assertIn("gitea_merge_pr", msg) +class TestMutationAuthorityLock(unittest.TestCase): + """Issue #194: verify that the CLI tool rejects profile overrides when mismatched with lock.""" + + @patch("review_pr.get_profile") + def test_cli_blocked_on_profile_mismatch(self, mock_get_profile): + mock_get_profile.return_value = {"profile_name": "prgs-reviewer"} + + original_exists = os.path.exists + def conditional_exists(path): + if "gitea_mutation_authority.lock" in str(path): + return True + return original_exists(path) + + original_open = open + def conditional_open(file, *args, **kwargs): + if "gitea_mutation_authority.lock" in str(file): + return io.StringIO('{"current_profile": "prgs-author"}') + return original_open(file, *args, **kwargs) + + from _pytest.monkeypatch import MonkeyPatch + import io + buf = io.StringIO() + monkeypatch = MonkeyPatch() + monkeypatch.setattr(sys, "stderr", buf) + + with patch("os.path.exists", side_effect=conditional_exists), \ + patch("builtins.open", side_effect=conditional_open): + try: + rc = review_pr.main([ + "--pr-number", "81", "--event", "APPROVE", + ]) + finally: + monkeypatch.undo() + + self.assertEqual(rc, 3) + 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, {}] + + original_exists = os.path.exists + def conditional_exists(path): + if "gitea_mutation_authority.lock" in str(path): + return True + return original_exists(path) + + original_open = open + def conditional_open(file, *args, **kwargs): + if "gitea_mutation_authority.lock" in str(file): + return io.StringIO('{"current_profile": "prgs-reviewer"}') + return original_open(file, *args, **kwargs) + + with patch("os.path.exists", side_effect=conditional_exists), \ + patch("builtins.open", side_effect=conditional_open): + rc = review_pr.main([ + "--pr-number", "81", "--event", "APPROVE", + ]) + self.assertEqual(rc, 0) + + if __name__ == "__main__": unittest.main() -- 2.43.7 From 88deed4e69b441987a6036751bcaaab6e2392533 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Sun, 5 Jul 2026 16:23:26 -0400 Subject: [PATCH 2/2] feat(author-workflow): enforce exact issue lock before branch/commit/push/PR (Closes #204) --- mcp_server.py | 114 ++++++++++++++++++++++++++++++++++++ review_proofs.py | 29 +++++++++ scripts/worktree-start | 10 ++++ tests/test_mcp_server.py | 95 ++++++++++++++++++++++++++++-- tests/test_review_proofs.py | 34 +++++++++++ tests/test_worktrees.py | 38 ++++++++++-- 6 files changed, 311 insertions(+), 9 deletions(-) diff --git a/mcp_server.py b/mcp_server.py index c21e306..f74a1f2 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -23,6 +23,7 @@ import subprocess LOCK_FILE = "/tmp/gitea_mutation_authority.lock" +ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json" def record_mutation_authority(profile_name: str | None, identity: str | None, remote: str | None, task: str | None): """Record the resolved capability context to fail-closed lock file.""" @@ -399,6 +400,84 @@ def gitea_create_issue( return _with_optional_url({"number": data["number"]}, data.get("html_url")) +@mcp.tool() +def gitea_lock_issue( + issue_number: int, + branch_name: str, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Lock exactly one Gitea issue and its branch name to ensure durable tracking. + + Args: + issue_number: The tracking issue number. + branch_name: The branch name (must match (fix|feat|docs|chore)/issue--). + remote: Known instance — 'dadeschools' or 'prgs'. + host: Override Gitea host. + org: Override Org. + repo: Override Repo. + """ + # 1. Enforce branch name includes issue number + expected_pattern = f"issue-{issue_number}" + if expected_pattern not in branch_name: + raise ValueError( + f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)" + ) + + # 2. Check if the issue already has an open PR (reuse protection) + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + url = f"{repo_api_url(h, o, r)}/pulls?state=open" + + try: + prs = api_get_all(url, auth) + except Exception as e: + raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}") + + for pr in prs: + pr_head = pr.get("head", {}).get("ref", "") + pr_title = pr.get("title", "") + pr_body = pr.get("body", "") + + if expected_pattern in pr_head: + raise ValueError( + f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)" + ) + + patterns = [ + f"closes #{issue_number}", + f"fixes #{issue_number}", + ] + text_to_check = f"{pr_title} {pr_body}".lower() + if any(p in text_to_check for p in patterns): + raise ValueError( + f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)" + ) + + data = { + "issue_number": issue_number, + "branch_name": branch_name, + "remote": remote, + "org": o, + "repo": r, + } + + try: + with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: + json.dump(data, f) + except Exception as e: + raise RuntimeError(f"Could not write issue lock file: {e}") + + return { + "success": True, + "message": f"Successfully locked issue #{issue_number} to branch '{branch_name}' (fail-closed check complete).", + "issue_number": issue_number, + "branch_name": branch_name, + } + + @mcp.tool() def gitea_create_pr( title: str, @@ -426,6 +505,41 @@ def gitea_create_pr( dict with 'number' of the created PR ('url' only with the reveal opt-in). """ h, o, r = _resolve(remote, host, org, repo) + + # ── Issue Lock Validation (Issue #194 / #196) ── + if not os.path.exists(ISSUE_LOCK_FILE): + raise RuntimeError("Issue lock is missing (fail closed). Call gitea_lock_issue first.") + + try: + with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f: + lock_data = json.load(f) + except Exception as e: + raise RuntimeError(f"Could not read issue lock file: {e} (fail closed)") + + locked_issue = lock_data.get("issue_number") + locked_branch = lock_data.get("branch_name") + + if head != locked_branch: + raise ValueError( + f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)" + ) + + # Check for forbidden terms anywhere in title/body + forbidden_terms = ["equivalent", "related", "same as"] + text_to_check = f"{title} {body}".lower() + for term in forbidden_terms: + if term in text_to_check: + raise ValueError( + f"PR title or body contains forbidden term '{term}' (fail closed)" + ) + + # Ensure Closes # or Fixes # is present exactly + closes_pattern = re.compile(rf"\b(closes|fixes)\s+#{locked_issue}\b", re.IGNORECASE) + if not closes_pattern.search(text_to_check): + raise ValueError( + f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)" + ) + auth = _auth(h) url = f"{repo_api_url(h, o, r)}/pulls" payload = {"title": title, "body": body, "head": head, "base": base} diff --git a/review_proofs.py b/review_proofs.py index a0e4fed..bab0f00 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -527,6 +527,35 @@ def assess_controller_handoff(report_text, role=None): "reasons": [f"handoff missing required field: {m}" for m in missing], } + + # Validate issue/PR references for exact number and no forbidden terms (Issue #194 / #196) + fields_dict = {} + for line in section: + stripped = line.strip().lstrip("-*").strip() + if ":" in stripped: + k, v = stripped.split(":", 1) + fields_dict[k.strip().lower()] = v.strip() + + for alias in ("selected issue", "pr number opened", "pr opened", "pr number", "selected pr"): + val = fields_dict.get(alias) + if val: + numbers = re.findall(r"\d+", val) + has_forbidden = any(term in val.lower() for term in ("equivalent", "related", "same", "/")) + if len(numbers) != 1 or has_forbidden: + field_name = "Selected issue/PR" + for name, aliases in list(HANDOFF_BASE_FIELDS) + list(HANDOFF_ROLE_FIELDS.get(role or "", ())): + if alias in aliases: + field_name = name + break + return { + "verdict": "incomplete", + "downgraded": True, + "missing_fields": [field_name], + "reasons": [ + f"{field_name} must specify exactly one number and no ambiguous references (got: '{val}')" + ], + } + return { "verdict": "complete", "downgraded": False, diff --git a/scripts/worktree-start b/scripts/worktree-start index cd90992..09c3bee 100755 --- a/scripts/worktree-start +++ b/scripts/worktree-start @@ -40,6 +40,16 @@ start_ref="${2:-prgs/master}" # Enforce issue-linked, traceable branch names (issue → branch → worktree → PR). if [[ "$allow_unlinked" -eq 0 ]]; then + if [[ ! -f "/tmp/gitea_issue_lock.json" ]]; then + echo "Error: Issue lock file '/tmp/gitea_issue_lock.json' is missing. You must lock exactly one issue before branch creation (fail closed)." >&2 + exit 2 + fi + locked_branch=$(python3 -c "import json; print(json.load(open('/tmp/gitea_issue_lock.json')).get('branch_name', ''))") + if [[ "$branch" != "$locked_branch" ]]; then + echo "Error: Requested branch '$branch' does not match locked branch '$locked_branch' (fail closed)." >&2 + exit 2 + fi + if [[ "$branch" =~ ^(fix|feat|docs|chore)/issue-[0-9]+-.+ ]] \ || [[ "$branch" =~ ^review/pr-[0-9]+-.+ ]]; then : diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index f1c1cce..d9f1029 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -33,6 +33,7 @@ from mcp_server import ( # noqa: E402 gitea_submit_pr_review, gitea_list_issue_comments, gitea_create_issue_comment, + gitea_lock_issue, ) from gitea_auth import get_profile # noqa: E402 import gitea_config # noqa: E402 @@ -90,10 +91,13 @@ class TestCreatePR(unittest.TestCase): @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_creates_pr(self, _auth, mock_api): + @patch("os.path.exists", return_value=True) + @patch("builtins.open") + def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api): + mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}' mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"} with patch.dict(os.environ, {}, clear=True): - result = gitea_create_pr(title="feat: X", head="feat/x", base="main") + result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main") self.assertEqual(result["number"], 3) self.assertNotIn("url", result) payload = mock_api.call_args[0][3] @@ -102,10 +106,13 @@ class TestCreatePR(unittest.TestCase): @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api): + @patch("os.path.exists", return_value=True) + @patch("builtins.open") + def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api): + mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}' mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"} with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True): - result = gitea_create_pr(title="feat: X", head="feat/x", base="main") + result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main") self.assertIn("pulls/3", result["url"]) @@ -2388,3 +2395,83 @@ class TestVerifyMutationAuthority(unittest.TestCase): # Should pass without exception _real_verify("prgs") + + +class TestIssueLocking(unittest.TestCase): + """Test issue locking and PR gating constraints.""" + + def tearDown(self): + if os.path.exists("/tmp/gitea_issue_lock.json"): + os.remove("/tmp/gitea_issue_lock.json") + + @patch("mcp_server.api_get_all") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_lock_issue_success(self, _auth, mock_api): + mock_api.return_value = [] # no open PRs + res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs") + self.assertTrue(res["success"]) + self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json")) + + def test_lock_issue_mismatch_branch_fails(self): + with self.assertRaises(ValueError) as ctx: + gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs") + self.assertIn("must contain locked issue pattern", str(ctx.exception)) + + @patch("mcp_server.api_get_all") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api): + mock_api.return_value = [{ + "number": 200, + "head": {"ref": "feat/issue-196-boundary"}, + "title": "Some PR", + "body": "No closes ref" + }] + with self.assertRaises(ValueError) as ctx: + gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs") + self.assertIn("already tied to an open PR", str(ctx.exception)) + + @patch("mcp_server.api_get_all") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api): + mock_api.return_value = [{ + "number": 200, + "head": {"ref": "feat/other-branch"}, + "title": "Some PR", + "body": "fixes #196" + }] + with self.assertRaises(ValueError) as ctx: + gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs") + self.assertIn("already tied to an open PR", str(ctx.exception)) + + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_create_pr_missing_lock_fails(self, _auth): + if os.path.exists("/tmp/gitea_issue_lock.json"): + os.remove("/tmp/gitea_issue_lock.json") + with self.assertRaises(RuntimeError) as ctx: + gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs") + self.assertIn("Issue lock is missing", str(ctx.exception)) + + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_create_pr_branch_mismatch_fails(self, _auth): + with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: + json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) + with self.assertRaises(ValueError) as ctx: + gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs") + self.assertIn("does not match locked branch", str(ctx.exception)) + + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_create_pr_forbidden_terms_fails(self, _auth): + with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: + json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) + for term in ("equivalent to #196", "related to #196", "same as #196"): + with self.assertRaises(ValueError) as ctx: + gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs") + self.assertIn("contains forbidden term", str(ctx.exception)) + + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_create_pr_missing_closes_ref_fails(self, _auth): + with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: + json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) + with self.assertRaises(ValueError) as ctx: + gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs") + self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception)) diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 07dfc40..68d90d1 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -657,6 +657,40 @@ class TestControllerHandoff(unittest.TestCase): self.assertEqual(result["verdict"], "incomplete") self.assertIn("No review/merge confirmation", result["missing_fields"]) + def test_author_role_rejects_equivalent_or_multiple_issues(self): + # 1. equivalent reference blocked + incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([ + "- Selected issue: Issue #194 / #196 equivalent", + "- Claim/comment status: comment-claimed", + "- PR number opened: #999", + "- No review/merge: confirmed", + ]) + res = assess_controller_handoff(incomplete_eq, role="author") + self.assertEqual(res["verdict"], "incomplete") + self.assertIn("Selected issue", res["missing_fields"]) + + # 2. multiple issues blocked + incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([ + "- Selected issue: #194, #196", + "- Claim/comment status: comment-claimed", + "- PR number opened: #999", + "- No review/merge: confirmed", + ]) + res = assess_controller_handoff(incomplete_multi, role="author") + self.assertEqual(res["verdict"], "incomplete") + self.assertIn("Selected issue", res["missing_fields"]) + + def test_author_role_rejects_fuzzy_pr_number(self): + incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([ + "- Selected issue: #196", + "- Claim/comment status: comment-claimed", + "- PR number opened: PR #203 / #204 equivalent", + "- No review/merge: confirmed", + ]) + res = assess_controller_handoff(incomplete_pr, role="author") + self.assertEqual(res["verdict"], "incomplete") + self.assertIn("PR number opened", res["missing_fields"]) + def test_inventory_role_requires_inventory_fields(self): complete = self.BASE_HANDOFF + "\n" + "\n".join([ "- Repositories checked: Gitea-Tools, mcp-control-plane", diff --git a/tests/test_worktrees.py b/tests/test_worktrees.py index 34d89a9..45f1ac5 100644 --- a/tests/test_worktrees.py +++ b/tests/test_worktrees.py @@ -14,11 +14,39 @@ BRANCHES = REPO / "branches" def run(script, *args): - proc = subprocess.run( - ["bash", str(SCRIPTS / script), *args], - capture_output=True, text=True, cwd=str(REPO), - ) - return proc.returncode, proc.stdout, proc.stderr + branch = None + for arg in args: + if not arg.startswith("-"): + branch = arg + break + + lock_file = Path("/tmp/gitea_issue_lock.json") + created_lock = False + if script == "worktree-start" and branch: + import re + import json + m = re.search(r"issue-(\d+)", branch) + if not m: + m = re.search(r"pr-(\d+)", branch) + issue_num = int(m.group(1)) if m else 999 + lock_file.write_text(json.dumps({ + "issue_number": issue_num, + "branch_name": branch, + "remote": "prgs", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools" + }), encoding="utf-8") + created_lock = True + + try: + proc = subprocess.run( + ["bash", str(SCRIPTS / script), *args], + capture_output=True, text=True, cwd=str(REPO), + ) + return proc.returncode, proc.stdout, proc.stderr + finally: + if created_lock and lock_file.exists(): + lock_file.unlink() class TestWorktreeStart(unittest.TestCase): -- 2.43.7