Consolidated 'git' refish parsing into a class

gclient_scm.py

 # Copyright (c) 2012 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Gclient-specific SCM-specific operations."""
 
 from __future__ import print_function
 
+import collections
 import errno
 import logging
 import os
 import posixpath
 import re
 import shlex
 import sys
 import tempfile
 import traceback
 import urlparse
@@ skipping 133 matching lines @@
 
     if not command in commands:
       raise gclient_utils.Error('Unknown command %s' % command)
 
     if not command in dir(self):
       raise gclient_utils.Error('Command %s not implemented in %s wrapper' % (
           command, self.__class__.__name__))
 
     return getattr(self, command)(options, args, file_list)
 
-  def GetActualRemoteURL(self, options):
+  def GetActualRemoteURL(self):
     """Attempt to determine the remote URL for this SCMWrapper."""
-    # Git
     if os.path.exists(os.path.join(self.checkout_path, '.git')):
       actual_remote_url = shlex.split(scm.GIT.Capture(
           ['config', '--local', '--get-regexp', r'remote.*.url'],
           cwd=self.checkout_path))[1]
 
       # If a cache_dir is used, obtain the actual remote URL from the cache.
       if getattr(self, 'cache_dir', None):
         url, _ = gclient_utils.SplitUrlRevision(self.url)
         mirror = git_cache.Mirror(url)
         if (mirror.exists() and mirror.mirror_path.replace('\\', '/') ==
             actual_remote_url.replace('\\', '/')):
           actual_remote_url = shlex.split(scm.GIT.Capture(
               ['config', '--local', '--get-regexp', r'remote.*.url'],
               cwd=mirror.mirror_path))[1]
       return actual_remote_url
 
     # Svn
     if os.path.exists(os.path.join(self.checkout_path, '.svn')):
       return scm.SVN.CaptureLocalInfo([], self.checkout_path)['URL']
     return None
 
   def DoesRemoteURLMatch(self, options):
     """Determine whether the remote URL of this checkout is the expected URL."""
     if not os.path.exists(self.checkout_path):
       # A checkout which doesn't exist can't be broken.
       return True
 
