blob: 9155446b1eeca25f2e5a1e8f10d3c17863ee0bb8 [file] [log] [blame]
# coding=utf-8
# 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.
"""Manages a project checkout.
Includes support only for git.
from __future__ import print_function
import fnmatch
import logging
import os
import re
import shutil
import subprocess
import sys
import tempfile
# The configparser module was renamed in Python 3.
import configparser
except ImportError:
import ConfigParser as configparser
import patch
import scm
import subprocess2
if sys.platform in ('cygwin', 'win32'):
# Disable timeouts on Windows since we can't have shells with timeouts.
# Default timeout of 15 minutes.
# Use a larger timeout for checkout since it can be a genuinely slower
# operation.
def get_code_review_setting(path, key,
"""Parses codereview.settings and return the value for the key if present.
Don't cache the values in case the file is changed."""
# TODO(maruel): Do not duplicate code.
settings = {}
settings_file = open(os.path.join(path, codereview_settings_file), 'r')
for line in settings_file.readlines():
if not line or line.startswith('#'):
if not ':' in line:
# Invalid file.
return None
k, v = line.split(':', 1)
settings[k.strip()] = v.strip()
except IOError:
return None
return settings.get(key, None)
def align_stdout(stdout):
"""Returns the aligned output of multiple stdouts."""
output = ''
for item in stdout:
item = item.strip()
if not item:
output += ''.join(' %s\n' % line for line in item.splitlines())
return output
class PatchApplicationFailed(Exception):
"""Patch failed to be applied."""
def __init__(self, errors, verbose):
super(PatchApplicationFailed, self).__init__(errors, verbose)
self.errors = errors
self.verbose = verbose
def __str__(self):
out = []
for e in self.errors:
p, status = e
if p and p.filename:
out.append('Failed to apply patch for %s:' % p.filename)
if status:
if p and self.verbose:
out.append('Patch: %s' % p.dump())
return '\n'.join(out)
class CheckoutBase(object):
# Set to None to have verbose output.
VOID = subprocess2.VOID
def __init__(self, root_dir, project_name, post_processors):
post_processor: list of lambda(checkout, patches) to call on each of the
modified files.
super(CheckoutBase, self).__init__()
self.root_dir = root_dir
self.project_name = project_name
if self.project_name is None:
self.project_path = self.root_dir
self.project_path = os.path.join(self.root_dir, self.project_name)
# Only used for logging purposes.
self._last_seen_revision = None
self.post_processors = post_processors
assert self.root_dir
assert self.project_path
assert os.path.isabs(self.project_path)
def get_settings(self, key):
return get_code_review_setting(self.project_path, key)
def prepare(self, revision):
"""Checks out a clean copy of the tree and removes any local modification.
This function shouldn't throw unless the remote repository is inaccessible,
there is no free disk space or hard issues like that.
revision: The revision it should sync to, SCM specific.
raise NotImplementedError()
def apply_patch(self, patches, post_processors=None, verbose=False):
"""Applies a patch and returns the list of modified files.
This function should throw patch.UnsupportedPatchFormat or
PatchApplicationFailed when relevant.
patches: patch.PatchSet object.
raise NotImplementedError()
def commit(self, commit_message, user):
"""Commits the patch upstream, while impersonating 'user'."""
raise NotImplementedError()
def revisions(self, rev1, rev2):
"""Returns the count of revisions from rev1 to rev2, e.g. len(]rev1, rev2]).
If rev2 is None, it means 'HEAD'.
Returns None if there is no link between the two.
raise NotImplementedError()
class GitCheckout(CheckoutBase):
"""Manages a git checkout."""
def __init__(self, root_dir, project_name, remote_branch, git_url,
commit_user, post_processors=None):
super(GitCheckout, self).__init__(root_dir, project_name, post_processors)
self.git_url = git_url
self.commit_user = commit_user
self.remote_branch = remote_branch
# The working branch where patches will be applied. It will track the
# remote branch.
self.working_branch = 'working_branch'
# There is no reason to not hardcode origin.
self.remote = 'origin'
# There is no reason to not hardcode master.
self.master_branch = 'master'
def prepare(self, revision):
"""Resets the git repository in a clean state.
Checks it out if not present and deletes the working branch.
assert self.remote_branch
assert self.git_url
if not os.path.isdir(self.project_path):
# Clone the repo if the directory is not present.
'Checking out %s in %s', self.project_name, self.project_path)
['clone', self.git_url, '-b', self.remote_branch, self.project_path],
cwd=None, timeout=FETCH_TIMEOUT)
# Throw away all uncommitted changes in the existing checkout.
self._check_call_git(['checkout', self.remote_branch])
['reset', '--hard', '--quiet',
'%s/%s' % (self.remote, self.remote_branch)])
if revision:
# Look if the commit hash already exist. If so, we can skip a
# 'git fetch' call.
revision = self._check_output_git(['rev-parse', revision]).rstrip()
except subprocess.CalledProcessError:
['fetch', self.remote, self.remote_branch, '--quiet'])
revision = self._check_output_git(['rev-parse', revision]).rstrip()
self._check_call_git(['checkout', '--force', '--quiet', revision])
branches, active = self._branches()
if active != self.master_branch:
['checkout', '--force', '--quiet', self.master_branch])
if self.working_branch in branches:
self._call_git(['branch', '-D', self.working_branch])
return self._get_head_commit_hash()
def _sync_remote_branch(self):
"""Syncs the remote branch."""
# We do a 'git pull origin master:refs/remotes/origin/master' instead of
# 'git pull origin master' because from the manpage for git-pull:
# A parameter <ref> without a colon is equivalent to <ref>: when
# pulling/fetching, so it merges <ref> into the current branch without
# storing the remote branch anywhere locally.
remote_tracked_path = 'refs/remotes/%s/%s' % (
self.remote, self.remote_branch)
['pull', self.remote,
'%s:%s' % (self.remote_branch, remote_tracked_path),
def _get_head_commit_hash(self):
"""Gets the current revision (in unicode) from the local branch."""
return unicode(self._check_output_git(['rev-parse', 'HEAD']).strip())
def apply_patch(self, patches, post_processors=None, verbose=False):
"""Applies a patch on 'working_branch' and switches to it.
The changes remain staged on the current branch.
post_processors = post_processors or self.post_processors or []
# It this throws, the checkout is corrupted. Maybe worth deleting it and
# trying again?
if self.remote_branch:
['checkout', '-b', self.working_branch, '-t', self.remote_branch,
errors = []
for index, p in enumerate(patches):
stdout = []
filepath = os.path.join(self.project_path, p.filename)
if p.is_delete:
if (not os.path.exists(filepath) and
any(p1.source_filename == p.filename for p1 in patches[0:index])):
# The file was already deleted if a prior patch with file rename
# was already processed because 'git apply' did it for us.
stdout.append(self._check_output_git(['rm', p.filename]))
assert(not os.path.exists(filepath))
dirname = os.path.dirname(p.filename)
full_dir = os.path.join(self.project_path, dirname)
if dirname and not os.path.isdir(full_dir):
stdout.append('Created missing directory %s.' % dirname)
if p.is_binary:
content = p.get()
with open(filepath, 'wb') as f:
stdout.append('Added binary file %d bytes' % len(content))
cmd = ['add', p.filename]
if verbose:
# No need to do anything special with p.is_new or if not
# p.diff_hunks. git apply manages all that already.
cmd = ['apply', '--index', '-3', '-p%s' % p.patchlevel]
if verbose:
stdout.append(self._check_output_git(cmd, stdin=p.get(True)))
for post in post_processors:
post(self, p)
if verbose:
except OSError as e:
errors.append((p, '%s%s' % (align_stdout(stdout), e)))
except subprocess.CalledProcessError as e:
'While running %s;\n%s%s' % (
' '.join(e.cmd),
align_stdout([getattr(e, 'stdout', '')]))))
if errors:
raise PatchApplicationFailed(errors, verbose)
found_files = self._check_output_git(
['-c', 'core.quotePath=false', 'diff', '--ignore-submodules',
'--name-only', '--staged']).splitlines(False)
if sorted(patches.filenames) != sorted(found_files):
extra_files = sorted(set(found_files) - set(patches.filenames))
unpatched_files = sorted(set(patches.filenames) - set(found_files))
if extra_files:
print('Found extra files: %r' % extra_files)
if unpatched_files:
print('Found unpatched files: %r' % unpatched_files)
def commit(self, commit_message, user):
"""Commits, updates the commit message and pushes."""
# TODO(hinoka): CQ no longer uses this, I think its deprecated.
# Delete this.
assert self.commit_user
assert isinstance(commit_message, unicode)
current_branch = self._check_output_git(
['rev-parse', '--abbrev-ref', 'HEAD']).strip()
assert current_branch == self.working_branch
commit_cmd = ['commit', '-m', commit_message]
if user and user != self.commit_user:
# We do not have the first or last name of the user, grab the username
# from the email and call it the original author's name.
# TODO(rmistry): Do not need the below if user is already in
# "Name <email>" format.
name = user.split('@')[0]
commit_cmd.extend(['--author', '%s <%s>' % (name, user)])
# Push to the remote repository.
['push', 'origin', '%s:%s' % (self.working_branch, self.remote_branch),
# Get the revision after the push.
revision = self._get_head_commit_hash()
# Switch back to the remote_branch and sync it.
self._check_call_git(['checkout', self.remote_branch])
# Delete the working branch since we are done with it.
self._check_call_git(['branch', '-D', self.working_branch])
return revision
def _check_call_git(self, args, **kwargs):
kwargs.setdefault('cwd', self.project_path)
kwargs.setdefault('stdout', self.VOID)
kwargs.setdefault('timeout', GLOBAL_TIMEOUT)
return subprocess2.check_call_out(['git'] + args, **kwargs)
def _call_git(self, args, **kwargs):
"""Like check_call but doesn't throw on failure."""
kwargs.setdefault('cwd', self.project_path)
kwargs.setdefault('stdout', self.VOID)
kwargs.setdefault('timeout', GLOBAL_TIMEOUT)
return['git'] + args, **kwargs)
def _check_output_git(self, args, **kwargs):
kwargs.setdefault('cwd', self.project_path)
kwargs.setdefault('timeout', GLOBAL_TIMEOUT)
return subprocess2.check_output(
['git'] + args, stderr=subprocess2.STDOUT, **kwargs)
def _branches(self):
"""Returns the list of branches and the active one."""
out = self._check_output_git(['branch']).splitlines(False)
branches = [l[2:] for l in out]
active = None
for l in out:
if l.startswith('*'):
active = l[2:]
return branches, active
def revisions(self, rev1, rev2):
"""Returns the number of actual commits between both hash."""
rev2 = rev2 or '%s/%s' % (self.remote, self.remote_branch)
# Revision range is ]rev1, rev2] and ordering matters.
out = self._check_output_git(
['log', '--format="%H"' , '%s..%s' % (rev1, rev2)])
except subprocess.CalledProcessError:
return None
return len(out.splitlines())
def _fetch_remote(self):
"""Fetches the remote without rebasing."""
# git fetch is always verbose even with -q, so redirect its output.
self._check_output_git(['fetch', self.remote, self.remote_branch],
class ReadOnlyCheckout(object):
"""Converts a checkout into a read-only one."""
def __init__(self, checkout, post_processors=None):
super(ReadOnlyCheckout, self).__init__()
self.checkout = checkout
self.post_processors = (post_processors or []) + (
self.checkout.post_processors or [])
def prepare(self, revision):
return self.checkout.prepare(revision)
def get_settings(self, key):
return self.checkout.get_settings(key)
def apply_patch(self, patches, post_processors=None, verbose=False):
return self.checkout.apply_patch(
patches, post_processors or self.post_processors, verbose)
def commit(self, message, user): # pylint: disable=no-self-use'Would have committed for %s with message: %s' % (
user, message))
return 'FAKE'
def revisions(self, rev1, rev2):
return self.checkout.revisions(rev1, rev2)
def project_name(self):
return self.checkout.project_name
def project_path(self):
return self.checkout.project_path