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={"records": [], "inventory_error": False}, ).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 == "GET" and url.rstrip("/").endswith("/Example-Repo"): # Repo reachability probe after branch 404 (R1 branch-scoped). return {"full_name": "Example-Org/Example-Repo"} 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["delete_acknowledged"]) self.assertTrue(res["verified_absent"]) 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_branch_scoped_not_found_is_verified_success(self): readback = guard.classify_branch_readback_http_status( 404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH ) result = guard.assess_post_delete_readback(readback) self.assertTrue(result["ok"]) self.assertTrue(result["verified_absent"]) self.assertTrue(result["readback"]["verified_absent"]) def test_generic_404_not_verified_absent(self): # R1: bare 404 must not verify absence for scope in (None, guard.NOT_FOUND_SCOPE_UNKNOWN, guard.NOT_FOUND_SCOPE_REPOSITORY, guard.NOT_FOUND_SCOPE_HOST): with self.subTest(scope=scope): readback = guard.classify_branch_readback_http_status( 404, not_found_scope=scope ) self.assertFalse(readback["verified_absent"]) result = guard.assess_post_delete_readback(readback) self.assertFalse(result["ok"]) self.assertFalse(result["verified_absent"]) def test_exception_substring_404_not_verified(self): # R1: substring/generic 404 without scope stays unverified result = guard.classify_branch_readback_exception( RuntimeError("HTTP 404: something not found") ) self.assertFalse(result["verified_absent"]) self.assertNotEqual(result.get("not_found_scope"), guard.NOT_FOUND_SCOPE_BRANCH) 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", "host": "gitea.prgs.cc", "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", host="gitea.prgs.cc", 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", host="gitea.prgs.cc", 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", host="gitea.prgs.cc", 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", host="gitea.prgs.cc", 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", host="gitea.prgs.cc", records=[self._base(status="stale", reclaim_allowed=False)], ) self.assertTrue(result["block"]) self.assertTrue( any( token in " ".join(result["reasons"]) for token in ("sticky", "reclaim not proven", "fail closed") ) ) 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"), self._base(host="gitea.other.host", status="active"), ] result = guard.assess_active_branch_ownership( remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", host="gitea.prgs.cc", records=records, ) self.assertFalse(result["block"]) self.assertEqual(len(result["ignored_out_of_scope"]), 4) def test_normalized_host_matching(self): # Host identity is normalized (scheme/path stripped) records = [self._base(host="https://gitea.prgs.cc/")] result = guard.assess_active_branch_ownership( remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", host="gitea.prgs.cc", records=records, ) self.assertTrue(result["block"]) 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", host="gitea.prgs.cc", 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={"records": [], "inventory_error": 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} # 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.assertFalse(res.get("verified_absent")) 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={"records": [], "inventory_error": False}, ).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={"records": [], "inventory_error": False}, ).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={ "records": [ { "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, "status": "active", "remote": "prgs", "host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo", "branch": branch, "reclaim_allowed": False, } ], "inventory_error": 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={"records": [], "inventory_error": False}, ).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={"records": [], "inventory_error": False}, ).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={"records": [], "inventory_error": 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("protected" in r for r in (res.get("reasons") or []))) class TestSecondRemediationR1R2(unittest.TestCase): """R1/R2 second-remediation: branch-scoped 404 and top-level fields.""" def test_cleanup_envelope_always_has_top_level_fields(self): env = guard.cleanup_result_envelope( success=False, performed=False, delete_acknowledged=False, verified_absent=False, reasons=["x"], ) for key in ("success", "performed", "delete_acknowledged", "verified_absent"): self.assertIn(key, env) self.assertIsInstance(env[key], bool) def test_repo_scoped_404_never_verified(self): rb = guard.classify_branch_readback_http_status( 404, not_found_scope=guard.NOT_FOUND_SCOPE_REPOSITORY ) self.assertFalse(rb["verified_absent"]) assessed = guard.assess_post_delete_readback(rb) self.assertFalse(assessed["ok"]) self.assertFalse(assessed["verified_absent"]) def test_wrong_host_404_never_verified(self): rb = guard.classify_branch_readback_http_status( 404, not_found_scope=guard.NOT_FOUND_SCOPE_HOST ) self.assertFalse(rb["verified_absent"]) class TestSecondRemediationOwnership(unittest.TestCase): """O1/O2/O3 ownership second-remediation.""" def test_inventory_error_category_blocks(self): result = guard.assess_active_branch_ownership( remote="prgs", org="Org", repo="Repo", branch="feat/x", host="gitea.example.com", records=[ { "category": guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR, "status": "unknown", "remote": "prgs", "host": "gitea.example.com", "org": "Org", "repo": "Repo", "branch": "feat/x", "reclaim_allowed": False, } ], ) self.assertTrue(result["block"]) self.assertIn( guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR, result["blocking_categories"], ) def test_expired_without_explicit_reclaim_blocks(self): # O2: expired must not auto-allow result = guard.assess_active_branch_ownership( remote="prgs", org="Org", repo="Repo", branch="feat/x", host="gitea.example.com", records=[ { "category": guard.OWNERSHIP_CATEGORY_MERGER_LEASE, "status": "expired", "remote": "prgs", "host": "gitea.example.com", "org": "Org", "repo": "Repo", "branch": "feat/x", # reclaim_allowed omitted / False "reclaim_allowed": False, } ], ) self.assertTrue(result["block"]) def test_expired_with_explicit_reclaim_allowed_does_not_block(self): result = guard.assess_active_branch_ownership( remote="prgs", org="Org", repo="Repo", branch="feat/x", host="gitea.example.com", records=[ { "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, "status": "expired", "remote": "prgs", "host": "gitea.example.com", "org": "Org", "repo": "Repo", "branch": "feat/x", "reclaim_allowed": True, } ], ) self.assertFalse(result["block"]) def test_active_reviewer_comment_lease_blocks(self): result = guard.assess_active_branch_ownership( remote="prgs", org="Org", repo="Repo", branch="feat/x", host="gitea.example.com", records=[ { "category": guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, "status": "active", "remote": "prgs", "host": "gitea.example.com", "org": "Org", "repo": "Repo", "branch": "feat/x", "reclaim_allowed": False, "role": "reviewer", } ], ) self.assertTrue(result["block"]) self.assertIn( guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, result["blocking_categories"], ) class TestSecondRemediationIntegration(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(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_r1_repo_404_after_delete_not_verified(self): """After DELETE, branch 404 + repo 404 must not verify absence.""" branch = "feat/r1-repo-404" patch( "mcp_server._collect_branch_ownership_records", return_value={"records": [], "inventory_error": False}, ).start() state = {"branch_gets": 0} def _api(method, url, *a, **k): if method == "GET" and "/pulls/" in url: return self._pr(branch) if method == "GET" and "/branches/" in url: state["branch_gets"] += 1 if state["branch_gets"] == 1: return {"name": branch} raise RuntimeError("HTTP 404: not found") if method == "GET" and url.rstrip("/").endswith("/Example-Repo"): raise RuntimeError("HTTP 404: repository not found") if method == "DELETE": return {} raise AssertionError(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["performed"]) self.assertTrue(res["delete_acknowledged"]) self.assertFalse(res["success"]) self.assertFalse(res["verified_absent"]) def test_o1_inventory_error_blocks_before_delete(self): branch = "feat/o1-inv" patch( "mcp_server._collect_branch_ownership_records", return_value={"records": [], "inventory_error": True}, ).start() def _api(method, url, *a, **k): if method == "GET" and "/pulls/" in url: return self._pr(branch) if method == "GET" and "/branches/" in url: return {"name": branch} raise AssertionError(f"no mutation expected {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["performed"]) self.assertFalse(res["delete_acknowledged"]) self.assertFalse(res["verified_absent"]) self.assertEqual(res.get("blocker_kind"), "active_branch_ownership") def test_o3_comment_reviewer_lease_in_collector(self): """Collector includes active comment-backed reviewer leases.""" active_lease = { "pr_number": 10, "phase": "claimed", "session_id": "s1", "expires_at": "2099-01-02T00:00:00Z", } with patch( "mcp_server.api_get_all", return_value=[{"id": 1, "body": "x"}] ), patch( "mcp_server.reviewer_pr_lease.find_active_reviewer_lease", return_value=active_lease, ), patch( "mcp_server.issue_lock_store.iter_lock_files", return_value=[] ), patch( "mcp_server.worktree_cleanup_audit.list_worktrees", return_value=[] ), patch.object( mcp_server.control_plane_db, "ControlPlaneDB", side_effect=RuntimeError("no cp"), ): bundle = mcp_server._collect_branch_ownership_records( remote="prgs", host="gitea.example.com", org="Example-Org", repo="Example-Repo", branch="feat/x", pr_number=10, project_root="/tmp/repo", auth=FAKE_AUTH, base_api=( "https://gitea.example.com/api/v1/repos/" "Example-Org/Example-Repo" ), ) cats = {r.get("category") for r in bundle.get("records") or []} self.assertIn(guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, cats) def test_o4_reconcile_merged_cleanups_runs_ownership_and_readback(self): from mcp_server import gitea_reconcile_merged_cleanups branch = "feat/reconcile-o4" ownership_calls = [] def fake_collect(**kwargs): ownership_calls.append(kwargs) return {"records": [], "inventory_error": False} probe_calls = [] def fake_probe(h, o, r, auth, br): probe_calls.append(br) return guard.classify_branch_readback_http_status( 404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH ) report = { "entries": [ { "pr_number": 1, "head_branch": branch, "remote_branch": {"safe_to_delete_remote": True}, "local_worktree": {"safe_to_remove_worktree": False}, } ], "reviewer_scratch_entries": [], } patch( "mcp_server.get_profile", return_value={ "profile_name": "prgs-reconciler", "role": "reconciler", "allowed_operations": [ "gitea.read", "gitea.branch.delete", "gitea.pr.close", ], "forbidden_operations": [], }, ).start() patch("mcp_server.api_get_all", return_value=[]).start() patch( "mcp_server.merged_cleanup_reconcile.build_reconciliation_report", return_value=report, ).start() patch( "mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees", return_value=[], ).start() patch( "mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed", return_value=(True, []), ).start() patch("mcp_server.verify_preflight_purity", return_value=None).start() patch( "mcp_server._collect_branch_ownership_records", side_effect=fake_collect, ).start() patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start() self.mock_api.side_effect = lambda *a, **k: {} res = gitea_reconcile_merged_cleanups( dry_run=False, execute_confirmed=True, remote="prgs", ) self.assertTrue(res.get("performed") or res.get("executed")) self.assertTrue(ownership_calls, "ownership must run before delete") self.assertTrue(probe_calls, "post-delete readback must run") actions = res.get("actions") or [] delete_actions = [ a for a in actions if a.get("action") == "delete_remote_branch" ] self.assertEqual(len(delete_actions), 1) self.assertIn("verified_absent", delete_actions[0]) self.assertIn("delete_acknowledged", delete_actions[0]) self.assertTrue(delete_actions[0].get("verified_absent")) if __name__ == "__main__": unittest.main()