Fix for clones of private repositories

Author: acabarbaye

continuous_delivery_scripts/utils/git_helpers.py

@@ -10,7 +10,7 @@
 from pathlib import Path
 from typing import Optional, List, Union, Any, Tuple
 
-from git import Repo, Actor
+from git import Repo, Actor, GitCommandError
 from packaging import version
 
 from .configuration import configuration, ConfigurationVariable
@@ -48,7 +48,8 @@ def _git_url_ssh_to_https(self, url: str) -> str:
         path = url.split("github.com", 1)[1][1:].strip()
         new = "https://{GITHUB_TOKEN}:[email protected]/%s" % path
         logger.info("rewriting git url to: %s" % new)
-        return new.format(GITHUB_TOKEN=configuration.get_value(ConfigurationVariable.GIT_TOKEN))
+        return new.format(GITHUB_TOKEN=configuration.get_value(
+            ConfigurationVariable.GIT_TOKEN))
 
     def clone(self, path: Path) -> "GitWrapper":
         """Clones this repository to the path.
@@ -59,9 +60,20 @@ def clone(self, path: Path) -> "GitWrapper":
         Returns:
             a wrapper over the cloned repository
         """
-        git_clone = self.repo.clone_from(
-            url=self.get_remote_url(), to_path=str(path), multi_options=["--recurse-submodules"]
-        )
+        try:
+            git_clone = self.repo.clone_from(
+                url=self.get_remote_url(), to_path=str(path),
+                multi_options=["--recurse-submodules"]
+            )
+        except GitCommandError as e:
+            logger.info("failed cloning repository: %s" % e)
+            logger.info("trying with authentication")
+            git_clone = self.repo.clone_from(
+                url=self._git_url_ssh_to_https(self.get_remote_url()),
+                to_path=str(path),
+                multi_options=["--recurse-submodules"]
+            )
+
         clone = GitWrapper(path=path, repo=git_clone)
         clone.set_remote_url(self.get_remote_url())
         clone.fetch()
@@ -78,8 +90,10 @@ def root(self) -> Path:
 
     def configure_author(self) -> None:
         """Sets the author."""
-        self.repo.config_writer().set_value("user", "name", self.author.name).release()
-        self.repo.config_writer().set_value("user", "email", self.author.email).release()
+        self.repo.config_writer().set_value("user", "name",
+                                            self.author.name).release()
+        self.repo.config_writer().set_value("user", "email",
+                                            self.author.email).release()
 
     def checkout_branch(self, branch_name: str) -> Any:
         """Checks out a branch from its name.
@@ -114,7 +128,8 @@ def _add_one_path(self, path_model: Path) -> None:
         if not path_model.is_absolute():
             path_model = Path(self.root).joinpath(path_model)
         if not path_model.exists():
-            logger.warning(f"[Git] {path_model} cannot be added because not found.")
+            logger.warning(
+                f"[Git] {path_model} cannot be added because not found.")
             return
         relative_path = str(path_model.relative_to(self.root))
         unix_relative_path = relative_path.replace("\\", "/")
@@ -135,7 +150,8 @@ def add(self, path: Union[list, set, str]) -> None:
         else:
             self._add_one_file_or_one_dir(path)
 
-    def commit(self, message: str, **kwargs: Optional[Tuple[str, Any]]) -> None:
+    def commit(self, message: str,
+               **kwargs: Optional[Tuple[str, Any]]) -> None:
         """Commits changes to the repository.
 
         Args:
@@ -151,15 +167,17 @@ def get_master_branch(self) -> Any:
         Returns:
             corresponding branch
         """
-        return self.get_branch(configuration.get_value(ConfigurationVariable.MASTER_BRANCH))
+        return self.get_branch(
+            configuration.get_value(ConfigurationVariable.MASTER_BRANCH))
 
     def get_beta_branch(self) -> Any:
         """Gets the `beta` branch.
 
         Returns:
             corresponding branch
         """
-        return self.get_branch(configuration.get_value(ConfigurationVariable.BETA_BRANCH))
+        return self.get_branch(
+            configuration.get_value(ConfigurationVariable.BETA_BRANCH))
 
     def is_release_branch(self, branch_name: Optional[str]) -> bool:
         """Checks whether the branch is a `release` branch or not.