-    actual_remote_url = self.GetActualRemoteURL(options)
+    actual_remote_url = self.GetActualRemoteURL()
     if actual_remote_url:
       return (gclient_utils.SplitUrlRevision(actual_remote_url)[0].rstrip('/')
               == gclient_utils.SplitUrlRevision(self.url)[0].rstrip('/'))
     else:
       # This may occur if the self.checkout_path exists but does not contain a
       # valid git or svn checkout.
       return False
 
   def _DeleteOrMove(self, force):
     """Delete the checkout directory or move it out of the way.
@@ skipping 20 matching lines @@
       dest_path = tempfile.mkdtemp(
           prefix=os.path.basename(self.relpath),
           dir=bad_scm_dir)
       self.Print('_____ Conflicting directory found in %s. Moving to %s.'
                  % (self.checkout_path, dest_path))
       gclient_utils.AddWarning('Conflicting directory %s moved to %s.'
                                % (self.checkout_path, dest_path))
       shutil.move(self.checkout_path, dest_path)
 
 
+_GitRefishBase = collections.namedtuple('GitRef', (
+    # The initial string that the parsed Refish came from
+    'source',
+    # Is this ref a branch?
+    'is_branch',
+    # The name of this ref when accessed through the local repository
+    'local_ref',
+    # The name of the remote that this refish resides on
+    'remote',
+    # The path of this refish on the remote (e.g., 'master')
+    'remote_ref',
+    # The remote-qualified path to this refish (e.g., 'origin/master')
+    'remote_refspec',
+    # If a branch, the expected name of the upstream branch that it should
+    # track; otherwise, 'None'
+    'upstream_branch',
+))
+
+
+class GitRefish(_GitRefishBase):
+  # Disable 'no __init__' warning | pylint: disable=W0232
+  """Implements refish parsing and properties.
+
+  This class is designed to concentrate and codify the assumptions made by
+  'gclient' code about refs and revisions.
+  """
+
+  _DEFAULT_REMOTE = 'origin'
+
+  _GIT_SHA1_RE = re.compile('[0-9a-f]{40}', re.IGNORECASE)
+  _GIT_SHA1_SHORT_RE = re.compile('[0-9a-f]{4,40}', re.IGNORECASE)
+
+  def __str__(self):
+    return self.source
+
+  @classmethod
+  def Parse(cls, ref, remote=None, other_remotes=None):
+    """Parses a ref into a 'GitRefish' instance.
+
+    Args:
+      ref: (str) The refish (branch, tag, hash, etc.) to parse
+      remote: (None/str) If supplied, the name of the primary remote. In
+          addtion to being recognized as a remote string, the primary remote
+          will also be used (as needed) when generating the remote refspec. If
+          omitted, the default remote name ('origin') will be used.
+      other_remotes: (None/iterable) If supplied, a list of strings to
+          recognize as remotes in addition to 'remote'.
+    """
+    assert ref == ref.strip()
+
+    # Use default set of remotes
+    remote = remote or cls._DEFAULT_REMOTE
+    remotes = {remote}
+    if other_remotes:
+      remotes.update(other_remotes)
+
+    # Treat 'ref' as a '/'-delimited set of items; analyze their contents
+    ref_split = local_ref = remote_ref = tuple(ref.split('/'))
+    is_branch = True
+    if len(ref_split) > 1:
+      if ref_split[0] == 'refs':
+        if len(ref_split) >= 3 and ref_split[1] == 'heads':
+          # refs/heads/foo/bar
+          #
+          # Local Ref:  foo/bar
+          # Remote Ref: foo/bar
+          local_ref = remote_ref = ref_split[2:]
+        elif len(ref_split) >= 4 and ref_split[1] == 'remotes':
+          # refs/remotes/<REMOTE>/foo/bar
+          # This is a bad assumption, and this logic can be improved, but it
+          # is consistent with traditional 'gclient' logic.
+          #
+          # Local Ref:  refs/remotes/<REMOTE>/foo/bar
+          # Remote:     <REMOTE>
+          # Remote Ref: foo/bar
+          remote = ref_split[2]
+          remote_ref = ref_split[3:]
+      elif len(ref_split) >= 2 and ref_split[0] in remotes:
+        # origin/master
+        #
+        # Local Ref:  refs/remotes/origin/master
+        # Remote Ref: master
+        remote = ref_split[0]
+        remote_ref = ref_split[1:]
+        local_ref = ('refs', 'remotes') + ref_split
+
+      # (Default) The refish has multiple paths. Assume at least that it's a
+      # branch name.
+      #
+      # foo/bar
+      # refs/foo/bar
+      #
+      # Local Ref:  foo/bar
+      # Remote ref: foo/bar
+    else:
+      # It could be a hash, a short-hash, or a tag
+      is_branch = False
+
+    # Determine how the ref should be referenced remotely
+    if is_branch:
+      # foo/bar/baz => origin/foo/bar/baz
+      remote_refspec = (remote,) + remote_ref
+    else:
+      # Refer to the hash/tag directly. This is actually not allowed in
+      # 'fetch', although it oftentimes works regardless.
+      #
+      # <HASH/TAG> => <HASH/TAG>
+      remote_refspec = (ref,)
+
+    # Calculate the upstream branch
+    if is_branch:
+      # Convert '/refs/heads/...' to 'refs/remotes/REMOTE/...'
+      if ref_split[:2] == ('refs', 'heads'):
+        upstream_branch = ('refs', 'remotes', remote) + ref_split[2:]
+      else:
+        upstream_branch = ref_split
+    else:
+      upstream_branch = None
+
+    def compose(ref_tuple):
+      return '/'.join(ref_tuple) if ref_tuple else ref_tuple
+
+    return cls(
+        source=ref,
+        is_branch=is_branch,
+        local_ref=compose(local_ref),
+        remote=remote,
+        remote_ref=compose(remote_ref),
+        remote_refspec=compose(remote_refspec),
+        upstream_branch=compose(upstream_branch),
+    )
+
+
 class GitWrapper(SCMWrapper):
   """Wrapper for Git"""
   name = 'git'
   remote = 'origin'
 
   cache_dir = None
 
   def __init__(self, url=None, *args):
     """Removes 'git+' fake prefix from git URL."""
     if url.startswith('git+http://') or url.startswith('git+https://'):
       url = url[4:]
     SCMWrapper.__init__(self, url, *args)
     filter_kwargs = { 'time_throttle': 1, 'out_fh': self.out_fh }
     if self.out_cb:
       filter_kwargs['predicate'] = self.out_cb
     self.filter = gclient_utils.GitFilter(**filter_kwargs)
 
   @staticmethod
   def BinaryExists():
     """Returns true if the command exists."""
     try:
       # We assume git is newer than 1.7.  See: crbug.com/114483
       result, version = scm.GIT.AssertVersion('1.7')
       if not result:
         raise gclient_utils.Error('Git version is older than 1.7: %s' % version)
       return result
     except OSError:
       return False
 
+  def ParseRefish(self, value, **kwargs):
+    kwargs.setdefault('remote', self.remote)
+    return GitRefish.Parse(value, **kwargs)
+
   def GetCheckoutRoot(self):
     return scm.GIT.GetCheckoutRoot(self.checkout_path)
 
   def GetRevisionDate(self, _revision):
     """Returns the given revision's date in ISO-8601 format (which contains the
     time zone)."""
     # TODO(floitsch): get the time-stamp of the given revision and not just the
     # time-stamp of the currently checked out revision.
     return self._Capture(['log', '-n', '1', '--format=%ai'])
 
@@ skipping 14 matching lines @@
 
     The patch file is generated from a diff of the merge base of HEAD and
     its upstream branch.
     """
     merge_base = self._Capture(['merge-base', 'HEAD', self.remote])
     gclient_utils.CheckCallAndFilter(
         ['git', 'diff', merge_base],
         cwd=self.checkout_path,
         filter_fn=GitDiffFilterer(self.relpath).Filter, print_func=self.Print)
 
