gclient_scm.py

# Copyright 2009 Google Inc.  All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import os
import re
import subprocess
import sys
import xml.dom.minidom

import gclient_utils

SVN_COMMAND = "svn"


### SCM abstraction layer


# Factory Method for SCM wrapper creation

def create_scm(url, root_dir, relpath, scm_name='svn'):
  # TODO(maruel): Deduce the SCM from the url.
  scm_map = {
    'svn' : SVNWrapper,
  }
  if not scm_name in scm_map:
    raise gclient_utils.Error('Unsupported scm %s' % scm_name)
  return scm_map[scm_name](url, root_dir, relpath, scm_name)


# SCMWrapper base class

class SCMWrapper(object):
  """Add necessary glue between all the supported SCM.

  This is the abstraction layer to bind to different SCM. Since currently only
  subversion is supported, a lot of subersionism remains. This can be sorted out
  once another SCM is supported."""
  def __init__(self, url, root_dir, relpath, scm_name='svn'):
    self.scm_name = scm_name
    self.url = url
    self._root_dir = root_dir.replace('/', os.sep)
    self.relpath = relpath.replace('/', os.sep)

  def FullUrlForRelativeUrl(self, url):
    # Find the forth '/' and strip from there. A bit hackish.
    return '/'.join(self.url.split('/')[:4]) + url

  def RunCommand(self, command, options, args, file_list=None):
    # file_list will have all files that are modified appended to it.
    if file_list is None:
      file_list = []

    commands = ['cleanup', 'export', 'update', 'revert',
                'status', 'diff', 'pack', 'runhooks']

    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 implemnted in %s wrapper' % (
          command, self.scm_name))

    return getattr(self, command)(options, args, file_list)


