import sys import unittest from unittest.mock import patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) import branch_cleanup_guard as guard # noqa: E402 import mcp_server # noqa: E402 import task_capability_map # noqa: E402 from final_report_validator import assess_final_report_validator # noqa: E402 from mcp_server import gitea_cleanup_merged_pr_branch, gitea_delete_branch # noqa: E402 FAKE_AUTH = "token fake" # Reconciler-shaped profile that holds branch.delete (recommended) plus # required pr.close/read so _role_kind classifies as reconciler. RECONCILER_WITH_DELETE = { "profile_name": "prgs-reconciler", "role": "reconciler", "allowed_operations": [ "gitea.read", "gitea.pr.close", "gitea.pr.comment", "gitea.issue.comment", "gitea.issue.close", "gitea.branch.delete", ], "forbidden_operations": [ "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", "gitea.pr.create", "gitea.branch.push", "gitea.repo.commit", ], } class TestRawBranchDeleteGuard(unittest.TestCase): def test_detects_local_and_remote_raw_git_delete_commands(self): text = """ Cleanup mutations: - git branch -d feat/issue-485-lease-comments-non-list-guard - git push prgs --delete feat/issue-485-lease-comments-non-list-guard """ commands = guard.raw_branch_delete_commands(text) self.assertEqual(len(commands), 2) self.assertIn("git branch -d", commands[0]) self.assertIn("git push prgs --delete", commands[1]) def test_final_report_blocks_raw_delete_as_cleanup_proof(self): report = """ ## Controller Handoff - Task: review_pr - Cleanup mutations: git push prgs --delete feat/branch """ result = assess_final_report_validator(report, "review_pr") self.assertTrue(result["blocked"]) self.assertTrue( any("shared.raw_branch_delete_bypass" in r for r in result["reasons"]) ) def test_cleanup_task_requires_branch_delete_capability(self): self.assertEqual( task_capability_map.required_permission("cleanup_merged_pr_branch"), "gitea.branch.delete", ) self.assertEqual( task_capability_map.required_role("cleanup_merged_pr_branch"), "reconciler", ) class TestMergedPrBranchCleanupTool(unittest.TestCase): def setUp(self): self._remotes = patch.dict( mcp_server.REMOTES, { "prgs": { "host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo", } }, ) self._remotes.start() patch("gitea_audit.audit_enabled", return_value=False).start() self.mock_api = patch("mcp_server.api_request").start() self.mock_all = patch("mcp_server.api_get_all", return_value=[]).start() patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start() patch( "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref", return_value=True, ).start() # Default: no active ownership records (tests that need ownership patch this). patch( "mcp_server._collect_branch_ownership_records", return_value=[], ).start() def tearDown(self): patch.stopall() def test_reviewer_without_delete_authority_fails_closed(self): patch( "mcp_server.get_profile", return_value={ "profile_name": "reviewer", "allowed_operations": ["gitea.read", "gitea.pr.review"], "forbidden_operations": ["gitea.branch.delete"], }, ).start() res = gitea_cleanup_merged_pr_branch( pr_number=487, confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch", branch="feat/branch", remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) self.assertFalse(res["performed"]) self.assertEqual(res["required_permission"], "gitea.branch.delete") self.mock_api.assert_not_called() def test_author_and_merger_without_delete_authority_fail_closed(self): role_profiles = { "author": [ "gitea.read", "gitea.branch.create", "gitea.branch.push", "gitea.repo.commit", "gitea.pr.create", ], "merger": ["gitea.read", "gitea.pr.merge"], } for name, allowed in role_profiles.items(): with self.subTest(role=name): profile_patch = patch( "mcp_server.get_profile", return_value={ "profile_name": name, "allowed_operations": allowed, "forbidden_operations": [], }, ) profile_patch.start() try: res = gitea_cleanup_merged_pr_branch( pr_number=487, confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch", branch="feat/branch", remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) finally: profile_patch.stop() self.assertFalse(res["performed"]) self.assertEqual( res["required_permission"], "gitea.branch.delete" ) self.mock_api.assert_not_called() def test_root_checkout_cleanup_fails_closed(self): patch( "mcp_server.get_profile", return_value=dict(RECONCILER_WITH_DELETE), ).start() res = gitea_cleanup_merged_pr_branch( pr_number=487, confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch", branch="feat/branch", remote="prgs", worktree_path="/tmp/repo/Gitea-Tools", ) self.assertFalse(res["performed"]) self.assertIn("root checkout", " ".join(res["reasons"])) self.mock_api.assert_not_called() def test_authorized_cleanup_deletes_through_api_path(self): branch = "feat/issue-485-lease-comments-non-list-guard" patch( "mcp_server.get_profile", return_value=dict(RECONCILER_WITH_DELETE), ).start() def _api(method, url, *args, **kwargs): if method == "GET" and "/pulls/" in url: return { "number": 487, "merged": True, "merged_at": "2026-07-08T01:00:00Z", "head": {"ref": branch, "sha": "a" * 40}, "base": {"ref": "master"}, } if method == "GET" and "/branches/" in url: # First pre-delete probe: present. Post-delete: not found. get_branch_calls = [ c for c in self.mock_api.call_args_list if c.args and c.args[0] == "GET" and "/branches/" in c.args[1] ] if len(get_branch_calls) <= 1: return {"name": branch} raise RuntimeError("HTTP 404: not found") if method == "DELETE": return {} raise AssertionError(f"unexpected {method} {url}") self.mock_api.side_effect = _api res = gitea_cleanup_merged_pr_branch( pr_number=487, confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", branch=branch, remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) self.assertTrue(res["success"]) self.assertTrue(res["performed"]) self.assertTrue((res.get("readback") or {}).get("verified_absent")) delete_calls = [ call for call in self.mock_api.call_args_list if call.args[0] == "DELETE" ] self.assertEqual(len(delete_calls), 1) self.assertIn("branches/feat%2Fissue-485", delete_calls[0].args[1]) def test_confirmation_required_before_delete(self): branch = "feat/issue-485-lease-comments-non-list-guard" patch( "mcp_server.get_profile", return_value=dict(RECONCILER_WITH_DELETE), ).start() self.mock_api.side_effect = [ { "number": 487, "merged": True, "merged_at": "2026-07-08T01:00:00Z", "head": {"ref": branch, "sha": "a" * 40}, "base": {"ref": "master"}, }, {}, ] res = gitea_cleanup_merged_pr_branch( pr_number=487, confirmation="wrong", branch=branch, remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) self.assertFalse(res["performed"]) self.assertIn("confirmation", " ".join(res["reasons"])) delete_calls = [ call for call in self.mock_api.call_args_list if call.args[0] == "DELETE" ] self.assertFalse(delete_calls) def test_reconciler_with_branch_delete_cannot_raw_delete(self): """#687: reconciler + gitea.branch.delete still cannot call raw delete.""" patch( "mcp_server.get_profile", return_value=dict(RECONCILER_WITH_DELETE), ).start() res = gitea_delete_branch( branch="fix/issue-683-workflow-guard-hardening", remote="prgs", ) self.assertFalse(res.get("success", True)) self.assertFalse(res.get("performed", True)) reasons = " ".join(res.get("reasons") or []) self.assertIn("raw gitea_delete_branch", reasons) self.assertIn("cleanup_merged_pr_branch", reasons) self.mock_api.assert_not_called() def test_reconciler_raw_delete_denies_preservation_branch(self): patch( "mcp_server.get_profile", return_value=dict(RECONCILER_WITH_DELETE), ).start() res = gitea_delete_branch( branch="chore/issue-681-preserve-review-session-wip", remote="prgs", ) self.assertFalse(res.get("performed", True)) self.mock_api.assert_not_called() def test_author_with_branch_delete_role_ok_but_preserve_blocked(self): """Author role may use raw delete path when permitted; preserve fails closed.""" patch( "mcp_server.get_profile", return_value={ "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"], }, ).start() res = gitea_delete_branch( branch="chore/issue-681-preserve-review-session-wip", remote="prgs", ) self.assertFalse(res.get("performed", True)) self.assertIn("preservation", " ".join(res.get("reasons") or [])) self.mock_api.assert_not_called() def test_unmerged_branch_cleanup_rejected(self): branch = "feat/unmerged-work" patch( "mcp_server.get_profile", return_value=dict(RECONCILER_WITH_DELETE), ).start() self.mock_api.side_effect = [ { "number": 999, "merged": False, "merged_at": None, "head": {"ref": branch, "sha": "b" * 40}, "base": {"ref": "master"}, }, {}, ] res = gitea_cleanup_merged_pr_branch( pr_number=999, confirmation=f"CLEANUP MERGED PR 999 BRANCH {branch}", branch=branch, remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) self.assertFalse(res["performed"]) self.assertTrue( any("not merged" in r for r in (res.get("reasons") or [])) ) delete_calls = [ call for call in self.mock_api.call_args_list if call.args[0] == "DELETE" ] self.assertFalse(delete_calls) def test_preservation_branch_cleanup_rejected(self): branch = "chore/issue-681-preserve-review-session-wip" patch( "mcp_server.get_profile", return_value=dict(RECONCILER_WITH_DELETE), ).start() self.mock_api.side_effect = [ { "number": 681, "merged": True, "merged_at": "2026-07-08T01:00:00Z", "head": {"ref": branch, "sha": "c" * 40}, "base": {"ref": "master"}, }, {}, ] res = gitea_cleanup_merged_pr_branch( pr_number=681, confirmation=f"CLEANUP MERGED PR 681 BRANCH {branch}", branch=branch, remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) self.assertFalse(res["performed"]) self.assertTrue( any("preservation" in r for r in (res.get("reasons") or [])) ) delete_calls = [ call for call in self.mock_api.call_args_list if call.args[0] == "DELETE" ] self.assertFalse(delete_calls) def test_non_reconciler_with_delete_denied_cleanup(self): patch( "mcp_server.get_profile", return_value={ "profile_name": "prgs-author", "role": "author", "allowed_operations": [ "gitea.read", "gitea.pr.create", "gitea.branch.push", "gitea.branch.delete", ], "forbidden_operations": [], }, ).start() res = gitea_cleanup_merged_pr_branch( pr_number=487, confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch", branch="feat/branch", remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) self.assertFalse(res["performed"]) self.assertEqual(res.get("required_role_kind"), "reconciler") self.mock_api.assert_not_called() def test_assess_guard_rejects_preservation_branch(self): assessment = guard.assess_merged_pr_branch_cleanup( pr_number=681, head_branch="chore/issue-681-preserve-review-session-wip", merged=True, remote_branch_exists=True, open_pr_heads=set(), head_on_target=True, delete_capability_allowed=True, confirmation=( "CLEANUP MERGED PR 681 BRANCH " "chore/issue-681-preserve-review-session-wip" ), ) self.assertFalse(assessment["safe_to_delete"]) self.assertTrue( any("preservation" in r for r in assessment["block_reasons"]) ) class TestPostDeleteReadback(unittest.TestCase): def test_not_found_is_verified_success(self): readback = guard.classify_branch_readback_http_status(404) result = guard.assess_post_delete_readback(readback) self.assertTrue(result["ok"]) self.assertTrue(result["readback"]["verified_absent"]) def test_exists_is_structured_failure(self): readback = guard.classify_branch_readback_http_status(200) result = guard.assess_post_delete_readback(readback) self.assertFalse(result["ok"]) self.assertFalse(result["readback"]["verified_absent"]) self.assertTrue(result["readback"]["branch_present"]) self.assertIn("still present", " ".join(result["reasons"])) def test_auth_failure_preserved(self): readback = guard.classify_branch_readback_http_status(401) result = guard.assess_post_delete_readback(readback) self.assertFalse(result["ok"]) self.assertEqual(result["readback"]["error_class"], "authentication") self.assertIn("authentication", " ".join(result["reasons"])) def test_authz_and_transport_failures(self): for code, err in ((403, "authorization"), (503, "transport")): with self.subTest(code=code): readback = guard.classify_branch_readback_http_status(code) result = guard.assess_post_delete_readback(readback) self.assertFalse(result["ok"]) self.assertEqual(result["readback"]["error_class"], err) def test_exception_classifier_no_secret_leak(self): class FakeHTTPError(Exception): def __init__(self): super().__init__("HTTP 401: token=super-secret-value Authorization: Bearer xyz") self.code = 401 result = guard.classify_branch_readback_exception(FakeHTTPError()) blob = str(result) self.assertNotIn("super-secret", blob) self.assertNotIn("Bearer", blob) self.assertEqual(result["error_class"], "authentication") class TestActiveBranchOwnership(unittest.TestCase): def _base(self, **overrides): rec = { "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, "status": "active", "remote": "prgs", "org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools", "branch": "feat/target", "reclaim_allowed": False, } rec.update(overrides) return rec def test_active_author_blocks(self): result = guard.assess_active_branch_ownership( remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", records=[self._base()], ) self.assertTrue(result["block"]) self.assertIn(guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, result["blocking_categories"]) def test_active_reviewer_lease_blocks(self): result = guard.assess_active_branch_ownership( remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", records=[ self._base( category=guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, status="active", ) ], ) self.assertTrue(result["block"]) self.assertIn(guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, result["blocking_categories"]) def test_merger_controller_reconciler_binding_blocks(self): for cat in ( guard.OWNERSHIP_CATEGORY_MERGER_LEASE, guard.OWNERSHIP_CATEGORY_CONTROLLER_LEASE, guard.OWNERSHIP_CATEGORY_RECONCILER_LEASE, guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING, ): with self.subTest(category=cat): result = guard.assess_active_branch_ownership( remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", records=[self._base(category=cat, status="active")], ) self.assertTrue(result["block"]) self.assertIn(cat, result["blocking_categories"]) def test_released_and_expired_reclaimable_do_not_block(self): records = [ self._base(status="released", reclaim_allowed=True), self._base( category=guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, status="expired", reclaim_allowed=True, ), ] result = guard.assess_active_branch_ownership( remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", records=records, ) self.assertFalse(result["block"]) def test_sticky_stale_blocks_per_recovery_policy(self): result = guard.assess_active_branch_ownership( remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", records=[self._base(status="stale", reclaim_allowed=False)], ) self.assertTrue(result["block"]) self.assertIn("sticky", " ".join(result["reasons"])) def test_other_repo_or_branch_no_false_block(self): records = [ self._base(repo="Other-Repo", status="active"), self._base(branch="feat/other", status="active"), self._base(remote="dadeschools", status="active"), ] result = guard.assess_active_branch_ownership( remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", records=records, ) self.assertFalse(result["block"]) self.assertEqual(len(result["ignored_out_of_scope"]), 3) def test_denial_has_no_secrets(self): result = guard.assess_active_branch_ownership( remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", records=[ self._base( status="active", # Poison fields that must never be echoed as secrets token="sekrit-token", authorization="Bearer abc", ) ], ) blob = str(result) self.assertNotIn("sekrit", blob) self.assertNotIn("Bearer", blob) self.assertIn("author_lease", blob) class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase): def setUp(self): self._remotes = patch.dict( mcp_server.REMOTES, { "prgs": { "host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo", } }, ) self._remotes.start() patch("gitea_audit.audit_enabled", return_value=False).start() self.mock_api = patch("mcp_server.api_request").start() patch("mcp_server.api_get_all", return_value=[]).start() patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start() patch( "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref", return_value=True, ).start() patch( "mcp_server.get_profile", return_value=dict(RECONCILER_WITH_DELETE), ).start() def tearDown(self): patch.stopall() def _pr_payload(self, branch, number=487): return { "number": number, "merged": True, "merged_at": "2026-07-08T01:00:00Z", "head": {"ref": branch, "sha": "a" * 40}, "base": {"ref": "master"}, } def test_delete_success_but_branch_remains(self): branch = "feat/still-there" patch( "mcp_server._collect_branch_ownership_records", return_value=[], ).start() def _api(method, url, *a, **k): if method == "GET" and "/pulls/" in url: return self._pr_payload(branch) if method == "GET" and "/branches/" in url: return {"name": branch} # present before and after if method == "DELETE": return {} raise AssertionError(method) self.mock_api.side_effect = _api res = gitea_cleanup_merged_pr_branch( pr_number=487, confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", branch=branch, remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) self.assertTrue(res.get("performed")) self.assertFalse(res.get("success")) self.assertTrue(res.get("delete_acknowledged")) self.assertIn("still present", " ".join(res.get("reasons") or [])) def test_readback_authentication_failure(self): branch = "feat/auth-fail-readback" patch( "mcp_server._collect_branch_ownership_records", return_value=[], ).start() state = {"branch_gets": 0} def _api(method, url, *a, **k): if method == "GET" and "/pulls/" in url: return self._pr_payload(branch) if method == "GET" and "/branches/" in url: state["branch_gets"] += 1 if state["branch_gets"] == 1: return {"name": branch} raise RuntimeError("HTTP 401: unauthorized") if method == "DELETE": return {} raise AssertionError(method) self.mock_api.side_effect = _api res = gitea_cleanup_merged_pr_branch( pr_number=487, confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", branch=branch, remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) self.assertTrue(res.get("performed")) self.assertFalse(res.get("success")) self.assertEqual((res.get("readback") or {}).get("error_class"), "authentication") def test_readback_transport_failure(self): branch = "feat/transport-fail-readback" patch( "mcp_server._collect_branch_ownership_records", return_value=[], ).start() state = {"branch_gets": 0} def _api(method, url, *a, **k): if method == "GET" and "/pulls/" in url: return self._pr_payload(branch) if method == "GET" and "/branches/" in url: state["branch_gets"] += 1 if state["branch_gets"] == 1: return {"name": branch} raise RuntimeError("HTTP 503: temporarily unavailable") if method == "DELETE": return {} raise AssertionError(method) self.mock_api.side_effect = _api res = gitea_cleanup_merged_pr_branch( pr_number=487, confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", branch=branch, remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) self.assertFalse(res.get("success")) self.assertEqual((res.get("readback") or {}).get("error_class"), "transport") def test_active_author_ownership_blocks_before_delete(self): branch = "feat/owned" patch( "mcp_server._collect_branch_ownership_records", return_value=[ { "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, "status": "active", "remote": "prgs", "org": "Example-Org", "repo": "Example-Repo", "branch": branch, "reclaim_allowed": False, } ], ).start() def _api(method, url, *a, **k): if method == "GET" and "/pulls/" in url: return self._pr_payload(branch) if method == "GET" and "/branches/" in url: return {"name": branch} raise AssertionError(f"unexpected mutation {method}") self.mock_api.side_effect = _api res = gitea_cleanup_merged_pr_branch( pr_number=487, confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", branch=branch, remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) self.assertFalse(res.get("performed")) self.assertFalse(res.get("success")) self.assertEqual(res.get("blocker_kind"), "active_branch_ownership") self.assertIn( guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, (res.get("ownership") or {}).get("blocking_categories") or [], ) delete_calls = [ c for c in self.mock_api.call_args_list if c.args and c.args[0] == "DELETE" ] self.assertFalse(delete_calls) def test_open_pr_guard_still_blocks(self): branch = "feat/open-pr-head" patch( "mcp_server._collect_branch_ownership_records", return_value=[], ).start() patch( "mcp_server.api_get_all", return_value=[{"head": {"ref": branch}, "number": 999}], ).start() def _api(method, url, *a, **k): if method == "GET" and "/pulls/" in url: return self._pr_payload(branch, number=487) if method == "GET" and "/branches/" in url: return {"name": branch} raise AssertionError(method) self.mock_api.side_effect = _api res = gitea_cleanup_merged_pr_branch( pr_number=487, confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", branch=branch, remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) self.assertFalse(res.get("performed")) self.assertTrue(any("open PR" in r for r in (res.get("reasons") or []))) def test_non_ancestor_guard_still_blocks(self): branch = "feat/not-ancestor" patch( "mcp_server._collect_branch_ownership_records", return_value=[], ).start() patch( "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref", return_value=False, ).start() def _api(method, url, *a, **k): if method == "GET" and "/pulls/" in url: return self._pr_payload(branch) if method == "GET" and "/branches/" in url: return {"name": branch} raise AssertionError(method) self.mock_api.side_effect = _api res = gitea_cleanup_merged_pr_branch( pr_number=487, confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", branch=branch, remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) self.assertFalse(res.get("performed")) self.assertTrue(any("ancestor" in r for r in (res.get("reasons") or []))) def test_protected_default_branch_guard(self): branch = "master" patch( "mcp_server._collect_branch_ownership_records", return_value=[], ).start() def _api(method, url, *a, **k): if method == "GET" and "/pulls/" in url: return self._pr_payload(branch) if method == "GET" and "/branches/" in url: return {"name": branch} raise AssertionError(method) self.mock_api.side_effect = _api res = gitea_cleanup_merged_pr_branch( pr_number=487, confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", branch=branch, remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) self.assertFalse(res.get("performed")) self.assertTrue(any("protected" in r for r in (res.get("reasons") or []))) if __name__ == "__main__": unittest.main()