@@ -170,7 +188,8 @@ def is_release_branch(self, branch_name: Optional[str]) -> bool:
         Returns:
             True if the branch is used for `release` code; False otherwise
         """
-        branch_pattern = configuration.get_value(ConfigurationVariable.RELEASE_BRANCH_PATTERN)
+        branch_pattern = configuration.get_value(
+            ConfigurationVariable.RELEASE_BRANCH_PATTERN)
         if not branch_pattern or not branch_name:
             return False
         is_release: Optional[Any] = re.search(branch_pattern, str(branch_name))
@@ -213,7 +232,8 @@ def get_current_branch(self) -> Any:
         try:
             return self.repo.active_branch
         except TypeError as e:
-            logger.warning(f"Could not determine the branch name using GitPython: {e}")
+            logger.warning(
+                f"Could not determine the branch name using GitPython: {e}")
         current_branch = self._get_branch_from_advanced_feature()
         if not current_branch:
             current_branch = self._get_branch_from_abbreviation("HEAD")
@@ -222,13 +242,17 @@ def get_current_branch(self) -> Any:
     def _get_branch_from_advanced_feature(self) -> Any:
         if version.parse(self.git_version()) >= version.parse("2.22"):
             current_branch = self.repo.git.branch(show_current=True)
-            current_branch = current_branch if isinstance(current_branch, str) else current_branch.decode("utf-8")
+            current_branch = current_branch if isinstance(current_branch,
+                                                          str) else current_branch.decode(
+                "utf-8")
             return self.get_branch(current_branch)
         return None
 
     def _get_branch_from_abbreviation(self, abbreviation: str) -> Any:
         current_branch = self.repo.git.rev_parse("--abbrev-ref", abbreviation)
-        current_branch = current_branch if isinstance(current_branch, str) else current_branch.decode("utf-8")
+        current_branch = current_branch if isinstance(current_branch,
+                                                      str) else current_branch.decode(
+            "utf-8")
         return self.get_branch(current_branch.strip())
 
     def get_commit_count(self) -> int:
@@ -282,7 +306,8 @@ def merge(self, branch: Any) -> None:
         current_branch = self.get_current_branch()
         merge_base = self.repo.merge_base(branch, current_branch)
         self.repo.index.merge_tree(current_branch, base=merge_base)
-        self.commit(f"Merge from {str(branch)}", parent_commits=(branch.commit, current_branch.commit))
+        self.commit(f"Merge from {str(branch)}",
+                    parent_commits=(branch.commit, current_branch.commit))
 
     def get_remote_url(self) -> str:
         """Gets the URL of the remote repository.
@@ -315,7 +340,9 @@ def set_remote_url(self, url: str) -> None:
         remote = self._get_remote()
         if remote:
             self.repo.delete_remote(str(remote))
-        self.repo.create_remote(configuration.get_value(ConfigurationVariable.REMOTE_ALIAS), url=url)
+        self.repo.create_remote(
+            configuration.get_value(ConfigurationVariable.REMOTE_ALIAS),
+            url=url)
 
     def get_remote_branch(self, branch_name: str) -> Optional[Any]:
         """Gets the branch present in the remote repository.
@@ -342,7 +369,8 @@ def set_upstream_branch(self, branch_name: str) -> None:
             branch_name: name of the remote branch.
         """
         if self.remote_branch_exists(branch_name):
-            self.repo.git.branch("--set-upstream-to", self.get_remote_branch(branch_name))
+            self.repo.git.branch("--set-upstream-to",
+                                 self.get_remote_branch(branch_name))
 
     def delete_branch(self, branch: Any) -> None:
         """Deletes a branch.
@@ -384,17 +412,21 @@ def remote_branch_exists(self, branch_name: str) -> bool:
         """
         return self.get_remote_branch(branch_name) is not None
 
