[ISV-5276] Create helper functions for GitHub branches for copy and delete operations.

operator-pipeline-images/operatorcert/github.py

@@ -8,7 +8,14 @@
 from typing import Any, Dict, List, Optional
 
 import requests
-from github import Github, Label, PaginatedList, PullRequest
+from github import (
+    Github,
+    Label,
+    PaginatedList,
+    PullRequest,
+    GithubException,
+    InputGitTreeElement,
+)
 from operatorcert.utils import add_session_retries
 
 LOGGER = logging.getLogger("operator-cert")
@@ -336,3 +343,125 @@ def close_pull_request(
     """
     pull_request.edit(state="closed")
     return pull_request
+
+
+def copy_branch(
+    github_client: Github,
+    src_repo_name: str,
+    src_branch_name: str,
+    dest_repo_name: str,
+    dest_branch_name: str,
+) -> None:
+    """
+    Copy a branch from Source Repository to Destination Repository.
+
+    Limitations:
+    - This method does not handle symbolic links or non-regular files.
+    - Binary files that cannot be decoded as UTF-8 text will not be copied correctly.
+    - A proper fix would require cloning the source
+      repository locally and pushing it to the destination.
+    - While this approach has limitations, it may be sufficient for short-term needs,
+      and improvements can be made in the future.
+
+    Args:
+        github_client (Github): A Github API client
+        src_repo_name (str): The source repository name in the format "organization/repository"
+        src_branch_name(str): The name of the branch to copy
+        dest_repo_name(str): The destination repository name in the format "organization/repository"
+        dest_branch_name(str): The name of the destination branch
+    """
+
+    try:
+        src_repository = github_client.get_repo(src_repo_name)
+        dest_repository = github_client.get_repo(dest_repo_name)
+
+        base_dest_branch = dest_repository.get_branch("main")
+        base_dest_commit_sha = base_dest_branch.commit.sha
+
+        ref = f"refs/heads/{dest_branch_name}"
+        dest_repository.create_git_ref(ref=ref, sha=base_dest_commit_sha)
+
+        tree_elements = []
+
+        def collect_files_recursive(path: str = "") -> None:
+            """
+            Recursively fetch and prepare files from the source repo.
+            Args:
+                path (str): Content file path
+            """
+            contents = src_repository.get_contents(path, ref=src_branch_name)
+
+            if isinstance(contents, list):
+                for content in contents:
+                    collect_files_recursive(content.path)
+            else:
+                file_data = contents.decoded_content
+                element = InputGitTreeElement(
+                    path=contents.path,
+                    mode="100644",
+                    type="blob",
+                    content=file_data.decode("utf-8"),
+                )
+                tree_elements.append(element)
+
+        collect_files_recursive()
+
+        new_tree = dest_repository.create_git_tree(
+            tree_elements, base_dest_branch.commit.commit.tree
+        )
+
+        commit_message = f"Copy from {src_branch_name} into {dest_branch_name}."
+        new_commit = dest_repository.create_git_commit(
+            commit_message, new_tree, [base_dest_branch.commit.commit]
+        )
+
+        ref = f"heads/{dest_branch_name}"
+        dest_repository.get_git_ref(ref).edit(new_commit.sha)
+
+        LOGGER.debug(
+            "Branch '%s' from '%s' copied to '%s' in '%s' successfully.",
+            src_branch_name,
+            src_repo_name,
+            dest_branch_name,
+            dest_repo_name,
+        )
+
+    except GithubException:
+        LOGGER.exception(
+            "Error while copying branch '%s' from '%s' to '%s' in '%s'.",
+            src_branch_name,
+            src_repo_name,
+            dest_branch_name,
+            dest_repo_name,
+        )
+        raise
+
+
+def delete_branch(
+    github_client: Github,
+    repository_name: str,
+    branch_name: str,
+) -> None:
+    """
+    Delete a branch from a Github repository.
+
+    Args:
+        github_client (Github): A Github API client
+        repository_name (str): A repository name in the format "organization/repository"
+        branch_name (str): The name of the branch to delete
+    """
+    try:
+        repository = github_client.get_repo(repository_name)
+        branch_ref = f"heads/{branch_name}"
+
+        repository.get_git_ref(branch_ref).delete()
+
+        LOGGER.debug(
+            "Branch '%s' deleted from '%s' successfully.", branch_name, repository_name
+        )
+
+    except GithubException:
+        LOGGER.exception(
+            "Error while deleting the '%s' from '%s'.", branch_name, repository_name
+        )
+        raise

operator-pipeline-images/tests/test_github.py

@@ -1,7 +1,10 @@
 from typing import Any, List
 from unittest.mock import ANY, MagicMock, call, patch
+import base64
 
 import pytest
+from github import GithubException
+from github.InputGitTreeElement import InputGitTreeElement
 from operatorcert import github
 from requests import HTTPError, Response
 
@@ -229,3 +232,141 @@ def test_close_pull_request() -> None:
     mock_pull_request.edit.assert_called_once_with(state="closed")
 
     assert resp == mock_pull_request
+
+
+def test_copy_branch_success() -> None:
+    mock_client = MagicMock()
+    src_repo = mock_client.get_repo.return_value
+    dest_repo = mock_client.get_repo.return_value
+
+    src_branch_ref = MagicMock()
+    src_branch_ref.commit.sha = "abc123"
+    src_repo.get_branch.return_value = src_branch_ref
+
+    dest_branch_ref = MagicMock()
+    dest_branch_ref.commit.sha = "def456"
+    dest_branch_ref.commit.commit.tree = MagicMock()
+    dest_repo.get_branch.return_value = dest_branch_ref
+
+    dest_repo.create_git_ref.return_value = MagicMock()
+
+    mock_file = MagicMock()
+    mock_file.path = "test.txt"
+    mock_file.type = "file"
+    mock_file.decoded_content = b"new content"
+
+    src_repo.get_contents.side_effect = lambda path, ref: (
+        [mock_file] if path == "" else mock_file
+    )
+
+    dest_repo.create_git_tree.return_value = MagicMock()
+    dest_repo.create_git_commit.return_value = MagicMock(sha="new_commit_sha")
+
+    ref_mock = MagicMock()
+    dest_repo.get_git_ref.return_value = ref_mock
+
+    github.copy_branch(
+        mock_client,
+        "org/source-repo",
+        "main",
+        "org/dest-repo",
+        "copied-branch",
+    )
+
+    dest_repo.create_git_ref.assert_called_once_with(
+        ref="refs/heads/copied-branch", sha="def456"
+    )
+    dest_repo.create_git_tree.assert_called()
+    dest_repo.create_git_commit.assert_called()
+    ref_mock.edit.assert_called_once_with("new_commit_sha")
+
+
+def test_copy_branch_with_nested_files() -> None:
+    mock_client = MagicMock()
+    src_repo = mock_client.get_repo.return_value
+    dest_repo = mock_client.get_repo.return_value
+
+    src_branch_ref = MagicMock()
+    src_branch_ref.commit.sha = "abc123"
+    src_repo.get_branch.return_value = src_branch_ref
+
+    dest_branch_ref = MagicMock()
+    dest_branch_ref.commit.sha = "def456"
+    dest_branch_ref.commit.commit.tree = MagicMock()
+    dest_repo.get_branch.return_value = dest_branch_ref
+
+    dest_repo.create_git_ref.return_value = MagicMock()
+
+    mock_dir = MagicMock()
+    mock_dir.path = "dir1"
+    mock_dir.type = "dir"
+
+    mock_file = MagicMock()
+    mock_file.path = "dir1/test.txt"
+    mock_file.type = "file"
+    mock_file.decoded_content = b"nested content"
+
+    src_repo.get_contents.side_effect = lambda path, ref: (
+        [mock_dir] if path == "" else [mock_file] if path == "dir1" else mock_file
+    )
+
+    dest_repo.create_git_tree.return_value = MagicMock()
+    dest_repo.create_git_commit.return_value = MagicMock(sha="new_commit_sha")
+
+    ref_mock = MagicMock()
+    dest_repo.get_git_ref.return_value = ref_mock
+
+    github.copy_branch(
+        mock_client,
+        "org/source-repo",
+        "main",
+        "org/dest-repo",
+        "copied-branch",
+    )
+
+    dest_repo.create_git_ref.assert_called_once_with(
+        ref="refs/heads/copied-branch", sha="def456"
+    )
+    dest_repo.create_git_tree.assert_called()
+    dest_repo.create_git_commit.assert_called()
+    ref_mock.edit.assert_called_once_with("new_commit_sha")
+
+
+def test_copy_branch_handles_github_exception() -> None:
+    mock_client = MagicMock()
+    mock_client.get_repo.side_effect = GithubException(500, "Internal Server Error", {})
+
+    with pytest.raises(GithubException):
+        github.copy_branch(
+            mock_client,
+            "org/source-repo",
+            "main",
+            "org/dest-repo",
+            "copied-branch",
+        )
+
+
+def test_delete_branch_success() -> None:
+    mock_client = MagicMock()
+    mock_repo = MagicMock()
+    mock_client.get_repo.return_value = mock_repo
+
+    branch_ref = MagicMock()
+    mock_repo.get_git_ref.return_value = branch_ref
+
+    github.delete_branch(mock_client, "org/repo-name", "old-feature-branch")
+
+    mock_repo.get_git_ref.assert_called_once_with("heads/old-feature-branch")
+    branch_ref.delete.assert_called_once()
+
+
+def test_delete_branch_when_branch_does_not_exist() -> None:
+    mock_client = MagicMock()
+    mock_repo = MagicMock()
+    mock_get_git_ref = MagicMock()
+    mock_client.get_repo.return_value = mock_repo
+    mock_repo.get_git_ref.return_value = mock_get_git_ref
+    mock_get_git_ref.delete.side_effect = GithubException(0, "err", None)
+
+    with pytest.raises(GithubException):
+        github.delete_branch(mock_client, "org/repo-name", "non-existent-branch")