Copy git tooling into testing_support.

testing_support/git/repo.py

+# Copyright 2014 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.
+
+import atexit
+import collections
+import datetime
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+
+
+from testing_support.git.schema import GitRepoSchema
+
+
+class GitRepo(object):
+  """Creates a real git repo for a GitRepoSchema.
+
+  Obtains schema and content information from the GitRepoSchema.
+
+  The format for the commit data supplied by GitRepoSchema.data_for is:
+    {
+      SPECIAL_KEY: special_value,
+      ...
+      "path/to/some/file": { 'data': "some data content for this file",
+                              'mode': 0755 },
+      ...
+    }
+
+  The SPECIAL_KEYs are the following attribues of the GitRepo class:
+    * AUTHOR_NAME
+    * AUTHOR_EMAIL
+    * AUTHOR_DATE - must be a datetime.datetime instance
+    * COMMITTER_NAME
+    * COMMITTER_EMAIL
+    * COMMITTER_DATE - must be a datetime.datetime instance
+
+  For file content, if 'data' is None, then this commit will `git rm` that file.
+  """
+  BASE_TEMP_DIR = tempfile.mkdtemp(suffix='base', prefix='git_repo')
+  atexit.register(shutil.rmtree, BASE_TEMP_DIR)
+
+  # Singleton objects to specify specific data in a commit dictionary.
+  AUTHOR_NAME = object()
+  AUTHOR_EMAIL = object()
+  AUTHOR_DATE = object()
+  COMMITTER_NAME = object()
+  COMMITTER_EMAIL = object()
+  COMMITTER_DATE = object()
+
+  DEFAULT_AUTHOR_NAME = 'Author McAuthorly'
+  DEFAULT_AUTHOR_EMAIL = 'author@example.com'
+  DEFAULT_COMMITTER_NAME = 'Charles Committish'
+  DEFAULT_COMMITTER_EMAIL = 'commitish@example.com'
+
+  COMMAND_OUTPUT = collections.namedtuple('COMMAND_OUTPUT', 'retcode stdout')
+
+  def __init__(self, schema):
+    """Makes new GitRepo.
+
+    Automatically creates a temp folder under GitRepo.BASE_TEMP_DIR. It's
+    recommended that you clean this repo up by calling nuke() on it, but if not,
+    GitRepo will automatically clean up all allocated repos at the exit of the
+    program (assuming a normal exit like with sys.exit)
+
+    Args:
+      schema - An instance of GitRepoSchema
+    """
+    self.repo_path = tempfile.mkdtemp(dir=self.BASE_TEMP_DIR)
+    self.commit_map = {}
+    self._date = datetime.datetime(1970, 1, 1)
+
+    self.to_schema_refs = ['--branches']
+
+    self.git('init')
+    self.git('config', 'user.name', 'testcase')
+    self.git('config', 'user.email', 'testcase@example.com')
+    for commit in schema.walk():
+      self._add_schema_commit(commit, schema.data_for(commit.name))
+      self.last_commit = self[commit.name]
+    if schema.master:
+      self.git('update-ref', 'refs/heads/master', self[schema.master])
+
+  def __getitem__(self, commit_name):
+    """Gets the hash of a commit by its schema name.
+
+    >>> r = GitRepo(GitRepoSchema('A B C'))
+    >>> r['B']
+    '7381febe1da03b09da47f009963ab7998a974935'
+    """
+    return self.commit_map[commit_name]
+
+  def _add_schema_commit(self, commit, commit_data):
+    commit_data = commit_data or {}
+
+    if commit.parents:
+      parents = list(commit.parents)
+      self.git('checkout', '--detach', '-q', self[parents[0]])
+      if len(parents) > 1:
+        self.git('merge', '--no-commit', '-q', *[self[x] for x in parents[1:]])
+    else:
+      self.git('checkout', '--orphan', 'root_%s' % commit.name)
+      self.git('rm', '-rf', '.')
+
+    env = self.get_git_commit_env(commit_data)
+
+    for fname, file_data in commit_data.iteritems():
+      deleted = False
+      if 'data' in file_data:
+        data = file_data.get('data')
+        if data is None:
+          deleted = True
+          self.git('rm', fname)
+        else:
+          path = os.path.join(self.repo_path, fname)
+          pardir = os.path.dirname(path)
+          if not os.path.exists(pardir):
+            os.makedirs(pardir)
+          with open(path, 'wb') as f:
+            f.write(data)
+
+      mode = file_data.get('mode')
+      if mode and not deleted:
+        os.chmod(path, mode)
+
+      self.git('add', fname)
+
+    rslt = self.git('commit', '--allow-empty', '-m', commit.name, env=env)
+    assert rslt.retcode == 0, 'Failed to commit %s' % str(commit)
+    self.commit_map[commit.name] = self.git('rev-parse', 'HEAD').stdout.strip()
+    self.git('tag', 'tag_%s' % commit.name, self[commit.name])
+    if commit.is_branch:
+      self.git('branch', '-f', 'branch_%s' % commit.name, self[commit.name])
+
+  def get_git_commit_env(self, commit_data=None):
+    commit_data = commit_data or {}
+    env = {}
+    for prefix in ('AUTHOR', 'COMMITTER'):
+      for suffix in ('NAME', 'EMAIL', 'DATE'):
+        singleton = '%s_%s' % (prefix, suffix)
+        key = getattr(self, singleton)
+        if key in commit_data:
+          val = commit_data[key]
+        else:
+          if suffix == 'DATE':
+            val = self._date
+            self._date += datetime.timedelta(days=1)
+          else:
+            val = getattr(self, 'DEFAULT_%s' % singleton)
+        env['GIT_%s' % singleton] = str(val)
+    return env
+
+
+  def git(self, *args, **kwargs):
+    """Runs a git command specified by |args| in this repo."""
+    assert self.repo_path is not None
+    try:
+      with open(os.devnull, 'wb') as devnull:
+        output = subprocess.check_output(
+          ('git',) + args, cwd=self.repo_path, stderr=devnull, **kwargs)
+      return self.COMMAND_OUTPUT(0, output)
+    except subprocess.CalledProcessError as e:
+      return self.COMMAND_OUTPUT(e.returncode, e.output)
+
+  def git_commit(self, message):
+    return self.git('commit', '-am', message, env=self.get_git_commit_env())
+
+  def nuke(self):
+    """Obliterates the git repo on disk.
+
+    Causes this GitRepo to be unusable.
+    """
+    shutil.rmtree(self.repo_path)
+    self.repo_path = None
+
+  def run(self, fn, *args, **kwargs):
+    """Run a python function with the given args and kwargs with the cwd set to
+    the git repo."""
+    assert self.repo_path is not None
+    curdir = os.getcwd()
+    try:
+      os.chdir(self.repo_path)
+      return fn(*args, **kwargs)
+    finally:
+      os.chdir(curdir)
+
+  def capture_stdio(self, fn, *args, **kwargs):
+    """Run a python function with the given args and kwargs with the cwd set to
+    the git repo.
+
+    Returns the (stdout, stderr) of whatever ran, instead of the what |fn|
+    returned.
+    """
+    stdout = sys.stdout
+    stderr = sys.stderr
+    try:
+      # "multiple statements on a line" pylint: disable=C0321
+      with tempfile.TemporaryFile() as out, tempfile.TemporaryFile() as err:
+        sys.stdout = out
+        sys.stderr = err
+        try:
+          self.run(fn, *args, **kwargs)
+        except SystemExit:
+          pass
+        out.seek(0)
+        err.seek(0)
+        return out.read(), err.read()
+    finally:
+      sys.stdout = stdout
+      sys.stderr = stderr
+
+  def open(self, path, mode='rb'):
+    return open(os.path.join(self.repo_path, path), mode)
+
+  def to_schema(self):
+    lines = self.git('rev-list', '--parents', '--reverse', '--topo-order',
+                     '--format=%s', *self.to_schema_refs).stdout.splitlines()
+    hash_to_msg = {}
+    ret = GitRepoSchema()
+    current = None
+    parents = []
+    for line in lines:
+      if line.startswith('commit'):
+        assert current is None
+        tokens = line.split()
+        current, parents = tokens[1], tokens[2:]
+        assert all(p in hash_to_msg for p in parents)
+      else:
+        assert current is not None
+        hash_to_msg[current] = line
+        ret.add_partial(line)
+        for parent in parents:
+          ret.add_partial(line, hash_to_msg[parent])
+        current = None
+        parents = []
+    assert current is None
+    return ret