-    def _get_specific_changes(self, change_type: Optional[str], commit1: Any, commit2: Any) -> List[str]:
+    def _get_specific_changes(self, change_type: Optional[str], commit1: Any,
+                              commit2: Any) -> List[str]:
         diff = commit1.diff(commit2)
         if change_type:
             change_type = change_type.upper()
             change_type = change_type if change_type in diff.change_type else None
-        diff_iterator = diff.iter_change_type(change_type) if change_type else diff
-        changes = [change.a_path if change.a_path else change.b_path for change in diff_iterator]
+        diff_iterator = diff.iter_change_type(
+            change_type) if change_type else diff
+        changes = [change.a_path if change.a_path else change.b_path for change
+                   in diff_iterator]
         return changes
 
     def get_changes_list(
-        self, commit1: Any, commit2: Any, change_type: Optional[str] = None, dir: Optional[str] = None
+            self, commit1: Any, commit2: Any,
+            change_type: Optional[str] = None, dir: Optional[str] = None
     ) -> List[str]:
         """Gets change list.
 
@@ -414,7 +446,8 @@ def get_changes_list(
         if dir:
             windows_path = dir.replace("/", "\\")
             linux_path = dir.replace("\\", "/")
-            return [change for change in changes if (linux_path in change) or (windows_path in change)]
+            return [change for change in changes if
+                    (linux_path in change) or (windows_path in change)]
         else:
             return changes
 
@@ -425,19 +458,22 @@ def pull_all(self) -> None:
     def pull(self) -> None:
         """Pulls changes on current branch from the remote repository."""
         if self.remote_branch_exists(self.get_current_branch()):
-            self.repo.git.pull(self._get_remote(), self.get_current_branch(), quiet=True)
+            self.repo.git.pull(self._get_remote(), self.get_current_branch(),
+                               quiet=True)
 
     def force_pull(self) -> None:
         """Force pulls changes from the remote repository."""
-        self.repo.git.pull(self._get_remote(), self.get_current_branch(), quiet=True, force=True)
+        self.repo.git.pull(self._get_remote(), self.get_current_branch(),
+                           quiet=True, force=True)
 
     def push(self) -> None:
         """Pushes commits.
 
         Pushes changes to the remote repository.
         Pushes also relevant annotated tags when pushing branches out.
         """
-        self.repo.git.push("--follow-tags", "--set-upstream", self._get_remote(), self.get_current_branch())
+        self.repo.git.push("--follow-tags", "--set-upstream",
+                           self._get_remote(), self.get_current_branch())
 
     def push_tag(self) -> None:
         """Pushes commits and tags.
