refactor: Extract git-clone functionality out of InitTemplates class

mypy.ini

@@ -59,6 +59,6 @@ ignore_missing_imports=True
 ignore_missing_imports=True
 
 # progressive add typechecks and these modules already complete the process, let's keep them clean
-[mypy-samcli.commands.build,samcli.lib.build.*,samcli.commands.local.cli_common.invoke_context,samcli.commands.local.lib.local_lambda,samcli.lib.providers.*]
+[mypy-samcli.commands.build,samcli.lib.build.*,samcli.commands.local.cli_common.invoke_context,samcli.commands.local.lib.local_lambda,samcli.lib.providers.*,samcli.lib.utils.git_repo.py]
 disallow_untyped_defs=True
 disallow_incomplete_defs=True

samcli/commands/init/__init__.py

@@ -256,7 +256,6 @@ def do_cli(
     app_template,
     no_input,
     extra_context,
-    auto_clone=True,
 ):
     """
     Implementation of the ``cli`` method
@@ -274,7 +273,7 @@ def do_cli(
     image_bool = name and pt_explicit and base_image
     if location or zip_bool or image_bool:
         # need to turn app_template into a location before we generate
-        templates = InitTemplates(no_interactive, auto_clone)
+        templates = InitTemplates(no_interactive)
         if package_type == IMAGE and image_bool:
             base_image, runtime = _get_runtime_from_image(base_image)
             options = templates.init_options(package_type, runtime, base_image, dependency_manager)

samcli/commands/init/init_templates.py

@@ -4,40 +4,32 @@
 
 import itertools
 import json
-import os
 import logging
-import platform
-import shutil
-import subprocess
-
+import os
 from pathlib import Path
 from typing import Dict
 
 import click
 
 from samcli.cli.main import global_cfg
 from samcli.commands.exceptions import UserException, AppTemplateUpdateException
-from samcli.lib.utils import osutils
-from samcli.lib.utils.osutils import rmtree_callback
-from samcli.local.common.runtime_template import RUNTIME_DEP_TEMPLATE_MAPPING, get_local_lambda_images_location
+from samcli.lib.utils.git_repo import GitRepo, CloneRepoException, CloneRepoUnstableStateException
 from samcli.lib.utils.packagetype import IMAGE
+from samcli.local.common.runtime_template import RUNTIME_DEP_TEMPLATE_MAPPING, get_local_lambda_images_location
 
 LOG = logging.getLogger(__name__)
+APP_TEMPLATES_REPO_URL = "https://github.com/aws/aws-sam-cli-app-templates"
+APP_TEMPLATES_REPO_NAME = "aws-sam-cli-app-templates"
 
 
 class InvalidInitTemplateError(UserException):
     pass
 
 
 class InitTemplates:
-    def __init__(self, no_interactive=False, auto_clone=True):
-        self._repo_url = "https://github.com/aws/aws-sam-cli-app-templates"
-        self._repo_name = "aws-sam-cli-app-templates"
-        self._temp_repo_name = "TEMP-aws-sam-cli-app-templates"
-        self.repo_path = None
-        self.clone_attempted = False
+    def __init__(self, no_interactive=False):
         self._no_interactive = no_interactive
-        self._auto_clone = auto_clone
+        self._git_repo: GitRepo = GitRepo(url=APP_TEMPLATES_REPO_URL)
 
     def prompt_for_location(self, package_type, runtime, base_image, dependency_manager):
         """