class SVNWrapper(SCMWrapper):
  """ Wrapper for SVN """

  def cleanup(self, options, args, file_list):
    """Cleanup working copy."""
    command = ['cleanup']
    command.extend(args)
    RunSVN(command, os.path.join(self._root_dir, self.relpath))

  def diff(self, options, args, file_list):
    # NOTE: This function does not currently modify file_list.
    command = ['diff']
    command.extend(args)
    RunSVN(command, os.path.join(self._root_dir, self.relpath))

  def export(self, options, args, file_list):
    assert len(args) == 1
    export_path = os.path.abspath(os.path.join(args[0], self.relpath))
    try:
      os.makedirs(export_path)
    except OSError:
      pass
    assert os.path.exists(export_path)
    command = ['export', '--force', '.']
    command.append(export_path)
    RunSVN(command, os.path.join(self._root_dir, self.relpath))

  def update(self, options, args, file_list):
    """Runs SCM 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.
    """
    # Only update if git is not controlling the directory.
    checkout_path = os.path.join(self._root_dir, self.relpath)
    git_path = os.path.join(self._root_dir, self.relpath, '.git')
    if os.path.exists(git_path):
      print("________ found .git directory; skipping %s" % self.relpath)
      return

    if args:
      raise gclient_utils.Error("Unsupported argument(s): %s" % ",".join(args))

    url = self.url
    components = url.split("@")
    revision = None
    forced_revision = False
    if options.revision:
      # Override the revision number.
      url = '%s@%s' % (components[0], str(options.revision))
      revision = int(options.revision)
      forced_revision = True
    elif len(components) == 2:
      revision = int(components[1])
      forced_revision = True

    rev_str = ""
    if revision:
      rev_str = ' at %d' % revision

    if not os.path.exists(checkout_path):
      # We need to checkout.
      command = ['checkout', url, checkout_path]
      if revision:
        command.extend(['--revision', str(revision)])
      RunSVNAndGetFileList(command, self._root_dir, file_list)
      return

    # Get the existing scm url and the revision number of the current checkout.
    from_info = CaptureSVNInfo(os.path.join(checkout_path, '.'), '.')
    if not from_info:
      raise gclient_utils.Error("Can't update/checkout %r if an unversioned "
                                "directory is present. Delete the directory "
                                "and try again." %
                                checkout_path)

    if options.manually_grab_svn_rev:
      # Retrieve the current HEAD version because svn is slow at null updates.
      if not revision:
        from_info_live = CaptureSVNInfo(from_info['URL'], '.')
        revision = int(from_info_live['Revision'])
        rev_str = ' at %d' % revision

    if from_info['URL'] != components[0]:
      to_info = CaptureSVNInfo(url, '.')
      can_switch = ((from_info['Repository Root'] != to_info['Repository Root'])
                    and (from_info['UUID'] == to_info['UUID']))
      if can_switch:
        print("\n_____ relocating %s to a new checkout" % self.relpath)
        # We have different roots, so check if we can switch --relocate.
        # Subversion only permits this if the repository UUIDs match.
        # Perform the switch --relocate, then rewrite the from_url
        # to reflect where we "are now."  (This is the same way that
        # Subversion itself handles the metadata when switch --relocate
        # is used.)  This makes the checks below for whether we
        # can update to a revision or have to switch to a different
        # branch work as expected.
        # TODO(maruel):  TEST ME !
        command = ["switch", "--relocate",
                   from_info['Repository Root'],
                   to_info['Repository Root'],
                   self.relpath]
        RunSVN(command, self._root_dir)
        from_info['URL'] = from_info['URL'].replace(
            from_info['Repository Root'],
            to_info['Repository Root'])
      else:
        if CaptureSVNStatus(checkout_path):
          raise gclient_utils.Error("Can't switch the checkout to %s; UUID "
                                    "don't match and there is local changes "
                                    "in %s. Delete the directory and "
                                    "try again." % (url, checkout_path))
        # Ok delete it.
        print("\n_____ switching %s to a new checkout" % self.relpath)
        gclient_utils.RemoveDirectory(checkout_path)
        # We need to checkout.
        command = ['checkout', url, checkout_path]
        if revision:
          command.extend(['--revision', str(revision)])
        RunSVNAndGetFileList(command, self._root_dir, file_list)
        return


    # If the provided url has a revision number that matches the revision
    # number of the existing directory, then we don't need to bother updating.
    if not options.force and from_info['Revision'] == revision:
      if options.verbose or not forced_revision:
        print("\n_____ %s%s" % (self.relpath, rev_str))
      return

    command = ["update", checkout_path]
    if revision:
      command.extend(['--revision', str(revision)])
    RunSVNAndGetFileList(command, self._root_dir, file_list)

  def revert(self, options, args, file_list):
    """Reverts local modifications. Subversion specific.

    All reverted files will be appended to file_list, even if Subversion
    doesn't know about them.
    """
    path = os.path.join(self._root_dir, self.relpath)
    if not os.path.isdir(path):
      # svn revert won't work if the directory doesn't exist. It needs to
      # checkout instead.
      print("\n_____ %s is missing, synching instead" % self.relpath)
      # Don't reuse the args.
      return self.update(options, [], file_list)

    files = CaptureSVNStatus(path)
    # Batch the command.
    files_to_revert = []
    for file in files:
      file_path = os.path.join(path, file[1])
      print(file_path)
      # Unversioned file or unexpected unversioned file.
      if file[0][0] in ('?', '~'):
        # Remove extraneous file. Also remove unexpected unversioned
        # directories. svn won't touch them but we want to delete these.
        file_list.append(file_path)
        try:
          os.remove(file_path)
        except EnvironmentError:
          gclient_utils.RemoveDirectory(file_path)

      if file[0][0] != '?':
        # For any other status, svn revert will work.
        file_list.append(file_path)
        files_to_revert.append(file[1])

    # Revert them all at once.
    if files_to_revert:
      accumulated_paths = []
      accumulated_length = 0
      command = ['revert']
      for p in files_to_revert:
        # Some shell have issues with command lines too long.
        if accumulated_length and accumulated_length + len(p) > 3072:
          RunSVN(command + accumulated_paths,
                 os.path.join(self._root_dir, self.relpath))
          accumulated_paths = [p]
          accumulated_length = len(p)
        else:
          accumulated_paths.append(p)
          accumulated_length += len(p)
      if accumulated_paths:
        RunSVN(command + accumulated_paths,
               os.path.join(self._root_dir, self.relpath))

  def runhooks(self, options, args, file_list):
    self.status(options, args, file_list)

  def status(self, options, args, file_list):
    """Display status information."""
    path = os.path.join(self._root_dir, self.relpath)
    command = ['status']
    command.extend(args)
    if not os.path.isdir(path):
      # svn status won't work if the directory doesn't exist.
      print("\n________ couldn't run \'%s\' in \'%s\':\nThe directory "
            "does not exist."
            % (' '.join(command), path))
      # There's no file list to retrieve.
    else:
      RunSVNAndGetFileList(command, path, file_list)

  def pack(self, options, args, file_list):
    """Generates a patch file which can be applied to the root of the
    repository."""
    path = os.path.join(self._root_dir, self.relpath)
    command = ['diff']
    command.extend(args)
    # Simple class which tracks which file is being diffed and
    # replaces instances of its file name in the original and
    # working copy lines of the svn diff output.
    class DiffFilterer(object):
      index_string = "Index: "
      original_prefix = "--- "
      working_prefix = "+++ "

      def __init__(self, relpath):
        # Note that we always use '/' as the path separator to be
        # consistent with svn's cygwin-style output on Windows
        self._relpath = relpath.replace("\\", "/")
        self._current_file = ""
        self._replacement_file = ""

      def SetCurrentFile(self, file):
        self._current_file = file
        # Note that we always use '/' as the path separator to be
        # consistent with svn's cygwin-style output on Windows
        self._replacement_file = self._relpath + '/' + file

      def ReplaceAndPrint(self, line):
        print(line.replace(self._current_file, self._replacement_file))

      def Filter(self, line):
        if (line.startswith(self.index_string)):
          self.SetCurrentFile(line[len(self.index_string):])
          self.ReplaceAndPrint(line)
        else:
          if (line.startswith(self.original_prefix) or
              line.startswith(self.working_prefix)):
            self.ReplaceAndPrint(line)
          else:
            print line

    filterer = DiffFilterer(self.relpath)
    RunSVNAndFilterOutput(command, path, False, False, filterer.Filter)


