"""Regression matrix for Issue #733. ``gitea_delete_branch`` accepts explicit ``org``/``repo`` but historically did not propagate them through the anti-stomp preflight: preflight resolved the remote-wide ``REMOTES`` default (bare ``prgs`` → ``Scaled-Tech-Consulting/ Timesheet``) and fail-closed with ``wrong_repo`` even though the caller explicitly targeted ``Scaled-Tech-Consulting/Gitea-Tools``. The fix has two parts, both exercised here: 1. ``gitea_delete_branch`` forwards the explicit ``org``/``repo`` into ``verify_preflight_purity`` so the shared #604 anti-stomp resolution validates the *targeted* repository instead of the remote-wide default. 2. A workspace-derived repository-binding gate (``_delete_branch_repository_binding_block``) validates explicit coordinates against the immutable workspace identity, rejecting wrong, substituted, or unverified targets — independent of the anti-stomp remote/repo guard, which by the #530 contract trusts explicit intent. Every existing gate (reconciler-only ownership; author/reviewer/merger denial; protected/preservation blocks) is preserved. """ import os import sys import unittest from unittest.mock import patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) import mcp_server from mcp_server import gitea_delete_branch import remote_repo_guard import task_capability_map import role_session_router FAKE_AUTH = "token fake" # The workspace is bound to Gitea-Tools even though the prgs REMOTES default # repo is Timesheet (the exact #733 scenario). WORKSPACE_SLUG = "Scaled-Tech-Consulting/Gitea-Tools" LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" RECONCILER = { "profile_name": "prgs-reconciler", "role": "reconciler", "allowed_operations": [ "gitea.read", "gitea.issue.comment", "gitea.pr.comment", "gitea.branch.delete", ], "forbidden_operations": ["gitea.pr.create", "gitea.pr.merge"], "audit_label": "prgs-reconciler", } AUTHOR_WITH_DELETE = { "profile_name": "prgs-author", "role": "author", "allowed_operations": [ "gitea.read", "gitea.pr.create", "gitea.branch.push", "gitea.branch.delete", ], "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"], "audit_label": "prgs-author", } REVIEWER_WITH_DELETE = { "profile_name": "prgs-reviewer", "role": "reviewer", "allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.branch.delete"], "forbidden_operations": [], "audit_label": "prgs-reviewer", } MERGER_WITH_DELETE = { "profile_name": "prgs-merger", "role": "merger", "allowed_operations": ["gitea.read", "gitea.pr.merge", "gitea.branch.delete"], "forbidden_operations": [], "audit_label": "prgs-merger", } class _DeleteBranchBase(unittest.TestCase): def setUp(self): self._snap = ( mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called, ) mcp_server._preflight_whoami_called = False mcp_server._preflight_capability_called = False # prgs default repo is Timesheet — the remote-wide default #733 must NOT # fall back to. self._remotes = patch.dict(mcp_server.REMOTES, { "prgs": { "host": "gitea.prgs.cc", "org": "Scaled-Tech-Consulting", "repo": "Timesheet", }, }) self._remotes.start() mcp_server._IDENTITY_CACHE.clear() patch("gitea_config.load_config", return_value={}).start() patch("gitea_config.is_runtime_switching_enabled", return_value=False).start() patch("gitea_audit.audit_enabled", return_value=False).start() self.mock_api = patch("mcp_server.api_request").start() self.mock_api.return_value = {} patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start() # Workspace is verifiably bound to Gitea-Tools. patch( "mcp_server._workspace_repository_slug", return_value=WORKSPACE_SLUG ).start() def tearDown(self): patch.stopall() mcp_server._IDENTITY_CACHE.clear() ( mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called, ) = self._snap def _bind(self, profile, role): patch("mcp_server.get_profile", return_value=profile).start() mcp_server.record_preflight_check("whoami") mcp_server.record_preflight_check("capability", resolved_role=role) def _delete_calls(self): return [ c for c in self.mock_api.call_args_list if c.args and c.args[0] == "DELETE" ] class TestPositiveTimesheetDefaultExplicitGiteaTools(_DeleteBranchBase): """AC: prgs default=Timesheet + explicit Gitea-Tools → preflight validates Gitea-Tools and permits an otherwise-eligible deletion.""" def test_explicit_gitea_tools_permits_delete_and_forwards_coords(self): self._bind(RECONCILER, "reconciler") with patch("mcp_server.verify_preflight_purity") as vpp: res = gitea_delete_branch( branch="feat/pr-sync-status", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertTrue(res["success"], res) self.assertIn("deleted", res["message"]) self.assertTrue(self._delete_calls()) # Forwarding: preflight received the explicit Gitea-Tools coordinates # and the delete_branch task (not the remote-wide Timesheet default). self.assertTrue(vpp.called) kwargs = vpp.call_args.kwargs self.assertEqual(kwargs.get("task"), "delete_branch") self.assertEqual(kwargs.get("org"), "Scaled-Tech-Consulting") self.assertEqual(kwargs.get("repo"), "Gitea-Tools") class TestNegativeRepositoryBinding(_DeleteBranchBase): """AC negatives: wrong / substituted / unverified coordinates fail closed.""" def test_wrong_explicit_repo_fails_closed(self): self._bind(RECONCILER, "reconciler") with patch("mcp_server.verify_preflight_purity") as vpp: res = gitea_delete_branch( branch="feat/x", remote="prgs", org="Scaled-Tech-Consulting", repo="Timesheet", # not the workspace-bound Gitea-Tools ) self.assertFalse(res["success"]) self.assertFalse(res["performed"]) self.assertEqual(res["blocker_kind"], "repository_binding") self.assertTrue(any("Timesheet" in r for r in res["reasons"])) self.assertFalse(self._delete_calls()) # Fail closed BEFORE any preflight/deletion side effect. vpp.assert_not_called() def test_repository_substitution_org_mismatch_rejected(self): self._bind(RECONCILER, "reconciler") res = gitea_delete_branch( branch="feat/x", remote="prgs", org="Some-Other-Org", repo="Gitea-Tools", ) self.assertFalse(res["success"]) self.assertEqual(res["blocker_kind"], "repository_binding") self.assertFalse(self._delete_calls()) def test_explicit_coords_without_workspace_identity_fail_closed(self): self._bind(RECONCILER, "reconciler") with patch("mcp_server._workspace_repository_slug", return_value=None): res = gitea_delete_branch( branch="feat/x", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertFalse(res["success"]) self.assertEqual(res["blocker_kind"], "repository_binding") self.assertTrue(any("unverified" in r for r in res["reasons"])) self.assertFalse(self._delete_calls()) def test_matching_explicit_passes_binding(self): self._bind(RECONCILER, "reconciler") block = mcp_server._delete_branch_repository_binding_block( "prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools" ) self.assertIsNone(block) def test_omitted_coords_pass_binding_and_are_revalidated_by_preflight(self): # Omitted coordinates pass the binding gate (nothing to corroborate) but # are forwarded as None so the anti-stomp resolution fails closed on the # remote-wide Timesheet default (see TestAntiStompResolution below). self._bind(RECONCILER, "reconciler") self.assertIsNone( mcp_server._delete_branch_repository_binding_block( "prgs", org=None, repo=None ) ) with patch("mcp_server.verify_preflight_purity") as vpp: gitea_delete_branch(branch="feat/x", remote="prgs") self.assertTrue(vpp.called) kwargs = vpp.call_args.kwargs self.assertIsNone(kwargs.get("org")) self.assertIsNone(kwargs.get("repo")) class TestRoleDenials(_DeleteBranchBase): """AC: author, reviewer, and merger remain denied even holding the perm.""" def test_author_with_delete_perm_denied(self): self._bind(AUTHOR_WITH_DELETE, "author") res = gitea_delete_branch( branch="feat/x", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertFalse(res["success"]) self.assertEqual(res["required_role_kind"], "reconciler") self.assertEqual(res["active_role_kind"], "author") self.assertFalse(self._delete_calls()) def test_reviewer_with_delete_perm_denied(self): self._bind(REVIEWER_WITH_DELETE, "reviewer") res = gitea_delete_branch( branch="feat/x", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertFalse(res["success"]) self.assertEqual(res["required_role_kind"], "reconciler") self.assertFalse(self._delete_calls()) def test_merger_with_delete_perm_denied(self): self._bind(MERGER_WITH_DELETE, "merger") res = gitea_delete_branch( branch="feat/x", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertFalse(res["success"]) self.assertEqual(res["required_role_kind"], "reconciler") self.assertFalse(self._delete_calls()) class TestProtectedAndPreservedBranches(_DeleteBranchBase): """AC: protected and preservation/evidence branches remain blocked.""" def test_protected_branch_blocked(self): self._bind(RECONCILER, "reconciler") for branch in ("master", "main", "dev"): with self.subTest(branch=branch): res = gitea_delete_branch( branch=branch, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertFalse(res["success"]) self.assertTrue(any("protected" in r for r in res["reasons"])) self.assertFalse(self._delete_calls()) def test_preservation_branch_blocked(self): self._bind(RECONCILER, "reconciler") for branch in ("chore/preserve-local-master", "feat/evidence-run"): with self.subTest(branch=branch): res = gitea_delete_branch( branch=branch, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", ) self.assertFalse(res["success"]) self.assertTrue( any("preservation" in r for r in res["reasons"]) ) self.assertFalse(self._delete_calls()) class TestAntiStompResolution(unittest.TestCase): """The anti-stomp repository resolution validates the *targeted* repo and fails closed on the remote-wide default when coordinates are omitted.""" def setUp(self): self._remotes = patch.dict(mcp_server.REMOTES, { "prgs": { "host": "gitea.prgs.cc", "org": "Scaled-Tech-Consulting", "repo": "Timesheet", }, }) self._remotes.start() def tearDown(self): patch.stopall() def _capture_resolution(self, org, repo): """Drive the live anti-stomp runner and capture the org/repo it resolved and passed to the pure assessor.""" passthrough = { "allowed": True, "block": False, "blockers": [], "reasons": [], "exact_next_action": "proceed", "blocker_kind": None, "checks": {}, } with patch.dict( os.environ, {"GITEA_TEST_FORCE_ANTI_STOMP": "1"}, clear=False ), patch( "mcp_server._local_git_remote_url", return_value=LOCAL_GITEA_TOOLS_URL ), patch( "mcp_server.get_profile", return_value=RECONCILER ), patch.object( mcp_server.anti_stomp_preflight, "assess_anti_stomp_preflight", return_value=passthrough, ) as m: mcp_server._run_anti_stomp_preflight( "delete_branch", remote="prgs", org=org, repo=repo ) self.assertTrue(m.called) return m.call_args.kwargs def test_explicit_gitea_tools_resolved_and_marked_explicit(self): kw = self._capture_resolution("Scaled-Tech-Consulting", "Gitea-Tools") self.assertEqual(kw["resolved_org"], "Scaled-Tech-Consulting") self.assertEqual(kw["resolved_repo"], "Gitea-Tools") self.assertTrue(kw["org_explicit"]) self.assertTrue(kw["repo_explicit"]) def test_missing_coords_resolve_remote_default_and_fail_closed(self): # Omitted coords resolve the remote-wide Timesheet default with # org/repo NOT explicit; the remote/repo guard then blocks against the # local Gitea-Tools remote — i.e. missing coordinates fail closed. kw = self._capture_resolution(None, None) self.assertEqual(kw["resolved_repo"], "Timesheet") self.assertFalse(kw["org_explicit"]) self.assertFalse(kw["repo_explicit"]) # Compose the guard exactly as the assessor would: Timesheet default, # not explicit, local remote is Gitea-Tools → blocked (wrong_repo). assessment = remote_repo_guard.assess_remote_repo_match( remote="prgs", resolved_org=kw["resolved_org"], resolved_repo=kw["resolved_repo"], local_remote_url=LOCAL_GITEA_TOOLS_URL, org_explicit=kw["org_explicit"], repo_explicit=kw["repo_explicit"], ) self.assertTrue(assessment["block"]) class TestCapabilityRoleMapConsistency(unittest.TestCase): """AC: task-capability and role-routing maps remain consistent (#729).""" def test_delete_branch_is_reconciler_in_both_maps(self): self.assertEqual( task_capability_map.required_role("delete_branch"), "reconciler" ) self.assertEqual( task_capability_map.required_permission("delete_branch"), "gitea.branch.delete", ) self.assertEqual( role_session_router.required_role_for_task("delete_branch"), "reconciler", ) self.assertEqual( task_capability_map.required_role("delete_branch"), role_session_router.TASK_REQUIRED_ROLE["delete_branch"], ) if __name__ == "__main__": unittest.main()