windmill/integration_tests/test/git_sync_test.py

import os
import shutil
import tempfile
import time
import unittest
import uuid

import git as gitpython

from .wmill_integration_test_utils import WindmillClient, GiteaClient


# Script content template for bun/TypeScript scripts
def ts_script(body: str) -> str:
    return f"export async function main() {{\n  {body}\n}}\n"


def unique_name(prefix: str = "git-sync-test") -> str:
    return f"{prefix}-{uuid.uuid4().hex[:8]}"


class TestGitSync(unittest.TestCase):
    _client: WindmillClient
    _gitea: GiteaClient
    _repos_to_cleanup: list
    _fork_workspaces_to_cleanup: list

    @classmethod
    def setUpClass(cls) -> None:
        print("Running {}".format(cls.__name__))
        cls._client = WindmillClient()
        cls._gitea = GiteaClient()
        cls._gitea.setup_admin()
        cls._repos_to_cleanup = []
        cls._fork_workspaces_to_cleanup = []

    @classmethod
    def tearDownClass(cls) -> None:
        # Disable git sync to avoid interfering with other tests
        try:
            cls._client.configure_git_sync({"repositories": []})
        except Exception as e:
            print(f"Warning: failed to disable git sync: {e}")

        for fork_id in cls._fork_workspaces_to_cleanup:
            try:
                cls._client.delete_workspace(fork_id)
            except Exception as e:
                print(f"Warning: failed to delete fork workspace {fork_id}: {e}")

        for repo_name in cls._repos_to_cleanup:
            cls._gitea.delete_repo(repo_name)

    def setUp(self):
        """Wait for any pending deployment callbacks from previous tests to drain."""
        time.sleep(2)
        # Wait until no new deployment callback jobs appear for 4 seconds
        prev_count = self._client.count_deployment_callback_jobs()
        for _ in range(3):
            time.sleep(2)
            cur_count = self._client.count_deployment_callback_jobs()
            if cur_count == prev_count:
                break
            prev_count = cur_count

    def _create_test_repo(self) -> tuple:
        """Create a Gitea repo and return (repo_name, docker_clone_url)."""
        name = unique_name()
        docker_url = self._gitea.create_repo(name)
        self._repos_to_cleanup.append(name)
        return name, docker_url

    def _setup_git_sync_resource(self, repo_name: str, branch: str = "main") -> str:
        """Create a git_repository resource pointing to the Gitea repo.
        Returns the resource path."""
        resource_path = f"u/admin/git_sync_{repo_name.replace('-', '_')}"
        docker_url = self._gitea.get_docker_clone_url(repo_name)
        self._client.create_resource(
            path=resource_path,
            resource_type="git_repository",
            value={
                "url": docker_url,
                "branch": branch,
                "is_github_app": False,
            },
            update_if_exists=True,
        )
        return resource_path

    def _configure_single_repo_sync(
        self,
        resource_path: str,
        include_type=None,
        include_path=None,
        use_individual_branch=False,
        group_by_folder=False,
        force_branch=None,
    ):
        """Configure git sync with a single repository (auto-managed script)."""
        repo_settings = {
            "git_repo_resource_path": f"$res:{resource_path}",
            "use_individual_branch": use_individual_branch,
            "group_by_folder": group_by_folder,
        }
        if force_branch:
            repo_settings["force_branch"] = force_branch
        if include_type or include_path:
            repo_settings["settings"] = {
                "include_type": include_type or [],
                "include_path": include_path if include_path is not None else ["**"],
            }

        self._client.configure_git_sync({
            "repositories": [repo_settings],
        })

    def _clone_repo(self, repo_name: str, branch: str = None) -> str:
        """Clone the repo to a temp dir and return the path."""
        host_url = self._gitea.get_host_clone_url(repo_name)
        tmp_dir = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, tmp_dir, ignore_errors=True)
        args = {}
        if branch:
            args["branch"] = branch
        gitpython.Repo.clone_from(host_url, tmp_dir, **args)
        return tmp_dir

    def _clone_repo_all_branches(self, repo_name: str) -> str:
        """Clone the repo fetching all branches."""
        host_url = self._gitea.get_host_clone_url(repo_name)
        tmp_dir = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, tmp_dir, ignore_errors=True)
        gitpython.Repo.clone_from(host_url, tmp_dir, no_single_branch=True)
        return tmp_dir

    def _list_repo_files(self, repo_dir: str, branch: str = None) -> list:
        """List all tracked files in the repo (relative paths)."""
        repo = gitpython.Repo(repo_dir)
        if branch:
            commit = repo.refs[branch].commit
        else:
            commit = repo.head.commit
        return [item.path for item in commit.tree.traverse()]

    def _read_file_content(self, repo_dir: str, file_path: str) -> str:
        """Read a file's content from the repo working tree."""
        full_path = os.path.join(repo_dir, file_path)
        with open(full_path, "r") as f:
            return f.read()

    def _get_commit_count(self, repo_dir: str, branch: str = "main") -> int:
        repo = gitpython.Repo(repo_dir)
        return len(list(repo.iter_commits(branch)))

    def _get_last_commit_message(self, repo_dir: str, branch: str = "main") -> str:
        repo = gitpython.Repo(repo_dir)
        return repo.iter_commits(branch).__next__().message

    def _get_branches(self, repo_dir: str) -> list:
        repo = gitpython.Repo(repo_dir)
        return [ref.name for ref in repo.remote().refs]

    def _create_folder(self, folder_name: str):
        """Create a folder in the workspace, ignoring errors if it already exists."""
        try:
            self._client._client.post(
                f"/api/w/{self._client._workspace}/folders/create",
                json={"name": folder_name},
            )
        except Exception:
            pass

    # ──────────────────────────────────────────────────
    # Core happy-path tests
    # ──────────────────────────────────────────────────

    def test_script_deploy_syncs_to_git(self):
        """Deploy a script and verify it appears in the git repo with correct content."""
        repo_name, _ = self._create_test_repo()
        resource_path = self._setup_git_sync_resource(repo_name)
        self._configure_single_repo_sync(
            resource_path,
            include_type=["script"],
        )

        initial_count = self._client.count_deployment_callback_jobs()

        script_path = f"u/admin/{unique_name('sync_test')}"
        self._client.create_script(
            path=script_path,
            content=ts_script("return 42"),
            language="bun",
        )

        self._client.wait_for_sync_jobs(initial_count, min_new=1)
        time.sleep(3)

        repo_dir = self._clone_repo(repo_name)
        files = self._list_repo_files(repo_dir)

        # The script should appear in the repo
        matching = [f for f in files if script_path in f]
        self.assertTrue(
            len(matching) > 0,
            f"Expected script '{script_path}' in repo files: {files}",
        )

        # Verify file content matches what we deployed
        script_file = [f for f in matching if f.endswith(".ts")][0]
        content = self._read_file_content(repo_dir, script_file)
        self.assertIn(
            "return 42",
            content,
            f"Expected 'return 42' in script content: {content}",
        )

    def test_multi_repo_routing(self):
        """Two repos with different path filters receive the correct objects."""
        repo_name_a, _ = self._create_test_repo()
        repo_name_b, _ = self._create_test_repo()
        res_path_a = self._setup_git_sync_resource(repo_name_a)
        res_path_b = self._setup_git_sync_resource(repo_name_b)

        folder_a = unique_name("folder_a")
        folder_b = unique_name("folder_b")

        self._client.configure_git_sync({
            "repositories": [
                {
                    "git_repo_resource_path": f"$res:{res_path_a}",
                    "use_individual_branch": False,
                    "group_by_folder": False,
                    "settings": {
                        "include_type": ["script"],
                        "include_path": [f"f/{folder_a}/**"],
                    },
                },
                {
                    "git_repo_resource_path": f"$res:{res_path_b}",
                    "use_individual_branch": False,
                    "group_by_folder": False,
                    "settings": {
                        "include_type": ["script"],
                        "include_path": [f"f/{folder_b}/**"],
                    },
                },
            ],
        })

        self._create_folder(folder_a)
        self._create_folder(folder_b)

        initial_count = self._client.count_deployment_callback_jobs()

        script_a = f"f/{folder_a}/script_a"
        script_b = f"f/{folder_b}/script_b"

        self._client.create_script(
            path=script_a,
            content=ts_script("return 'a'"),
            language="bun",
        )
        self._client.create_script(
            path=script_b,
            content=ts_script("return 'b'"),
            language="bun",
        )

        # Wait for at least 2 deployment callback jobs
        self._client.wait_for_sync_jobs(initial_count, min_new=2)
        time.sleep(3)

        # Verify repo A has script_a but not script_b
        repo_dir_a = self._clone_repo(repo_name_a)
        files_a = self._list_repo_files(repo_dir_a)
        self.assertTrue(
            any("script_a" in f for f in files_a),
            f"Expected script_a in repo A files: {files_a}",
        )
        self.assertFalse(
            any("script_b" in f for f in files_a),
            f"Did not expect script_b in repo A files: {files_a}",
        )

        # Verify repo B has script_b but not script_a
        repo_dir_b = self._clone_repo(repo_name_b)
        files_b = self._list_repo_files(repo_dir_b)
        self.assertTrue(
            any("script_b" in f for f in files_b),
            f"Expected script_b in repo B files: {files_b}",
        )
        self.assertFalse(
            any("script_a" in f for f in files_b),
            f"Did not expect script_a in repo B files: {files_b}",
        )

    def test_script_update_creates_new_commit_with_updated_content(self):
        """Updating a script should produce a new commit with the new content."""
        repo_name, _ = self._create_test_repo()
        resource_path = self._setup_git_sync_resource(repo_name)
        self._configure_single_repo_sync(resource_path, include_type=["script"])

        script_path = f"u/admin/{unique_name('update_test')}"

        # Create initial script
        initial_count = self._client.count_deployment_callback_jobs()
        self._client.create_script(
            path=script_path,
            content=ts_script("return 1"),
            language="bun",
        )
        self._client.wait_for_sync_jobs(initial_count, min_new=1)
        time.sleep(3)

        repo_dir = self._clone_repo(repo_name)
        initial_commits = self._get_commit_count(repo_dir)

        # Update the script
        update_count = self._client.count_deployment_callback_jobs()
        self._client.update_script(
            path=script_path,
            content=ts_script("return 2"),
            language="bun",
        )
        self._client.wait_for_sync_jobs(update_count, min_new=1)
        time.sleep(3)

        # Re-clone and check commit count increased
        repo_dir2 = self._clone_repo(repo_name)
        new_commits = self._get_commit_count(repo_dir2)
        self.assertGreater(
            new_commits,
            initial_commits,
            f"Expected more commits after update: {new_commits} vs {initial_commits}",
        )

        # Verify file content reflects the update
        files = self._list_repo_files(repo_dir2)
        script_file = [f for f in files if script_path in f and f.endswith(".ts")][0]
        content = self._read_file_content(repo_dir2, script_file)
        self.assertIn(
            "return 2",
            content,
            f"Expected 'return 2' in updated script content: {content}",
        )
        self.assertNotIn(
            "return 1",
            content,
            f"Did not expect 'return 1' in updated script content: {content}",
        )

    def test_deploy_multiple_object_types(self):
        """Deploy a script, flow, and variable and verify all appear in the repo."""
        repo_name, _ = self._create_test_repo()
        resource_path = self._setup_git_sync_resource(repo_name)
        self._configure_single_repo_sync(
            resource_path,
            include_type=["script", "flow", "variable"],
        )

        initial_count = self._client.count_deployment_callback_jobs()

        suffix = unique_name("multi")
        script_path = f"u/admin/{suffix}_script"
        flow_path = f"u/admin/{suffix}_flow"
        var_path = f"u/admin/{suffix}_var"

        self._client.create_script(
            path=script_path,
            content=ts_script("return 'multi'"),
            language="bun",
        )
        self._client.create_flow(
            path=flow_path,
            flow_value_json="""{
                "summary": "test flow",
                "value": {
                    "modules": [{
                        "id": "a",
                        "value": {
                            "type": "rawscript",
                            "content": "export async function main() { return 1 }",
                            "language": "bun",
                            "input_transforms": {},
                            "tag": ""
                        }
                    }]
                },
                "schema": {
                    "$schema": "https://json-schema.org/draft/2020-12/schema",
                    "properties": {},
                    "required": [],
                    "type": "object",
                    "order": []
                }
            }""",
        )
        self._client.create_variable(
            path=var_path,
            value="test_value",
        )

        # Wait for 3 deployment callbacks (one per object)
        self._client.wait_for_sync_jobs(initial_count, min_new=3)
        time.sleep(3)

        repo_dir = self._clone_repo(repo_name)
        files = self._list_repo_files(repo_dir)
        files_str = "\n".join(files)

        self.assertTrue(
            any(suffix + "_script" in f for f in files),
            f"Expected script in repo:\n{files_str}",
        )
        self.assertTrue(
            any(suffix + "_flow" in f for f in files),
            f"Expected flow in repo:\n{files_str}",
        )
        self.assertTrue(
            any(suffix + "_var" in f for f in files),
            f"Expected variable in repo:\n{files_str}",
        )

    # ──────────────────────────────────────────────────
    # Commit message verification
    # ──────────────────────────────────────────────────

    def test_commit_message_format(self):
        """Verify commit messages have the [WM] prefix."""
        repo_name, _ = self._create_test_repo()
        resource_path = self._setup_git_sync_resource(repo_name)
        self._configure_single_repo_sync(resource_path, include_type=["script"])

        initial_count = self._client.count_deployment_callback_jobs()
        script_path = f"u/admin/{unique_name('commit_msg')}"
        self._client.create_script(
            path=script_path,
            content=ts_script("return 'msg'"),
            language="bun",
        )
        self._client.wait_for_sync_jobs(initial_count, min_new=1)
        time.sleep(3)

        repo_dir = self._clone_repo(repo_name)
        commit_msg = self._get_last_commit_message(repo_dir)

        self.assertTrue(
            commit_msg.startswith("[WM]"),
            f"Expected commit message to start with '[WM]', got: {commit_msg!r}",
        )

    # ──────────────────────────────────────────────────
    # Rename handling
    # ──────────────────────────────────────────────────

    def test_rename_removes_old_file(self):
        """Renaming a script should remove the old file and create the new one."""
        repo_name, _ = self._create_test_repo()
        resource_path = self._setup_git_sync_resource(repo_name)
        self._configure_single_repo_sync(resource_path, include_type=["script"])

        old_path = f"u/admin/{unique_name('rename_old')}"

        # Create initial script
        initial_count = self._client.count_deployment_callback_jobs()
        self._client.create_script(
            path=old_path,
            content=ts_script("return 'old'"),
            language="bun",
        )
        self._client.wait_for_sync_jobs(initial_count, min_new=1)
        time.sleep(3)

        # Verify old script exists in repo
        repo_dir = self._clone_repo(repo_name)
        files = self._list_repo_files(repo_dir)
        old_name = old_path.split("/")[-1]
        self.assertTrue(
            any(old_name in f for f in files),
            f"Expected old script '{old_name}' in repo: {files}",
        )

        # Create new script at different path (simulates rename)
        new_path = f"u/admin/{unique_name('rename_new')}"
        rename_count = self._client.count_deployment_callback_jobs()
        self._client.create_script(
            path=new_path,
            content=ts_script("return 'renamed'"),
            language="bun",
        )
        # Also delete the old script
        self._client.delete_script(old_path)
        # Wait for both create and delete deployment callbacks
        self._client.wait_for_sync_jobs(rename_count, min_new=2)
        time.sleep(3)

        # Verify new script exists
        repo_dir2 = self._clone_repo(repo_name)
        files2 = self._list_repo_files(repo_dir2)
        new_name = new_path.split("/")[-1]
        self.assertTrue(
            any(new_name in f for f in files2),
            f"Expected new script '{new_name}' in repo: {files2}",
        )
        self.assertFalse(
            any(old_name in f for f in files2),
            f"Expected old script '{old_name}' to be removed: {files2}",
        )

    # ──────────────────────────────────────────────────
    # Promotion mode (individual branches)
    # ──────────────────────────────────────────────────

    def test_promotion_mode_creates_per_object_branches(self):
        """In promotion mode (use_individual_branch=True), each deploy creates
        a branch named wm_deploy/{workspace}/{path_type}/{path} with the content
        on that branch, not on main."""
        repo_name, _ = self._create_test_repo()
        resource_path = self._setup_git_sync_resource(repo_name)
        self._configure_single_repo_sync(
            resource_path,
            include_type=["script"],
            use_individual_branch=True,
        )

        initial_count = self._client.count_deployment_callback_jobs()
        script_path = f"u/admin/{unique_name('promo')}"
        self._client.create_script(
            path=script_path,
            content=ts_script("return 'promotion'"),
            language="bun",
        )
        self._client.wait_for_sync_jobs(initial_count, min_new=1)
        time.sleep(3)

        # Clone with all branches
        repo_dir = self._clone_repo_all_branches(repo_name)
        branches = self._get_branches(repo_dir)

        # Should have a branch matching wm_deploy pattern
        wm_branches = [b for b in branches if "wm_deploy/" in b]
        self.assertTrue(
            len(wm_branches) > 0,
            f"Expected wm_deploy/ branch, got branches: {branches}",
        )

        # The branch name should contain 'script' (the path_type)
        deploy_branch = wm_branches[0]
        self.assertIn(
            "script",
            deploy_branch,
            f"Expected 'script' in branch name: {deploy_branch}",
        )

        # The script path (with / replaced by __) should appear in the branch name
        script_name = script_path.split("/")[-1]
        self.assertIn(
            script_name,
            deploy_branch.replace("/", "__"),
            f"Expected script name '{script_name}' in branch: {deploy_branch}",
        )

        # Verify main branch does NOT have the script
        main_files = self._list_repo_files(repo_dir, branch="origin/main")
        self.assertFalse(
            any(script_name in f for f in main_files),
            f"Did not expect script on main branch, but found it: {main_files}",
        )

        # Verify the deploy branch HAS the script
        local_branch_name = deploy_branch.replace("origin/", "")
        repo = gitpython.Repo(repo_dir)
        repo.git.checkout(local_branch_name)
        branch_files = self._list_repo_files(repo_dir)
        self.assertTrue(
            any(script_name in f for f in branch_files),
            f"Expected script on deploy branch '{local_branch_name}': {branch_files}",
        )

    def test_promotion_mode_group_by_folder(self):
        """With use_individual_branch=True and group_by_folder=True, the branch name
        uses the folder prefix (first 2 path segments joined by __) instead of the
        full path."""
        repo_name, _ = self._create_test_repo()
        resource_path = self._setup_git_sync_resource(repo_name)
        self._configure_single_repo_sync(
            resource_path,
            include_type=["script"],
            use_individual_branch=True,
            group_by_folder=True,
        )

        folder_name = unique_name("grp")
        self._create_folder(folder_name)

        initial_count = self._client.count_deployment_callback_jobs()
        script_path = f"f/{folder_name}/{unique_name('grp_script')}"
        self._client.create_script(
            path=script_path,
            content=ts_script("return 'grouped'"),
            language="bun",
        )
        self._client.wait_for_sync_jobs(initial_count, min_new=1)
        time.sleep(3)

        repo_dir = self._clone_repo_all_branches(repo_name)
        branches = self._get_branches(repo_dir)

        wm_branches = [b for b in branches if "wm_deploy/" in b]
        self.assertTrue(
            len(wm_branches) > 0,
            f"Expected wm_deploy/ branch with group_by_folder: {branches}",
        )

        # With group_by_folder, the branch should contain the folder prefix
        # format: wm_deploy/{workspace}/f__{folder_name}
        deploy_branch = wm_branches[0]
        expected_folder_part = f"f__{folder_name}"
        self.assertIn(
            expected_folder_part,
            deploy_branch,
            f"Expected folder-grouped branch name containing '{expected_folder_part}', got: {deploy_branch}",
        )

    # ──────────────────────────────────────────────────
    # force_branch with wmill.yaml
    # ──────────────────────────────────────────────────

    def test_force_branch_with_wmill_yaml(self):
        """force_branch passes --branch to wmill sync pull, which selects the
        matching gitBranches config from wmill.yaml. With branch-specific variables
        configured, the variable file should use a branch-specific path."""
        repo_name, _ = self._create_test_repo()
        resource_path = self._setup_git_sync_resource(repo_name)

        # Push a wmill.yaml to the repo that configures branch-specific variables
        wmill_yaml_content = """\
includes:
  - "**"
gitBranches:
  staging:
    specificItems:
      variables:
        - "**"
"""
        self._gitea.create_file(repo_name, "wmill.yaml", wmill_yaml_content)

        self._configure_single_repo_sync(
            resource_path,
            include_type=["variable"],
            force_branch="staging",
        )

        initial_count = self._client.count_deployment_callback_jobs()
        var_path = f"u/admin/{unique_name('env_var')}"
        self._client.create_variable(
            path=var_path,
            value="staging_value",
        )
        self._client.wait_for_sync_jobs(initial_count, min_new=1)
        time.sleep(3)

        repo_dir = self._clone_repo(repo_name)
        files = self._list_repo_files(repo_dir)

        var_name = var_path.split("/")[-1]

        # With force_branch="staging" and specificItems for variables,
        # the variable file should have ".staging." in its name
        staging_files = [f for f in files if var_name in f and ".staging." in f]
        self.assertTrue(
            len(staging_files) > 0,
            f"Expected variable file with '.staging.' in name for branch-specific item, "
            f"got files: {files}",
        )

    # ──────────────────────────────────────────────────
    # Exclude path filtering
    # ──────────────────────────────────────────────────

    def test_exclude_path_filtering(self):
        """Scripts in excluded paths should not be synced to the repo."""
        repo_name, _ = self._create_test_repo()
        resource_path = self._setup_git_sync_resource(repo_name)

        folder_inc = unique_name("inc")
        folder_exc = unique_name("exc")
        self._create_folder(folder_inc)
        self._create_folder(folder_exc)

        self._client.configure_git_sync({
            "repositories": [{
                "git_repo_resource_path": f"$res:{resource_path}",
                "use_individual_branch": False,
                "group_by_folder": False,
                "settings": {
                    "include_type": ["script"],
                    "include_path": ["f/**"],
                    "exclude_path": [f"f/{folder_exc}/**"],
                },
            }],
        })

        initial_count = self._client.count_deployment_callback_jobs()

        script_inc = f"f/{folder_inc}/included_script"
        script_exc = f"f/{folder_exc}/excluded_script"

        self._client.create_script(
            path=script_inc,
            content=ts_script("return 'included'"),
            language="bun",
        )
        self._client.create_script(
            path=script_exc,
            content=ts_script("return 'excluded'"),
            language="bun",
        )

        # Only 1 sync job expected (the excluded one should not trigger)
        self._client.wait_for_sync_jobs(initial_count, min_new=1)
        time.sleep(5)
        # Verify no extra sync jobs arrived for the excluded script
        final_count = self._client.count_deployment_callback_jobs()
        self.assertEqual(
            final_count, initial_count + 1,
            f"Expected exactly 1 new sync job, got {final_count - initial_count}",
        )

        repo_dir = self._clone_repo(repo_name)
        files = self._list_repo_files(repo_dir)

        self.assertTrue(
            any("included_script" in f for f in files),
            f"Expected included_script in repo: {files}",
        )
        self.assertFalse(
            any("excluded_script" in f for f in files),
            f"Did not expect excluded_script in repo: {files}",
        )

    # ──────────────────────────────────────────────────
    # Workspace fork
    # ──────────────────────────────────────────────────

    def test_workspace_fork_creates_branch(self):
        """Forking a workspace with git sync configured should create a
        fork branch in the git repo."""
        repo_name, _ = self._create_test_repo()
        resource_path = self._setup_git_sync_resource(repo_name)

        # Configure git sync on the parent workspace (sync mode, not promotion)
        self._configure_single_repo_sync(
            resource_path,
            include_type=["script"],
        )

        # Deploy a script first so there's content in the repo
        initial_count = self._client.count_deployment_callback_jobs()
        script_path = f"u/admin/{unique_name('fork_base')}"
        self._client.create_script(
            path=script_path,
            content=ts_script("return 'base'"),
            language="bun",
        )
        self._client.wait_for_sync_jobs(initial_count, min_new=1)
        time.sleep(3)

        # Create workspace fork
        fork_id = f"wm-fork-{uuid.uuid4().hex[:8]}"
        fork_name = f"Fork {fork_id}"
        self._fork_workspaces_to_cleanup.append(fork_id)

        # Step 1: Create git branches for the fork
        job_ids = self._client.create_workspace_fork_branch(fork_id, fork_name)
        if job_ids:
            self._client.wait_for_jobs_by_ids(job_ids, timeout=90)
            time.sleep(3)

        # Step 2: Create the fork workspace
        self._client.create_workspace_fork(fork_id, fork_name)

        # Verify a fork branch was created in the git repo
        repo_dir = self._clone_repo_all_branches(repo_name)
        branches = self._get_branches(repo_dir)

        # Fork branches are named: wm-fork/{original_branch}/{fork_id}
        fork_branches = [b for b in branches if "wm-fork" in b]
        self.assertTrue(
            len(fork_branches) > 0,
            f"Expected a wm-fork branch in the repo after forking, got: {branches}",
        )