-  def _FetchAndReset(self, revision, file_list, options):
+  def _FetchAndReset(self, refish, file_list, options):
     """Equivalent to git fetch; git reset."""
     quiet = []
     if not options.verbose:
       quiet = ['--quiet']
     self._UpdateBranchHeads(options, fetch=False)
 
     self._Fetch(options, prune=True, quiet=options.verbose)
-    self._Run(['reset', '--hard', revision] + quiet, options)
+    self._Run(['reset', '--hard', refish.local_ref] + quiet, options)
     if file_list is not None:
       files = self._Capture(['ls-files']).splitlines()
       file_list.extend([os.path.join(self.checkout_path, f) for f in files])
 
   def _DisableHooks(self):
     hook_dir = os.path.join(self.checkout_path, '.git', 'hooks')
     if not os.path.isdir(hook_dir):
       return
     for f in os.listdir(hook_dir):
       if not f.endswith('.sample') and not f.endswith('.disabled'):
         os.rename(os.path.join(hook_dir, f),
                   os.path.join(hook_dir, f + '.disabled'))
 
   def update(self, options, args, file_list):
     """Runs git to update or transparently checkout the working copy.
 
     All updated files will be appended to file_list.
 
     Raises:
       Error: if can't get URL for relative path.
     """
     if args:
       raise gclient_utils.Error("Unsupported argument(s): %s" % ",".join(args))
 
     self._CheckMinVersion("1.6.6")
 
     # If a dependency is not pinned, track the default remote branch.
-    default_rev = 'refs/remotes/%s/master' % self.remote
+    default_rev = 'refs/remotes/%s/master' % (self.remote,)
     url, deps_revision = gclient_utils.SplitUrlRevision(self.url)
     rev_str = ""
     revision = deps_revision
     managed = True
     if options.revision:
       # Override the revision number.
       revision = str(options.revision)
     if revision == 'unmanaged':
       # Check again for a revision in case an initial ref was specified
       # in the url, for example bla.git@refs/heads/custombranch
@@ skipping 16 matching lines @@
     rev_str = ' at %s' % revision
     files = [] if file_list is not None else None
 
     printed_path = False
     verbose = []
     if options.verbose:
       self.Print('_____ %s%s' % (self.relpath, rev_str), timestamp=False)
       verbose = ['--verbose']
       printed_path = True
 
-    if revision.startswith('refs/'):
-      rev_type = "branch"
-    elif revision.startswith(self.remote + '/'):
-      # Rewrite remote refs to their local equivalents.
-      revision = 'refs/remotes/' + revision
-      rev_type = "branch"
-    else:
-      # hash is also a tag, only make a distinction at checkout
-      rev_type = "hash"
-
+    refish = self.ParseRefish(revision)
     mirror = self._GetMirror(url, options)
     if mirror:
       url = mirror.mirror_path
 
     if (not os.path.exists(self.checkout_path) or
         (os.path.isdir(self.checkout_path) and
          not os.path.exists(os.path.join(self.checkout_path, '.git')))):
       if mirror:
         self._UpdateMirror(mirror, options)
       try:
-        self._Clone(revision, url, options)
+        self._Clone(refish.local_ref, url, options)
       except subprocess2.CalledProcessError:
         self._DeleteOrMove(options.force)