# -----------------------------------------------------------------------------
# SVN utils:


def RunSVN(args, in_directory):
  """Runs svn, sending output to stdout.

  Args:
    args: A sequence of command line parameters to be passed to svn.
    in_directory: The directory where svn is to be run.

  Raises:
    Error: An error occurred while running the svn command.
  """
  c = [SVN_COMMAND]
  c.extend(args)

  gclient_utils.SubprocessCall(c, in_directory)


def CaptureSVN(args, in_directory=None, print_error=True):
  """Runs svn, capturing output sent to stdout as a string.

  Args:
    args: A sequence of command line parameters to be passed to svn.
    in_directory: The directory where svn is to be run.

  Returns:
    The output sent to stdout as a string.
  """
  c = [SVN_COMMAND]
  c.extend(args)

  # *Sigh*:  Windows needs shell=True, or else it won't search %PATH% for
  # the svn.exe executable, but shell=True makes subprocess on Linux fail
  # when it's called with a list because it only tries to execute the
  # first string ("svn").
  stderr = None
  if not print_error:
    stderr = subprocess.PIPE
  return subprocess.Popen(c,
                          cwd=in_directory,
                          shell=(sys.platform == 'win32'),
                          stdout=subprocess.PIPE,
                          stderr=stderr).communicate()[0]


def RunSVNAndGetFileList(args, in_directory, file_list):
  """Runs svn checkout, update, or status, output to stdout.

  The first item in args must be either "checkout", "update", or "status".

  svn's stdout is parsed to collect a list of files checked out or updated.
  These files are appended to file_list.  svn's stdout is also printed to
  sys.stdout as in RunSVN.

  Args:
    args: A sequence of command line parameters to be passed to svn.
    in_directory: The directory where svn is to be run.

  Raises:
    Error: An error occurred while running the svn command.
  """
  command = [SVN_COMMAND]
  command.extend(args)

  # svn update and svn checkout use the same pattern: the first three columns
  # are for file status, property status, and lock status.  This is followed
  # by two spaces, and then the path to the file.
  update_pattern = '^...  (.*)$'

  # The first three columns of svn status are the same as for svn update and
  # svn checkout.  The next three columns indicate addition-with-history,
  # switch, and remote lock status.  This is followed by one space, and then
  # the path to the file.
  status_pattern = '^...... (.*)$'

  # args[0] must be a supported command.  This will blow up if it's something
  # else, which is good.  Note that the patterns are only effective when
  # these commands are used in their ordinary forms, the patterns are invalid
  # for "svn status --show-updates", for example.
  pattern = {
        'checkout': update_pattern,
        'status':   status_pattern,
        'update':   update_pattern,
      }[args[0]]

  compiled_pattern = re.compile(pattern)

  def CaptureMatchingLines(line):
    match = compiled_pattern.search(line)
    if match:
      file_list.append(match.group(1))

  RunSVNAndFilterOutput(args,
                        in_directory,
                        True,
                        True,
                        CaptureMatchingLines)

def RunSVNAndFilterOutput(args,
                          in_directory,
                          print_messages,
                          print_stdout,
                          filter):
  """Runs svn checkout, update, status, or diff, optionally outputting
  to stdout.

  The first item in args must be either "checkout", "update",
  "status", or "diff".

  svn's stdout is passed line-by-line to the given filter function. If
  print_stdout is true, it is also printed to sys.stdout as in RunSVN.

  Args:
    args: A sequence of command line parameters to be passed to svn.
    in_directory: The directory where svn is to be run.
    print_messages: Whether to print status messages to stdout about
      which Subversion commands are being run.
    print_stdout: Whether to forward Subversion's output to stdout.
    filter: A function taking one argument (a string) which will be
      passed each line (with the ending newline character removed) of
      Subversion's output for filtering.

  Raises:
    Error: An error occurred while running the svn command.
  """
  command = [SVN_COMMAND]
  command.extend(args)

  gclient_utils.SubprocessCallAndFilter(command,
                                        in_directory,
                                        print_messages,
                                        print_stdout,
                                        filter=filter)

