SG-39245 Refactor git descriptors

python/tank/descriptor/io_descriptor/git.py

@@ -8,20 +8,21 @@
 # agreement to the Shotgun Pipeline Toolkit Source Code License. All rights
 # not expressly granted therein are reserved by Shotgun Software Inc.
 import os
-import uuid
-import shutil
-import tempfile
+import platform
 import subprocess
 
+from time import time
+
 from .downloadable import IODescriptorDownloadable
 from ... import LogManager
 from ...util.process import subprocess_check_output, SubprocessCalledProcessError
 
-from ..errors import TankError
+from ..errors import TankError, TankDescriptorError
 from ...util import filesystem
 from ...util import is_windows
 
 log = LogManager.get_logger(__name__)
+IS_WINDOWS = platform.system() == "Windows"
 
 
 def _can_hide_terminal():
@@ -34,7 +35,8 @@
         subprocess.STARTF_USESHOWWINDOW
         subprocess.SW_HIDE
         return True
-    except Exception:
+    except AttributeError as e:
+        log.debug("Terminal cant be hidden: %s" % e)
         return False
 
 
@@ -59,7 +61,40 @@
     pass
 
 
-class IODescriptorGit(IODescriptorDownloadable):
+class _IODescriptorGitCache(type):
+    """Use as metaclass. Caches object instances for 2min."""
+
+    _instances = {}
+
+    def __call__(cls, descriptor_dict, sg_connection, bundle_type):
+        now = int(time() / 100)
+        floored_time = now - now % 2  # Cache is valid for 2min
+
+        if (
+            descriptor_dict["type"] == "git_branch"
+        ):  # cant fetch last commit here, too soon
+            version = "-".join(
+                filter(
+                    None, [descriptor_dict.get("version"), descriptor_dict["branch"]]
+                )
+            )
+        else:
+            version = descriptor_dict["version"]
+
+        id_ = "-".join([descriptor_dict["type"], descriptor_dict["path"], version])
+
+        cached_time, self = cls._instances.get(id_, (-1, None))
+        if cached_time < floored_time:
+            log.debug(
+                "{} {} cache expired: cachedTime:{}".format(self, id_, cached_time)
+            )
+            self = super().__call__(descriptor_dict, sg_connection, bundle_type)
+            cls._instances[id_] = (floored_time, self)
+
+        return self
+
+
+class IODescriptorGit(IODescriptorDownloadable, metaclass=_IODescriptorGitCache):
     """
     Base class for git descriptors.
 
@@ -81,11 +116,67 @@
             descriptor_dict, sg_connection, bundle_type
         )
 
-        self._path = descriptor_dict.get("path")
-        # strip trailing slashes - this is so that when we build
-        # the name later (using os.basename) we construct it correctly.
-        if self._path.endswith("/") or self._path.endswith("\\"):
-            self._path = self._path[:-1]
+        _path = self._normalize_path(descriptor_dict.get("path"))
+
+        if self._path_is_local(_path):
+            self._path = self._execute_git_commands(
+                ["git", "-C", _path, "remote", "get-url", "origin"]
+            )
+            log.debug(
+                "Get remote url from local repo: {} -> {}".format(_path, self._path)
+            )
+            filtered_local_data = {
+                k: v
+                for k, v in self._fetch_local_data(_path).items()
+                if k not in descriptor_dict
+            }
+            descriptor_dict.update(filtered_local_data)
+        else:
+            self._path = _path
+
+    def is_git_available(self):
+        log.debug("Checking that git exists and can be executed...")
+
+        if IS_WINDOWS:
+            cmd = "where"
+        else:
+            cmd = "which"
+
+        try:
+            output = _check_output([cmd, "git"])
+        except SubprocessCalledProcessError:
+            raise TankGitError(
+                "Cannot execute the 'git' command. Please make sure that git is "
+                "installed on your system and that the git executable has been added to the PATH."
+            )
+        else:
+            log.debug("Git installed: %s" % output)
+            return True
+
+    def _execute_git_commands(self, commands):
+        # first probe to check that git exists in our PATH
+        self.is_git_available()
+
+        log.debug("Executing command '%s' using subprocess module." % commands)
+
+        # It's important to pass GIT_TERMINAL_PROMPT=0 or the git subprocess will
+        # just hang waiting for credentials to be entered on the missing terminal.
+        # I would have expected Windows to give an error about stdin being close and
+        # aborting the git command but at least on Windows 10 that is not the case.
+        environ = os.environ.copy()
+        environ["GIT_TERMINAL_PROMPT"] = "0"
+
+        try:
+            output = _check_output(commands, env=environ)
+        except SubprocessCalledProcessError as e:
+            raise TankGitError(
+                "Error executing git operation '%s': %s (Return code %s)"
+                % (commands, e.output, e.returncode)
+            )
+        else:
+            output = output.strip().strip("'")
+            log.debug("Execution successful. stderr/stdout: '%s'" % output)
+            return output
 
     @LogManager.log_timing
     def _clone_then_execute_git_commands(
@@ -124,127 +215,19 @@
         """
         # ensure *parent* folder exists
         parent_folder = os.path.dirname(target_path)