-        self._Clone(revision, url, options)
+        self._Clone(refish.local_ref, url, options)
       if deps_revision and deps_revision.startswith('branch-heads/'):
         deps_branch = deps_revision.replace('branch-heads/', '')
         self._Capture(['branch', deps_branch, deps_revision])
         self._Checkout(options, deps_branch, quiet=True)
       if file_list is not None:
         files = self._Capture(['ls-files']).splitlines()
         file_list.extend([os.path.join(self.checkout_path, f) for f in files])
       if not verbose:
         # Make the output a little prettier. It's nice to have some whitespace
         # between projects when cloning.
@@ skipping 20 matching lines @@
     if (current_url.rstrip('/') != url.rstrip('/') and
         url != 'git://foo' and
         subprocess2.capture(
             ['git', 'config', 'remote.%s.gclient-auto-fix-url' % self.remote],
             cwd=self.checkout_path).strip() != 'False'):
       self.Print('_____ switching %s to a new upstream' % self.relpath)
       # Make sure it's clean
       self._CheckClean(rev_str)
       # Switch over to the new upstream
       self._Run(['remote', 'set-url', self.remote, url], options)
-      self._FetchAndReset(revision, file_list, options)
+      self._FetchAndReset(refish, file_list, options)
       return_early = True
 
     if return_early:
       return self._Capture(['rev-parse', '--verify', 'HEAD'])
 
     cur_branch = self._GetCurrentBranch()
 
     # Cases:
     # 0) HEAD is detached. Probably from our initial clone.
     #   - make sure HEAD is contained by a named ref, then update.
@@ skipping 22 matching lines @@
       upstream_branch = scm.GIT.GetUpstreamBranch(self.checkout_path)
       if not upstream_branch or not upstream_branch.startswith('refs/remotes'):
         current_type = "hash"
         logging.debug("Current branch is not tracking an upstream (remote)"
                       " branch.")
       elif upstream_branch.startswith('refs/remotes'):
         current_type = "branch"
       else:
         raise gclient_utils.Error('Invalid Upstream: %s' % upstream_branch)
 
-    if not scm.GIT.IsValidRevision(self.checkout_path, revision, sha_only=True):
+    if not scm.GIT.IsValidRevision(self.checkout_path, refish.local_ref,
+                                   sha_only=True):
       # Update the remotes first so we have all the refs.
       remote_output = scm.GIT.Capture(['remote'] + verbose + ['update'],
               cwd=self.checkout_path)
       if verbose:
         self.Print(remote_output)
 
       self._UpdateBranchHeads(options, fetch=True)
 
     # This is a big hammer, debatable if it should even be here...
     if options.force or options.reset:
       target = 'HEAD'
       if options.upstream and upstream_branch:
         target = upstream_branch
       self._Run(['reset', '--hard', target], options)
 
     if current_type == 'detached':
       # case 0
       self._CheckClean(rev_str)
       self._CheckDetachedHead(rev_str, options)
-      if self._Capture(['rev-list', '-n', '1', 'HEAD']) == revision:
+      if self._Capture(['rev-list', '-n', '1', 'HEAD']) == refish.local_ref:
         self.Print('Up-to-date; skipping checkout.')
       else:
         # 'git checkout' may need to overwrite existing untracked files. Allow
         # it only when nuclear options are enabled.
         self._Checkout(
             options,
             revision,
             force=(options.force and options.delete_unversioned_trees),
             quiet=True,
         )
       if not printed_path:
         self.Print('_____ %s%s' % (self.relpath, rev_str), timestamp=False)
     elif current_type == 'hash':
       # case 1
       if scm.GIT.IsGitSvn(self.checkout_path) and upstream_branch is not None:
         # Our git-svn branch (upstream_branch) is our upstream
         self._AttemptRebase(upstream_branch, files, options,
-                            newbase=revision, printed_path=printed_path,
-                            merge=options.merge)
+                            newbase=refish.local_ref,
+                            printed_path=printed_path, merge=options.merge)
         printed_path = True
       else:
         # Can't find a merge-base since we don't know our upstream. That makes
         # this command VERY likely to produce a rebase failure. For now we
         # assume origin is our upstream since that's what the old behavior was.
         upstream_branch = self.remote
         if options.revision or deps_revision:
-          upstream_branch = revision
+          upstream_branch = refish.local_ref
         self._AttemptRebase(upstream_branch, files, options,
                             printed_path=printed_path, merge=options.merge)
         printed_path = True