def CaptureSVNInfo(relpath, in_directory=None, print_error=True):
  """Returns a dictionary from the svn info output for the given file.

  Args:
    relpath: The directory where the working copy resides relative to
      the directory given by in_directory.
    in_directory: The directory where svn is to be run.
  """
  output = CaptureSVN(["info", "--xml", relpath], in_directory, print_error)
  dom = gclient_utils.ParseXML(output)
  result = {}
  if dom:
    GetNamedNodeText = gclient_utils.GetNamedNodeText
    GetNodeNamedAttributeText = gclient_utils.GetNodeNamedAttributeText
    def C(item, f):
      if item is not None: return f(item)
    # /info/entry/
    #   url
    #   reposityory/(root|uuid)
    #   wc-info/(schedule|depth)
    #   commit/(author|date)
    # str() the results because they may be returned as Unicode, which
    # interferes with the higher layers matching up things in the deps
    # dictionary.
    # TODO(maruel): Fix at higher level instead (!)
    result['Repository Root'] = C(GetNamedNodeText(dom, 'root'), str)
    result['URL'] = C(GetNamedNodeText(dom, 'url'), str)
    result['UUID'] = C(GetNamedNodeText(dom, 'uuid'), str)
    result['Revision'] = C(GetNodeNamedAttributeText(dom, 'entry', 'revision'),
                           int)
    result['Node Kind'] = C(GetNodeNamedAttributeText(dom, 'entry', 'kind'),
                            str)
    result['Schedule'] = C(GetNamedNodeText(dom, 'schedule'), str)
    result['Path'] = C(GetNodeNamedAttributeText(dom, 'entry', 'path'), str)
    result['Copied From URL'] = C(GetNamedNodeText(dom, 'copy-from-url'), str)
    result['Copied From Rev'] = C(GetNamedNodeText(dom, 'copy-from-rev'), str)
  return result


def CaptureSVNHeadRevision(url):
  """Get the head revision of a SVN repository.

  Returns:
    Int head revision
  """
  info = CaptureSVN(["info", "--xml", url], os.getcwd())
  dom = xml.dom.minidom.parseString(info)
  return int(dom.getElementsByTagName('entry')[0].getAttribute('revision'))


def CaptureSVNStatus(files):
  """Returns the svn 1.5 svn status emulated output.

  @files can be a string (one file) or a list of files.

  Returns an array of (status, file) tuples."""
  command = ["status", "--xml"]
  if not files:
    pass
  elif isinstance(files, basestring):
    command.append(files)
  else:
    command.extend(files)

  status_letter = {
    None: ' ',
    '': ' ',
    'added': 'A',
    'conflicted': 'C',
    'deleted': 'D',
    'external': 'X',
    'ignored': 'I',
    'incomplete': '!',
    'merged': 'G',
    'missing': '!',
    'modified': 'M',
    'none': ' ',
    'normal': ' ',
    'obstructed': '~',
    'replaced': 'R',
    'unversioned': '?',
  }
  dom = gclient_utils.ParseXML(CaptureSVN(command))
  results = []
  if dom:
    # /status/target/entry/(wc-status|commit|author|date)
    for target in dom.getElementsByTagName('target'):
      base_path = target.getAttribute('path')
      for entry in target.getElementsByTagName('entry'):
        file = entry.getAttribute('path')
        wc_status = entry.getElementsByTagName('wc-status')
        assert len(wc_status) == 1
        # Emulate svn 1.5 status ouput...
        statuses = [' ' for i in range(7)]
        # Col 0
        xml_item_status = wc_status[0].getAttribute('item')
        if xml_item_status in status_letter:
          statuses[0] = status_letter[xml_item_status]
        else:
          raise Exception('Unknown item status "%s"; please implement me!' %
                          xml_item_status)
        # Col 1
        xml_props_status = wc_status[0].getAttribute('props')
        if xml_props_status == 'modified':
          statuses[1] = 'M'
        elif xml_props_status == 'conflicted':
          statuses[1] = 'C'
        elif (not xml_props_status or xml_props_status == 'none' or
              xml_props_status == 'normal'):
          pass
        else:
          raise Exception('Unknown props status "%s"; please implement me!' %
                          xml_props_status)
        # Col 2
        if wc_status[0].getAttribute('wc-locked') == 'true':
          statuses[2] = 'L'
        # Col 3
        if wc_status[0].getAttribute('copied') == 'true':
          statuses[3] = '+'
        item = (''.join(statuses), file)
        results.append(item)
  return results