Gitea-Tools/tests/test_branch_cleanup_guard.py

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  # noqa: E402

FAKE_AUTH = "token fake"


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()

    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_root_checkout_cleanup_fails_closed(self):
        patch(
            "mcp_server.get_profile",
            return_value={
                "profile_name": "branch-cleanup",
                "allowed_operations": ["gitea.read", "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/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={
                "profile_name": "branch-cleanup",
                "allowed_operations": ["gitea.read", "gitea.branch.delete"],
                "forbidden_operations": [],
            },
        ).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=f"CLEANUP MERGED PR 487 BRANCH {branch}",
            branch=branch,
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertTrue(res["performed"])
        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={
                "profile_name": "branch-cleanup",
                "allowed_operations": ["gitea.read", "gitea.branch.delete"],
                "forbidden_operations": [],
            },
        ).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)


if __name__ == "__main__":
    unittest.main()