@@ -89,7 +81,7 @@ def prompt_for_location(self, package_type, runtime, base_image, dependency_mana
         if template_md.get("init_location") is not None:
             return (template_md["init_location"], template_md["appTemplate"])
         if template_md.get("directory") is not None:
-            return (os.path.join(self.repo_path, template_md["directory"]), template_md["appTemplate"])
+            return os.path.join(self._git_repo.local_path, template_md["directory"]), template_md["appTemplate"]
         raise InvalidInitTemplateError("Invalid template. This should not be possible, please raise an issue.")
 
     def location_from_app_template(self, package_type, runtime, base_image, dependency_manager, app_template):
@@ -99,7 +91,7 @@ def location_from_app_template(self, package_type, runtime, base_image, dependen
             if template.get("init_location") is not None:
                 return template["init_location"]
             if template.get("directory") is not None:
-                return os.path.join(self.repo_path, template["directory"])
+                return os.path.join(self._git_repo.local_path, template["directory"])
             raise InvalidInitTemplateError("Invalid template. This should not be possible, please raise an issue.")
         except StopIteration as ex:
             msg = "Can't find application template " + app_template + " - check valid values in interactive init."
@@ -112,14 +104,23 @@ def _check_app_template(entry: Dict, app_template: str) -> bool:
         return bool(entry["appTemplate"] == app_template)
 
     def init_options(self, package_type, runtime, base_image, dependency_manager):
-        if not self.clone_attempted:
-            self._clone_repo()
-        if self.repo_path is None:
+        if not self._git_repo.clone_attempted:
+            shared_dir: Path = global_cfg.config_dir
+            try:
+                self._git_repo.clone(clone_dir=shared_dir, clone_name=APP_TEMPLATES_REPO_NAME, replace_existing=True)
+            except CloneRepoUnstableStateException as ex:
+                raise AppTemplateUpdateException(str(ex)) from ex
+            except (OSError, CloneRepoException):
+                # If can't clone, try using an old clone from a previous run if already exist
+                expected_previous_clone_local_path: Path = shared_dir.joinpath(APP_TEMPLATES_REPO_NAME)
+                if expected_previous_clone_local_path.exists():
+                    self._git_repo.local_path = expected_previous_clone_local_path
+        if self._git_repo.local_path is None:
             return self._init_options_from_bundle(package_type, runtime, dependency_manager)
         return self._init_options_from_manifest(package_type, runtime, base_image, dependency_manager)
 
     def _init_options_from_manifest(self, package_type, runtime, base_image, dependency_manager):
-        manifest_path = os.path.join(self.repo_path, "manifest.json")
+        manifest_path = os.path.join(self._git_repo.local_path, "manifest.json")
         with open(str(manifest_path)) as fp:
             body = fp.read()
             manifest_body = json.loads(body)
@@ -154,109 +155,6 @@ def _init_options_from_bundle(package_type, runtime, dependency_manager):
         )
         raise InvalidInitTemplateError(msg)
 
-    @staticmethod
-    def _shared_dir_check(shared_dir: Path) -> bool:
-        try:
-            shared_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
-            return True
-        except OSError as ex:
-            LOG.warning("WARN: Unable to create shared directory.", exc_info=ex)
-            return False
-
-    def _clone_repo(self):
-        if not self._auto_clone:
-            return  # Unit test escape hatch
-        # check if we have templates stored already
-        shared_dir = global_cfg.config_dir
-        if not self._shared_dir_check(shared_dir):
-            # Nothing we can do if we can't access the shared config directory, use bundled.
-            return
-        expected_path = os.path.normpath(os.path.join(shared_dir, self._repo_name))
-        if self._template_directory_exists(expected_path):
-            self._overwrite_existing_templates(expected_path)
-        else:
-            # simply create the app templates repo
-            self._clone_new_app_templates(shared_dir, expected_path)
-        self.clone_attempted = True
-
-    def _overwrite_existing_templates(self, expected_path: str):
-        self.repo_path = expected_path
-        # workflow to clone a copy to a new directory and overwrite
-        with osutils.mkdir_temp(ignore_errors=True) as tempdir:
-            try:
-                expected_temp_path = os.path.normpath(os.path.join(tempdir, self._repo_name))
-                LOG.info("\nCloning app templates from %s", self._repo_url)
-                subprocess.check_output(
-                    [self._git_executable(), "clone", self._repo_url, self._repo_name],
-                    cwd=tempdir,
-                    stderr=subprocess.STDOUT,
-                )
-                # Now we need to delete the old repo and move this one.
-                self._replace_app_templates(expected_temp_path, expected_path)
-                self.repo_path = expected_path
-            except OSError as ex:
-                LOG.warning("WARN: Could not clone app template repo.", exc_info=ex)
-            except subprocess.CalledProcessError as clone_error:
-                output = clone_error.output.decode("utf-8")
-                if "not found" in output.lower():
-                    click.echo("WARN: Could not clone app template repo.")
-
-    @staticmethod
-    def _replace_app_templates(temp_path: str, dest_path: str) -> None:
-        try:
-            LOG.debug("Removing old templates from %s", dest_path)
-            shutil.rmtree(dest_path, onerror=rmtree_callback)
-            LOG.debug("Copying templates from %s to %s", temp_path, dest_path)
-            shutil.copytree(temp_path, dest_path, ignore=shutil.ignore_patterns("*.git"))
-        except (OSError, shutil.Error) as ex:
-            # UNSTABLE STATE
-            # it's difficult to see how this scenario could happen except weird permissions, user will need to debug
-            raise AppTemplateUpdateException(
-                "Unstable state when updating app templates. "
-                "Check that you have permissions to create/delete files in the AWS SAM shared directory "
-                "or file an issue at https://github.com/awslabs/aws-sam-cli/issues"
-            ) from ex
-
-    def _clone_new_app_templates(self, shared_dir, expected_path):
-        with osutils.mkdir_temp(ignore_errors=True) as tempdir:
-            expected_temp_path = os.path.normpath(os.path.join(tempdir, self._repo_name))
-            try:
-                LOG.info("\nCloning app templates from %s", self._repo_url)
-                subprocess.check_output(
-                    [self._git_executable(), "clone", self._repo_url],
-                    cwd=tempdir,
-                    stderr=subprocess.STDOUT,
-                )
-                shutil.copytree(expected_temp_path, expected_path, ignore=shutil.ignore_patterns("*.git"))
-                self.repo_path = expected_path
-            except OSError as ex:
-                LOG.warning("WARN: Can't clone app repo, git executable not found", exc_info=ex)
-            except subprocess.CalledProcessError as clone_error:
-                output = clone_error.output.decode("utf-8")
-                if "not found" in output.lower():
-                    click.echo("WARN: Could not clone app template repo.")
-
-    @staticmethod
-    def _template_directory_exists(expected_path: str) -> bool:
-        path = Path(expected_path)
-        return path.exists()
-
-    @staticmethod
-    def _git_executable() -> str:
-        execname = "git"
-        if platform.system().lower() == "windows":
-            options = [execname, "{}.cmd".format(execname), "{}.exe".format(execname), "{}.bat".format(execname)]
-        else:
-            options = [execname]
-        for name in options:
-            try:
-                subprocess.Popen([name], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-                # No exception. Let's pick this
-                return name
-            except OSError as ex:
-                LOG.debug("Unable to find executable %s", name, exc_info=ex)
-        raise OSError("Cannot find git, was looking at executables: {}".format(options))
-
     def is_dynamic_schemas_template(self, package_type, app_template, runtime, base_image, dependency_manager):
         """
         Check if provided template is dynamic template e.g: AWS Schemas template.

samcli/lib/utils/git_repo.py

@@ -0,0 +1,160 @@
+""" Manage Git repo """
+
+import logging
+import os
+import platform
+import shutil
+import subprocess
+from pathlib import Path
+from typing import Optional
+
+from samcli.lib.utils import osutils
+from samcli.lib.utils.osutils import rmtree_callback
+
+LOG = logging.getLogger(__name__)
+
+
+class CloneRepoException(Exception):
+    """
+    Exception class when clone repo fails.
+    """
+
+
+class CloneRepoUnstableStateException(CloneRepoException):
+    """
+    Exception class when clone repo enters an unstable state.
+    """
+
+
+class GitRepo:
+    """
+    Class for managing a Git repo, currently it has a clone functionality only
+
+    Attributes
+    ----------
+    url: str
+        The URL of this Git repository, example "https://github.com/aws/aws-sam-cli"
+    local_path: Path
+        The path of the last local clone of this Git repository. Can be used in conjunction with clone_attempted
+        to avoid unnecessary multiple cloning of the repository.
+    clone_attempted: bool
+        whether an attempt to clone this Git repository took place or not. Can be used in conjunction with local_path
+        to avoid unnecessary multiple cloning of the repository
+
+    Methods
+    -------
+    clone(self, clone_dir: Path, clone_name, replace_existing=False) -> Path:
+        creates a local clone of this Git repository. (more details in the method documentation).
+    """
+
+    def __init__(self, url: str) -> None:
+        self.url: str = url
+        self.local_path: Optional[Path] = None
+        self.clone_attempted: bool = False
+
+    @staticmethod
+    def _ensure_clone_directory_exists(clone_dir: Path) -> None:
+        try:
+            clone_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
+        except OSError as ex:
+            LOG.warning("WARN: Unable to create clone directory.", exc_info=ex)
+            raise
+
+    @staticmethod
+    def _git_executable() -> str:
+        if platform.system().lower() == "windows":
+            executables = ["git", "git.cmd", "git.exe", "git.bat"]
+        else:
+            executables = ["git"]
+
+        for executable in executables:
+            try:
+                subprocess.Popen([executable], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+                # No exception. Let's pick this
+                return executable
+            except OSError as ex:
+                LOG.debug("Unable to find executable %s", executable, exc_info=ex)
+
+        raise OSError("Cannot find git, was looking at executables: {}".format(executables))
+
+    def clone(self, clone_dir: Path, clone_name: str, replace_existing: bool = False) -> Path:
+        """
+        creates a local clone of this Git repository.
+        This method is different from the standard Git clone in the following:
+        1. It accepts the path to clone into as a clone_dir (the parent directory to clone in) and a clone_name (The
+           name of the local folder) instead of accepting the full path (the join of both) in one parameter
+        2. It removes the "*.git" files/directories so the clone is not a GitRepo any more
+        3. It has the option to replace the local folder(destination) if already exists
+
+        Parameters
+        ----------
+        clone_dir: Path
+            The directory to create the local clone inside
+        clone_name: str
+            The dirname of the local clone
+        replace_existing: bool
+            Whether to replace the current local clone directory if already exists or not
+
+        Returns
+        -------
+            The path of the created local clone
+
+        Raises
+        ------
+        OSError:
+            when file management errors like unable to mkdir, copytree, rmtree ...etc
+        CloneRepoException:
+            General errors like for example; if an error occurred while running `git clone`
+            or if the local_clone already exists and replace_existing is not set
+        CloneRepoUnstableStateException:
+            when reaching unstable state, for example with replace_existing flag set, unstable state can happen
+            if removed the current local clone but failed to copy the new one from the temp location to the destination
+        """
+
+        GitRepo._ensure_clone_directory_exists(clone_dir=clone_dir)
+        # clone to temp then move to the destination(repo_local_path)
+        with osutils.mkdir_temp(ignore_errors=True) as tempdir:
+            try:
+                temp_path = os.path.normpath(os.path.join(tempdir, clone_name))
+                git_executable: str = GitRepo._git_executable()
+                LOG.info("\nCloning from %s", self.url)
+                subprocess.check_output(
+                    [git_executable, "clone", self.url, clone_name],
+                    cwd=tempdir,
+                    stderr=subprocess.STDOUT,
+                )
+                self.local_path = self._persist_local_repo(temp_path, clone_dir, clone_name, replace_existing)
+                return self.local_path
+            except OSError as ex:
+                LOG.warning("WARN: Could not clone repo %s", self.url, exc_info=ex)
+                raise
+            except subprocess.CalledProcessError as clone_error:
+                output = clone_error.output.decode("utf-8")
+                if "not found" in output.lower():
+                    LOG.warning("WARN: Could not clone repo %s", self.url, exc_info=clone_error)
+                raise CloneRepoException from clone_error
+            finally:
+                self.clone_attempted = True
+
+    @staticmethod
+    def _persist_local_repo(temp_path: str, dest_dir: Path, dest_name: str, replace_existing: bool) -> Path:
+        dest_path = os.path.normpath(dest_dir.joinpath(dest_name))
+        try:
+            if Path(dest_path).exists():
+                if not replace_existing:
+                    raise CloneRepoException(f"Can not clone to {dest_path}, directory already exist")
+                LOG.debug("Removing old repo at %s", dest_path)
+                shutil.rmtree(dest_path, onerror=rmtree_callback)
+
+            LOG.debug("Copying from %s to %s", temp_path, dest_path)
+            # Todo consider not removing the .git files/directories
+            shutil.copytree(temp_path, dest_path, ignore=shutil.ignore_patterns("*.git"))
+            return Path(dest_path)
+        except (OSError, shutil.Error) as ex:
+            # UNSTABLE STATE
+            # it's difficult to see how this scenario could happen except weird permissions, user will need to debug
+            raise CloneRepoUnstableStateException(
+                "Unstable state when updating repo. "
+                f"Check that you have permissions to create/delete files in {dest_dir} directory "
+                "or file an issue at https://github.com/aws/aws-sam-cli/issues"
+            ) from ex