@@ -500,21 +536,25 @@ def git_version(self) -> str:
         Returns:
             the version of git in use.
         """
-        return ".".join([str(element) for element in self.repo.git.version_info])
+        return ".".join(
+            [str(element) for element in self.repo.git.version_info])
 
     def _get_remote(self) -> Optional[Any]:
         try:
-            return self.repo.remote(configuration.get_value(ConfigurationVariable.REMOTE_ALIAS))
+            return self.repo.remote(
+                configuration.get_value(ConfigurationVariable.REMOTE_ALIAS))
         except (IndexError, ValueError) as e:
             logger.warning(e)
             return None
 
     def list_files_added_on_current_branch(self) -> List[str]:
         """Returns a list of files changed against master branch."""
-        master_branch_commit = self.repo.commit(configuration.get_value(ConfigurationVariable.MASTER_BRANCH))
+        master_branch_commit = self.repo.commit(
+            configuration.get_value(ConfigurationVariable.MASTER_BRANCH))
         current_branch_commit = self.repo.commit(self.get_current_branch())
         changes = self.get_changes_list(
-            self.get_branch_point(master_branch_commit, current_branch_commit), current_branch_commit, change_type="a"
+            self.get_branch_point(master_branch_commit, current_branch_commit),
+            current_branch_commit, change_type="a"
         )
         return changes
 
@@ -537,7 +577,8 @@ def uncommitted_changes(self) -> List[Path]:
         if not status:
             return []
 
-        return [Path(self.root).joinpath(line.strip().split(" ")[-1]) for line in status.splitlines()]
+        return [Path(self.root).joinpath(line.strip().split(" ")[-1]) for line
+                in status.splitlines()]
 
     @staticmethod
     def _apply_modifications(destination: Path, modified_file: Path) -> None:
@@ -573,8 +614,10 @@ class ProjectGitWrapper(GitWrapper):
     def __init__(self) -> None:
         """Creates a Git Wrapper."""
         super().__init__(
-            path=Path(configuration.get_value(ConfigurationVariable.PROJECT_ROOT)),
-            repo=Repo(configuration.get_value(ConfigurationVariable.PROJECT_ROOT)),
+            path=Path(
+                configuration.get_value(ConfigurationVariable.PROJECT_ROOT)),
+            repo=Repo(
+                configuration.get_value(ConfigurationVariable.PROJECT_ROOT)),
         )
 
 
@@ -615,7 +658,8 @@ def __init__(self, path: Path, initial_path: Path, repo: Repo) -> None:
     @staticmethod
     def wrap(repo: GitWrapper, initial_location: Path) -> "GitClone":
         """Wraps around around a repository."""
-        return GitClone(repo=repo.repo, path=repo.root, initial_path=initial_location)
+        return GitClone(repo=repo.repo, path=repo.root,
+                        initial_path=initial_location)
 
     @property
     def initial_location(self) -> Path:
@@ -638,26 +682,31 @@ def get_corresponding_path(self, path_in_initial_repo: Path) -> Path:
             return Path(self.root).joinpath(path_in_initial_repo)
         try:
             # Tyring to find if the path corresponds to a file/directory present in intial repository
-            return Path(self.root).joinpath(path_in_initial_repo.relative_to(self.initial_location))
+            return Path(self.root).joinpath(
+                path_in_initial_repo.relative_to(self.initial_location))
         except ValueError:
             return path_in_initial_repo
 
 
 class GitTempClone:
     """Context manager providing a temporary cloned repository."""
 
-    def __init__(self, desired_branch_name: Optional[str], repository_to_clone: GitWrapper):
+    def __init__(self, desired_branch_name: Optional[str],
+                 repository_to_clone: GitWrapper):
         """Constructor.
 
         Args:
             desired_branch_name: the branch to consider. I
             repository_to_clone: the repository to clone. If not specified, the project repository will be used.
         """
         self._temporary_dir = TemporaryDirectory()
-        logger.info(f"Creating a temporary repository in {self._temporary_dir}")
+        logger.info(
+            f"Creating a temporary repository in {self._temporary_dir}")
         self._repo = repository_to_clone
-        _current_branch_name = desired_branch_name if desired_branch_name else str(self._repo.get_current_branch())
-        self._clone = GitClone.wrap(self._repo.clone(self._temporary_dir.path), initial_location=self._repo.root)
+        _current_branch_name = desired_branch_name if desired_branch_name else str(
+            self._repo.get_current_branch())
+        self._clone = GitClone.wrap(self._repo.clone(self._temporary_dir.path),
+                                    initial_location=self._repo.root)
         self._clone.checkout(_current_branch_name)
         self._repo.apply_uncommitted_changes(self._clone)
 
@@ -688,4 +737,5 @@ def __init__(self, desired_branch_name: Optional[str] = None):
             system will try to identify the current branch in the repository which
             will work in most cases but probably not on CI.
         """
-        super().__init__(desired_branch_name=desired_branch_name, repository_to_clone=ProjectGitWrapper())
+        super().__init__(desired_branch_name=desired_branch_name,
+                         repository_to_clone=ProjectGitWrapper())

news/.keep

@@ -0,0 +1 @@
+Fix for clones of private repository