-
-        filesystem.ensure_folder_exists(parent_folder)
-
-        # first probe to check that git exists in our PATH
-        log.debug("Checking that git exists and can be executed...")
-        try:
-            output = _check_output(["git", "--version"])
-        except:
-            log.exception("Unexpected error:")
-            raise TankGitError(
-                "Cannot execute the 'git' command. Please make sure that git is "
-                "installed on your system and that the git executable has been added to the PATH."
-            )
-
-        log.debug("Git installed: %s" % output)
+        os.makedirs(parent_folder, exist_ok=True)
 
         # Make sure all git commands are correct according to the descriptor type
-        cmd = self._validate_git_commands(
+        cmd = self._get_git_clone_commands(
             target_path, depth=depth, ref=ref, is_latest_commit=is_latest_commit
         )
+        self._execute_git_commands(cmd)
 
-        run_with_os_system = True
+        if commands:
+            full_commands = ["git", "-C", os.path.normpath(target_path)]
+            full_commands.extend(commands)
 
-        # We used to call only os.system here. On macOS and Linux this behaved correctly,
-        # i.e. if stdin was open you would be prompted on the terminal and if not then an
-        # error would be raised. This is the best we could do.
-        #
-        # However, for Windows, os.system actually pops a terminal, which is annoying, especially
-        # if you don't require to authenticate. To avoid this popup, we will first launch
-        # git through the subprocess module and instruct it to not show a terminal for the
-        # subprocess.
-        #
-        # If that fails, then we'll assume that it failed because credentials were required.
-        # Unfortunately, we can't tell why it failed.
-        #
-        # Note: We only try this workflow if we can actually hide the terminal on Windows.
-        # If we can't there's no point doing all of this and we should just use
-        # os.system.
-        if is_windows() and _can_hide_terminal():
-            log.debug("Executing command '%s' using subprocess module." % cmd)
-            try:
-                # It's important to pass GIT_TERMINAL_PROMPT=0 or the git subprocess will
-                # just hang waiting for credentials to be entered on the missing terminal.
-                # I would have expected Windows to give an error about stdin being close and
-                # aborting the git command but at least on Windows 10 that is not the case.
-                environ = {}
-                environ.update(os.environ)
-                environ["GIT_TERMINAL_PROMPT"] = "0"
-                _check_output(cmd, env=environ)
-
-                # If that works, we're done and we don't need to use os.system.
-                run_with_os_system = False
-                status = 0
-            except SubprocessCalledProcessError:
-                log.debug("Subprocess call failed.")
-
-        if run_with_os_system:
-            # Make sure path and repo path are quoted.
-            log.debug("Executing command '%s' using os.system" % cmd)
-            log.debug(
-                "Note: in a terminal environment, this may prompt for authentication"
-            )
-            status = os.system(cmd)
-
-        log.debug("Command returned exit code %s" % status)
-        if status != 0:
-            raise TankGitError(
-                "Error executing git operation. The git command '%s' "
-                "returned error code %s." % (cmd, status)
-            )
-        log.debug("Git clone into '%s' successful." % target_path)
-
-        # clone worked ok! Now execute git commands on this repo
-
-        output = None
-
-        for command in commands:
-            # we use git -C to specify the working directory where to execute the command
-            # this option was added in as part of git 1.9
-            # and solves an issue with UNC paths on windows.
-            full_command = 'git -C "%s" %s' % (target_path, command)
-            log.debug("Executing '%s'" % full_command)
-
-            try:
-                output = _check_output(full_command, shell=True)
-
-                # note: it seems on windows, the result is sometimes wrapped in single quotes.
-                output = output.strip().strip("'")
-
-            except SubprocessCalledProcessError as e:
-                raise TankGitError(
-                    f"Error executing GIT operation '{full_command}': {e.output}"
-                    f" (Return code {e.returncode}). "
-                    " Supported GIT version: 1.9+."
-                )
-            log.debug("Execution successful. stderr/stdout: '%s'" % output)
-
-        # return the last returned stdout/stderr
-        return output
-
-    def _tmp_clone_then_execute_git_commands(self, commands, depth=None, ref=None):
-        """
-        Clone into a temp location and executes the given
-        list of git commands.
-
-        For more details, see :meth:`_clone_then_execute_git_commands`.
-
-        :param commands: list git commands to execute, e.g. ['checkout x']
-        :returns: stdout and stderr of the last command executed as a string
-        """
-        clone_tmp = os.path.join(
-            tempfile.gettempdir(), "sgtk_clone_%s" % uuid.uuid4().hex
-        )
-        filesystem.ensure_folder_exists(clone_tmp)
-        try:
-            return self._clone_then_execute_git_commands(
-                clone_tmp, commands, depth, ref
-            )
-        finally:
-            log.debug("Cleaning up temp location '%s'" % clone_tmp)
-            shutil.rmtree(clone_tmp, ignore_errors=True)
+            return self._execute_git_commands(full_commands)
 
     def get_system_name(self):
         """
@@ -268,10 +251,9 @@
         can_connect = True
         try:
             log.debug("%r: Probing if a connection to git can be established..." % self)
-            # clone repo into temp folder
-            self._tmp_clone_then_execute_git_commands([], depth=1)
+            self._execute_git_commands(["git", "ls-remote", "--heads", self._path])
             log.debug("...connection established")
-        except Exception as e:
+        except (OSError, SubprocessCalledProcessError) as e:
             log.debug("...could not establish connection: %s" % e)
             can_connect = False
         return can_connect
@@ -293,17 +275,44 @@
         # make sure item exists locally
         self.ensure_local()
         # copy descriptor into target.
-        # the skip list contains .git folders by default, so pass in []
-        # to turn that restriction off. In the case of the git descriptor,
-        # we want to transfer this folder as well.
-        filesystem.copy_folder(
-            self.get_path(),
-            target_path,
-            # Make we do not pass none or we will be getting the default skip list.
-            skip_list=skip_list or [],
-        )
+        filesystem.copy_folder(self.get_path(), target_path, skip_list=skip_list or [])
+
+    def _normalize_path(self, path):
+        if path.endswith("/") or path.endswith("\\"):
+            new_path = path[:-1]
+        else:
+            new_path = path
+
+        if os.path.isdir(new_path):
+            if not new_path.endswith(".git"):
+                new_path = os.path.join(new_path, ".git")
+
+        return new_path
+
+    def _path_is_local(self, path):
+        """
+        Check if path value is an existing folder, and if contain a .git folder.
+        """
+        if os.path.isdir(path):
+            output = self._execute_git_commands(
+                [
+                    "git",
+                    "-C",
+                    os.path.normpath(path),
+                    "rev-parse",
+                    "--git-dir",
+                ]
+            )
+            if output.startswith("fatal: not a git repository"):
+                raise TankDescriptorError(
+                    "Folder is not a git repository: {}".format(path)
+                )
+            elif output == ".":
+                return True
+
+        return False
 
-    def _validate_git_commands(
+    def _get_git_clone_commands(
         self, target_path, depth=None, ref=None, is_latest_commit=None
     ):
         """
@@ -323,23 +332,22 @@
         # complications in cleanup scenarios and with file copying. We want
         # each repo that we clone to be completely independent on a filesystem level.
         log.debug("Git Cloning %r into %s" % (self, target_path))
-        depth = "--depth %s" % depth if depth else ""
-        ref = "-b %s" % ref if ref else ""
-        cmd = 'git clone --no-hardlinks -q "%s" %s "%s" %s' % (
-            self._path,
-            ref,
-            target_path,
-            depth,
-        )
-        if self._descriptor_dict.get("type") == "git_branch":
-            if not is_latest_commit:
-                if "--depth" in cmd:
-                    depth = ""
-                    cmd = 'git clone --no-hardlinks -q "%s" %s "%s" %s' % (
-                        self._path,
-                        ref,
-                        target_path,
-                        depth,
-                    )
+
+        if self._descriptor_dict.get("type") == "git_branch" and not is_latest_commit:
+            depth = None
+
+        cmd = ["git", "clone", "--no-hardlinks", "-q"]
+
+        if ref:
+            cmd.extend(["-b", str(ref)])
+
+        if depth:
+            cmd.extend(["--depth", str(depth)])
+
+        cmd.append(self._path)
+        cmd.append(os.path.normpath(target_path))
 
         return cmd
+
+    def _fetch_local_data(self, path):
+        return {}