-    elif rev_type == 'hash':
+    elif not refish.is_branch:
       # case 2
       self._AttemptRebase(upstream_branch, files, options,
-                          newbase=revision, printed_path=printed_path,
+                          newbase=refish.local_ref, printed_path=printed_path,
                           merge=options.merge)
       printed_path = True
-    elif revision.replace('heads', 'remotes/' + self.remote) != upstream_branch:
+    elif refish.upstream_branch != upstream_branch:
       # case 4
-      new_base = revision.replace('heads', 'remotes/' + self.remote)
+      new_base = refish.upstream_branch
       if not printed_path:
         self.Print('_____ %s%s' % (self.relpath, rev_str), timestamp=False)
       switch_error = ("Switching upstream branch from %s to %s\n"
                      % (upstream_branch, new_base) +
                      "Please merge or rebase manually:\n" +
                      "cd %s; git rebase %s\n" % (self.checkout_path, new_base) +
                      "OR git checkout -b <some new branch> %s" % new_base)
       raise gclient_utils.Error(switch_error)
     else:
       # case 3 - the default case
@@ skipping 110 matching lines @@
       return self.update(options, [], file_list)
 
     default_rev = "refs/heads/master"
     if options.upstream:
       if self._GetCurrentBranch():
         upstream_branch = scm.GIT.GetUpstreamBranch(self.checkout_path)
         default_rev = upstream_branch or default_rev
     _, deps_revision = gclient_utils.SplitUrlRevision(self.url)
     if not deps_revision:
       deps_revision = default_rev
-    if deps_revision.startswith('refs/heads/'):
-      deps_revision = deps_revision.replace('refs/heads/', self.remote + '/')
-    deps_revision = self.GetUsableRev(deps_revision, options)
+    refish = self.ParseRefish(deps_revision)
+    deps_revision = self.GetUsableRev(refish.remote_refspec, options)
 
     if file_list is not None:
       files = self._Capture(['diff', deps_revision, '--name-only']).split()
 
     self._Run(['reset', '--hard', deps_revision], options)
     self._Run(['clean', '-f', '-d'], options)
 
     if file_list is not None:
       file_list.extend([os.path.join(self.checkout_path, f) for f in files])
 
@@ skipping 156 matching lines @@
       gclient_utils.safe_rename(os.path.join(tmp_dir, '.git'),
                                 os.path.join(self.checkout_path, '.git'))
     except:
       traceback.print_exc(file=self.out_fh)
       raise
     finally:
       if os.listdir(tmp_dir):
         self.Print('_____ removing non-empty tmp dir %s' % tmp_dir)
       gclient_utils.rmtree(tmp_dir)
     self._UpdateBranchHeads(options, fetch=True)
-    self._Checkout(options, revision.replace('refs/heads/', ''), quiet=True)
+
+    refish = self.ParseRefish(revision)
+    self._Checkout(options, refish.local_ref, quiet=True)
     if self._GetCurrentBranch() is None:
       # Squelch git's very verbose detached HEAD warning and use our own
       self.Print(
-        ('Checked out %s to a detached HEAD. Before making any commits\n'
-         'in this repo, you should use \'git checkout <branch>\' to switch to\n'
-         'an existing branch or use \'git checkout %s -b <branch>\' to\n'
-         'create a new branch for your work.') % (revision, self.remote))
+        "Checked out %s to a detached HEAD. Before making any commits\n"
+        "in this repo, you should use 'git checkout <branch>' to switch to\n"
+        "an existing branch or use 'git checkout %s -b <branch>' to\n"
+        "create a new branch for your work." % (revision, self.remote))
 
   def _AskForData(self, prompt, options):
     if options.jobs > 1:
       self.Print(prompt)
       raise gclient_utils.Error("Background task requires input. Rerun "
                                 "gclient with --jobs=1 so that\n"
                                 "interaction is possible.")
     try:
       return raw_input(prompt)
     except KeyboardInterrupt:
@@ skipping 676 matching lines @@
       new_command.append('--force')
       if command[0] != 'checkout' and scm.SVN.AssertVersion('1.6')[0]:
         new_command.extend(('--accept', 'theirs-conflict'))
     elif options.manually_grab_svn_rev:
       new_command.append('--force')
       if command[0] != 'checkout' and scm.SVN.AssertVersion('1.6')[0]:
         new_command.extend(('--accept', 'postpone'))
     elif command[0] != 'checkout' and scm.SVN.AssertVersion('1.6')[0]:
       new_command.extend(('--accept', 'postpone'))
     return new_command