Move commit-queue/checkout into depot_tools so it can be reused by the try server.

BUG=
TEST=

Review URL: http://codereview.chromium.org/6877055

git-svn-id: svn://svn.chromium.org/chrome/trunk/tools/depot_tools@82414 0039d316-1c4b-4281-b951-d872f2087c98
experimental/szager/collated-output
maruel@chromium.org 15 years ago
parent 39f645ff57
commit dfaecd2bd4

5
.gitignore vendored

@ -1,5 +1,4 @@
*.pyc
/tests/_rietveld
/python
/python.bat
/python_bin
@ -7,3 +6,7 @@
/svn.bat
/svn_bin
/svnversion.bat
/tests/_rietveld
/tests/subversion_config/README.txt
/tests/subversion_config/auth
/tests/subversion_config/servers

@ -0,0 +1,625 @@
# coding=utf8
# Copyright (c) 2011 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 for svn, git-svn and git.
"""
from __future__ import with_statement
import ConfigParser
import fnmatch
import logging
import os
import re
import subprocess
import sys
import tempfile
import patch
import scm
import subprocess2
def get_code_review_setting(path, key,
codereview_settings_file='codereview.settings'):
"""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 = {}
try:
settings_file = open(os.path.join(path, codereview_settings_file), 'r')
try:
for line in settings_file.readlines():
if not line or line.startswith('#'):
continue
if not ':' in line:
# Invalid file.
return None
k, v = line.split(':', 1)
settings[k.strip()] = v.strip()
finally:
settings_file.close()
except OSError:
return None
return settings.get(key, None)
class PatchApplicationFailed(Exception):
"""Patch failed to be applied."""
def __init__(self, filename, status):
super(PatchApplicationFailed, self).__init__(filename, status)
self.filename = filename
self.status = status
class CheckoutBase(object):
# Set to None to have verbose output.
VOID = subprocess2.VOID
def __init__(self, root_dir, project_name):
self.root_dir = root_dir
self.project_name = project_name
self.project_path = os.path.join(self.root_dir, self.project_name)
# Only used for logging purposes.
self._last_seen_revision = None
assert self.root_dir
assert self.project_name
assert self.project_path
def get_settings(self, key):
return get_code_review_setting(self.project_path, key)
def prepare(self):
"""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.
"""
raise NotImplementedError()
def apply_patch(self, patches):
"""Applies a patch and returns the list of modified files.
This function should throw patch.UnsupportedPatchFormat or
PatchApplicationFailed when relevant.
"""
raise NotImplementedError()
def commit(self, commit_message, user):
"""Commits the patch upstream, while impersonating 'user'."""
raise NotImplementedError()
class RawCheckout(CheckoutBase):
"""Used to apply a patch locally without any intent to commit it.
To be used by the try server.
"""
def prepare(self):
"""Stubbed out."""
pass
def apply_patch(self, patches):
for p in patches:
try:
stdout = ''
filename = os.path.join(self.project_path, p.filename)
if p.is_delete:
os.remove(filename)
else:
dirname = os.path.dirname(p.filename)
full_dir = os.path.join(self.project_path, dirname)
if dirname and not os.path.isdir(full_dir):
os.makedirs(full_dir)
if p.is_binary:
with open(os.path.join(filename), 'wb') as f:
f.write(p.get())
else:
stdout = subprocess2.check_output(
['patch', '-p%s' % p.patchlevel],
stdin=p.get(),
cwd=self.project_path)
# Ignore p.svn_properties.
except OSError, e:
raise PatchApplicationFailed(p.filename, '%s%s' % (stdout, e))
except subprocess.CalledProcessError, e:
raise PatchApplicationFailed(
p.filename, '%s%s' % (stdout, getattr(e, 'stdout', None)))
def commit(self, commit_message, user):
"""Stubbed out."""
raise NotImplementedError('RawCheckout can\'t commit')
class SvnConfig(object):
"""Parses a svn configuration file."""
def __init__(self, svn_config_dir=None):
self.svn_config_dir = svn_config_dir
self.default = not bool(self.svn_config_dir)
if not self.svn_config_dir:
if sys.platform == 'win32':
self.svn_config_dir = os.path.join(os.environ['APPDATA'], 'Subversion')
else:
self.svn_config_dir = os.path.join(os.environ['HOME'], '.subversion')
svn_config_file = os.path.join(self.svn_config_dir, 'config')
parser = ConfigParser.SafeConfigParser()
if os.path.isfile(svn_config_file):
parser.read(svn_config_file)
else:
parser.add_section('auto-props')
self.auto_props = dict(parser.items('auto-props'))
class SvnMixIn(object):
"""MixIn class to add svn commands common to both svn and git-svn clients."""
# These members need to be set by the subclass.
commit_user = None
commit_pwd = None
svn_url = None
project_path = None
# Override at class level when necessary. If used, --non-interactive is
# implied.
svn_config = SvnConfig()
# Set to True when non-interactivity is necessary but a custom subversion
# configuration directory is not necessary.
non_interactive = False
def _add_svn_flags(self, args, non_interactive):
args = ['svn'] + args
if not self.svn_config.default:
args.extend(['--config-dir', self.svn_config.svn_config_dir])
if not self.svn_config.default or self.non_interactive or non_interactive:
args.append('--non-interactive')
if self.commit_user:
args.extend(['--username', self.commit_user])
if self.commit_pwd:
args.extend(['--password', self.commit_pwd])
return args
def _check_call_svn(self, args, **kwargs):
"""Runs svn and throws an exception if the command failed."""
kwargs.setdefault('cwd', self.project_path)
kwargs.setdefault('stdout', self.VOID)
return subprocess2.check_call(self._add_svn_flags(args, False), **kwargs)
def _check_output_svn(self, args, **kwargs):
"""Runs svn and throws an exception if the command failed.
Returns the output.
"""
kwargs.setdefault('cwd', self.project_path)
return subprocess2.check_output(self._add_svn_flags(args, True), **kwargs)
@staticmethod
def _parse_svn_info(output, key):
"""Returns value for key from svn info output.
Case insensitive.
"""
values = {}
key = key.lower()
for line in output.splitlines(False):
if not line:
continue
k, v = line.split(':', 1)
k = k.strip().lower()
v = v.strip()
assert not k in values
values[k] = v
return values.get(key, None)
class SvnCheckout(CheckoutBase, SvnMixIn):
"""Manages a subversion checkout."""
def __init__(self, root_dir, project_name, commit_user, commit_pwd, svn_url):
super(SvnCheckout, self).__init__(root_dir, project_name)
self.commit_user = commit_user
self.commit_pwd = commit_pwd
self.svn_url = svn_url
assert bool(self.commit_user) >= bool(self.commit_pwd)
assert self.svn_url
def prepare(self):
"""Creates the initial checkouts for the repo."""
# Will checkout if the directory is not present.
if not os.path.isdir(self.project_path):
logging.info('Checking out %s in %s' %
(self.project_name, self.project_path))
revision = self._revert()
if revision != self._last_seen_revision:
logging.info('Updated at revision %d' % revision)
self._last_seen_revision = revision
return revision
def apply_patch(self, patches):
"""Applies a patch."""
for p in patches:
try:
stdout = ''
if p.is_delete:
stdout += self._check_output_svn(['delete', p.filename, '--force'])
else:
new = not os.path.exists(p.filename)
# svn add while creating directories otherwise svn add on the
# contained files will silently fail.
# First, find the root directory that exists.
dirname = os.path.dirname(p.filename)
dirs_to_create = []
while (dirname and
not os.path.isdir(os.path.join(self.project_path, dirname))):
dirs_to_create.append(dirname)
dirname = os.path.dirname(dirname)
for dir_to_create in reversed(dirs_to_create):
os.mkdir(os.path.join(self.project_path, dir_to_create))
stdout += self._check_output_svn(
['add', dir_to_create, '--force'])
if p.is_binary:
with open(os.path.join(self.project_path, p.filename), 'wb') as f:
f.write(p.get())
else:
cmd = ['patch', '-p%s' % p.patchlevel, '--forward', '--force']
stdout += subprocess2.check_output(
cmd, stdin=p.get(), cwd=self.project_path)
if new:
stdout += self._check_output_svn(['add', p.filename, '--force'])
for prop in p.svn_properties:
stdout += self._check_output_svn(
['propset', prop[0], prop[1], p.filename])
for prop, value in self.svn_config.auto_props.iteritems():
if fnmatch.fnmatch(p.filename, prop):
stdout += self._check_output_svn(
['propset'] + value.split('=', 1) + [p.filename])
except OSError, e:
raise PatchApplicationFailed(p.filename, '%s%s' % (stdout, e))
except subprocess.CalledProcessError, e:
raise PatchApplicationFailed(
p.filename, '%s%s' % (stdout, getattr(e, 'stdout', '')))
def commit(self, commit_message, user):
logging.info('Committing patch for %s' % user)
assert self.commit_user
handle, commit_filename = tempfile.mkstemp(text=True)
try:
os.write(handle, commit_message)
os.close(handle)
# When committing, svn won't update the Revision metadata of the checkout,
# so if svn commit returns "Committed revision 3.", svn info will still
# return "Revision: 2". Since running svn update right after svn commit
# creates a race condition with other committers, this code _must_ parse
# the output of svn commit and use a regexp to grab the revision number.
# Note that "Committed revision N." is localized but subprocess2 forces
# LANGUAGE=en.
args = ['commit', '--file', commit_filename]
# realauthor is parsed by a server-side hook.
if user and user != self.commit_user:
args.extend(['--with-revprop', 'realauthor=%s' % user])
out = self._check_output_svn(args)
finally:
os.remove(commit_filename)
lines = filter(None, out.splitlines())
match = re.match(r'^Committed revision (\d+).$', lines[-1])
if not match:
raise PatchApplicationFailed(
None,
'Couldn\'t make sense out of svn commit message:\n' + out)
return int(match.group(1))
def _revert(self):
"""Reverts local modifications or checks out if the directory is not
present. Use depot_tools's functionality to do this.
"""
flags = ['--ignore-externals']
if not os.path.isdir(self.project_path):
logging.info(
'Directory %s is not present, checking it out.' % self.project_path)
self._check_call_svn(
['checkout', self.svn_url, self.project_path] + flags, cwd=None)
else:
scm.SVN.Revert(self.project_path)
# Revive files that were deleted in scm.SVN.Revert().
self._check_call_svn(['update', '--force'] + flags)
out = self._check_output_svn(['info', '.'])
return int(self._parse_svn_info(out, 'revision'))
class GitCheckoutBase(CheckoutBase):
"""Base class for git checkout. Not to be used as-is."""
def __init__(self, root_dir, project_name, remote_branch):
super(GitCheckoutBase, self).__init__(root_dir, project_name)
# There is no reason to not hardcode it.
self.remote = 'origin'
self.remote_branch = remote_branch
self.working_branch = 'working_branch'
assert self.remote_branch
def prepare(self):
"""Resets the git repository in a clean state.
Checks it out if not present and deletes the working branch.
"""
assert os.path.isdir(self.project_path)
self._check_call_git(['reset', '--hard', '--quiet'])
branches, active = self._branches()
if active != 'master':
self._check_call_git(['checkout', 'master', '--force', '--quiet'])
self._check_call_git(['pull', self.remote, self.remote_branch, '--quiet'])
if self.working_branch in branches:
self._call_git(['branch', '-D', self.working_branch])
def apply_patch(self, patches):
"""Applies a patch on 'working_branch' and switch to it."""
# It this throws, the checkout is corrupted. Maybe worth deleting it and
# trying again?
self._check_call_git(
['checkout', '-b', self.working_branch,
'%s/%s' % (self.remote, self.remote_branch)])
for p in patches:
try:
stdout = ''
if p.is_delete:
stdout += self._check_output_git(['rm', p.filename])
else:
dirname = os.path.dirname(p.filename)
full_dir = os.path.join(self.project_path, dirname)
if dirname and not os.path.isdir(full_dir):
os.makedirs(full_dir)
if p.is_binary:
with open(os.path.join(self.project_path, p.filename), 'wb') as f:
f.write(p.get())
stdout += self._check_output_git(['add', p.filename])
else:
stdout += self._check_output_git(
['apply', '--index', '-p%s' % p.patchlevel], stdin=p.get())
for prop in p.svn_properties:
# Ignore some known auto-props flags through .subversion/config,
# bails out on the other ones.
# TODO(maruel): Read ~/.subversion/config and detect the rules that
# applies here to figure out if the property will be correctly
# handled.
if not prop[0] in ('svn:eol-style', 'svn:executable'):
raise patch.UnsupportedPatchFormat(
p.filename,
'Cannot apply svn property %s to file %s.' % (
prop[0], p.filename))
except OSError, e:
raise PatchApplicationFailed(p.filename, '%s%s' % (stdout, e))
except subprocess.CalledProcessError, e:
raise PatchApplicationFailed(
p.filename, '%s%s' % (stdout, getattr(e, 'stdout', None)))
# Once all the patches are processed and added to the index, commit the
# index.
self._check_call_git(['commit', '-m', 'Committed patch'])
# TODO(maruel): Weirdly enough they don't match, need to investigate.
#found_files = self._check_output_git(
# ['diff', 'master', '--name-only']).splitlines(False)
#assert sorted(patches.filenames) == sorted(found_files), (
# sorted(out), sorted(found_files))
def commit(self, commit_message, user):
"""Updates the commit message.
Subclass needs to dcommit or push.
"""
self._check_call_git(['commit', '--amend', '-m', commit_message])
return self._check_output_git(['rev-parse', 'HEAD']).strip()
def _check_call_git(self, args, **kwargs):
kwargs.setdefault('cwd', self.project_path)
kwargs.setdefault('stdout', self.VOID)
return subprocess2.check_call(['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)
return subprocess2.call(['git'] + args, **kwargs)
def _check_output_git(self, args, **kwargs):
kwargs.setdefault('cwd', self.project_path)
return subprocess2.check_output(['git'] + args, **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:]
break
return branches, active
class GitSvnCheckoutBase(GitCheckoutBase, SvnMixIn):
"""Base class for git-svn checkout. Not to be used as-is."""
def __init__(self,
root_dir, project_name, remote_branch,
commit_user, commit_pwd,
svn_url, trunk):
"""trunk is optional."""
super(GitSvnCheckoutBase, self).__init__(
root_dir, project_name + '.git', remote_branch)
self.commit_user = commit_user
self.commit_pwd = commit_pwd
# svn_url in this case is the root of the svn repository.
self.svn_url = svn_url
self.trunk = trunk
assert bool(self.commit_user) >= bool(self.commit_pwd)
assert self.svn_url
assert self.trunk
self._cache_svn_auth()
def prepare(self):
"""Resets the git repository in a clean state."""
self._check_call_git(['reset', '--hard', '--quiet'])
branches, active = self._branches()
if active != 'master':
if not 'master' in branches:
self._check_call_git(
['checkout', '--quiet', '-b', 'master',
'%s/%s' % (self.remote, self.remote_branch)])
else:
self._check_call_git(['checkout', 'master', '--force', '--quiet'])
# git svn rebase --quiet --quiet doesn't work, use two steps to silence it.
self._check_call_git_svn(['fetch', '--quiet', '--quiet'])
self._check_call_git(
['rebase', '--quiet', '--quiet',
'%s/%s' % (self.remote, self.remote_branch)])
if self.working_branch in branches:
self._call_git(['branch', '-D', self.working_branch])
return int(self._git_svn_info('revision'))
def _git_svn_info(self, key):
"""Calls git svn info. This doesn't support nor need --config-dir."""
return self._parse_svn_info(self._check_output_git(['svn', 'info']), key)
def commit(self, commit_message, user):
"""Commits a patch."""
logging.info('Committing patch for %s' % user)
# Fix the commit message and author. It returns the git hash, which we
# ignore unless it's None.
if not super(GitSvnCheckoutBase, self).commit(commit_message, user):
return None
# TODO(maruel): git-svn ignores --config-dir as of git-svn version 1.7.4 and
# doesn't support --with-revprop.
# Either learn perl and upstream or suck it.
kwargs = {}
if self.commit_pwd:
kwargs['stdin'] = self.commit_pwd + '\n'
self._check_call_git_svn(
['dcommit', '--rmdir', '--find-copies-harder',
'--username', self.commit_user],
**kwargs)
revision = int(self._git_svn_info('revision'))
return revision
def _cache_svn_auth(self):
"""Caches the svn credentials. It is necessary since git-svn doesn't prompt
for it."""
if not self.commit_user or not self.commit_pwd:
return
# Use capture to lower noise in logs.
self._check_output_svn(['ls', self.svn_url], cwd=None)
def _check_call_git_svn(self, args, **kwargs):
"""Handles svn authentication while calling git svn."""
args = ['svn'] + args
if not self.svn_config.default:
args.extend(['--config-dir', self.svn_config.svn_config_dir])
return self._check_call_git(args, **kwargs)
def _get_revision(self):
revision = int(self._git_svn_info('revision'))
if revision != self._last_seen_revision:
logging.info('Updated at revision %d' % revision)
self._last_seen_revision = revision
return revision
class GitSvnPremadeCheckout(GitSvnCheckoutBase):
"""Manages a git-svn clone made out from an initial git-svn seed.
This class is very similar to GitSvnCheckout but is faster to bootstrap
because it starts right off with an existing git-svn clone.
"""
def __init__(self,
root_dir, project_name, remote_branch,
commit_user, commit_pwd,
svn_url, trunk, git_url):
super(GitSvnPremadeCheckout, self).__init__(
root_dir, project_name, remote_branch,
commit_user, commit_pwd,
svn_url, trunk)
self.git_url = git_url
assert self.git_url
def prepare(self):
"""Creates the initial checkout for the repo."""
if not os.path.isdir(self.project_path):
logging.info('Checking out %s in %s' %
(self.project_name, self.project_path))
assert self.remote == 'origin'
# self.project_path doesn't exist yet.
self._check_call_git(
['clone', self.git_url, self.project_name],
cwd=self.root_dir)
try:
configured_svn_url = self._check_output_git(
['config', 'svn-remote.svn.url']).strip()
except subprocess.CalledProcessError:
configured_svn_url = ''
if configured_svn_url.strip() != self.svn_url:
self._check_call_git_svn(
['init',
'--prefix', self.remote + '/',
'-T', self.trunk,
self.svn_url])
self._check_call_git_svn(['fetch'])
super(GitSvnPremadeCheckout, self).prepare()
return self._get_revision()
class GitSvnCheckout(GitSvnCheckoutBase):
"""Manages a git-svn clone.
Using git-svn hides some of the complexity of using a svn checkout.
"""
def __init__(self,
root_dir, project_name,
commit_user, commit_pwd,
svn_url, trunk):
super(GitSvnCheckout, self).__init__(
root_dir, project_name, 'trunk',
commit_user, commit_pwd,
svn_url, trunk)
def prepare(self):
"""Creates the initial checkout for the repo."""
if not os.path.isdir(self.project_path):
logging.info('Checking out %s in %s' %
(self.project_name, self.project_path))
# TODO: Create a shallow clone.
# self.project_path doesn't exist yet.
self._check_call_git_svn(
['clone',
'--prefix', self.remote + '/',
'-T', self.trunk,
self.svn_url, self.project_path],
cwd=self.root_dir)
super(GitSvnCheckout, self).prepare()
return self._get_revision()
class ReadOnlyCheckout(object):
"""Converts a checkout into a read-only one."""
def __init__(self, checkout):
self.checkout = checkout
def prepare(self):
return self.checkout.prepare()
def get_settings(self, key):
return self.checkout.get_settings(key)
def apply_patch(self, patches):
return self.checkout.apply_patch(patches)
def commit(self, message, user): # pylint: disable=R0201
logging.info('Would have committed for %s with message: %s' % (
user, message))
return 'FAKE'
@property
def project_name(self):
return self.checkout.project_name
@property
def project_path(self):
return self.checkout.project_path

@ -0,0 +1,516 @@
#!/usr/bin/env python
# Copyright (c) 2011 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.
"""Unit tests for checkout.py."""
from __future__ import with_statement
import logging
import os
import shutil
import sys
import unittest
from xml.etree import ElementTree
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
BASE_DIR = os.path.join(ROOT_DIR, '..')
sys.path.insert(0, BASE_DIR)
import checkout
import patch
import subprocess2
from tests import fake_repos
# pass -v to enable it.
DEBUGGING = False
# A naked patch.
NAKED_PATCH = ("""\
--- svn_utils_test.txt
+++ svn_utils_test.txt
@@ -3,6 +3,7 @@ bb
ccc
dd
e
+FOO!
ff
ggg
hh
""")
# A patch generated from git.
GIT_PATCH = ("""\
diff --git a/svn_utils_test.txt b/svn_utils_test.txt
index 0e4de76..8320059 100644
--- a/svn_utils_test.txt
+++ b/svn_utils_test.txt
@@ -3,6 +3,7 @@ bb
ccc
dd
e
+FOO!
ff
ggg
hh
""")
# A patch that will fail to apply.
BAD_PATCH = ("""\
diff --git a/svn_utils_test.txt b/svn_utils_test.txt
index 0e4de76..8320059 100644
--- a/svn_utils_test.txt
+++ b/svn_utils_test.txt
@@ -3,7 +3,8 @@ bb
ccc
dd
+FOO!
ff
ggg
hh
""")
PATCH_ADD = ("""\
diff --git a/new_dir/subdir/new_file b/new_dir/subdir/new_file
new file mode 100644
--- /dev/null
+++ b/new_dir/subdir/new_file
@@ -0,0 +1,2 @@
+A new file
+should exist.
""")
class FakeRepos(fake_repos.FakeReposBase):
def populateSvn(self):
"""Creates a few revisions of changes files."""
subprocess2.check_call(
['svn', 'checkout', self.svn_base, self.svn_checkout, '-q',
'--non-interactive', '--no-auth-cache',
'--username', self.USERS[0][0], '--password', self.USERS[0][1]])
assert os.path.isdir(os.path.join(self.svn_checkout, '.svn'))
fs = {}
fs['trunk/origin'] = 'svn@1'
fs['trunk/codereview.settings'] = (
'# Test data\n'
'bar: pouet\n')
fs['trunk/svn_utils_test.txt'] = (
'a\n'
'bb\n'
'ccc\n'
'dd\n'
'e\n'
'ff\n'
'ggg\n'
'hh\n'
'i\n'
'jj\n'
'kkk\n'
'll\n'
'm\n'
'nn\n'
'ooo\n'
'pp\n'
'q\n')
self._commit_svn(fs)
fs['trunk/origin'] = 'svn@2\n'
fs['trunk/extra'] = 'dummy\n'
fs['trunk/bin_file'] = '\x00'
self._commit_svn(fs)
def populateGit(self):
raise NotImplementedError()
# pylint: disable=R0201
class BaseTest(fake_repos.FakeReposTestBase):
name = 'foo'
FAKE_REPOS_CLASS = FakeRepos
def setUp(self):
# Need to enforce subversion_config first.
checkout.SvnMixIn.svn_config_dir = os.path.join(
ROOT_DIR, 'subversion_config')
super(BaseTest, self).setUp()
self._old_call = subprocess2.call
def redirect_call(args, **kwargs):
if not DEBUGGING:
kwargs.setdefault('stdout', subprocess2.PIPE)
kwargs.setdefault('stderr', subprocess2.STDOUT)
return self._old_call(args, **kwargs)
subprocess2.call = redirect_call
self.usr, self.pwd = self.FAKE_REPOS.USERS[0]
self.previous_log = None
def tearDown(self):
subprocess2.call = self._old_call
super(BaseTest, self).tearDown()
def get_patches(self):
return patch.PatchSet([
patch.FilePatchDiff(
'svn_utils_test.txt', GIT_PATCH, []),
patch.FilePatchBinary('bin_file', '\x00', []),
patch.FilePatchDelete('extra', False),
patch.FilePatchDiff('new_dir/subdir/new_file', PATCH_ADD, []),
])
def get_trunk(self, modified):
tree = {}
subroot = 'trunk/'
for k, v in self.FAKE_REPOS.svn_revs[-1].iteritems():
if k.startswith(subroot):
f = k[len(subroot):]
assert f not in tree
tree[f] = v
if modified:
content_lines = tree['svn_utils_test.txt'].splitlines(True)
tree['svn_utils_test.txt'] = ''.join(
content_lines[0:5] + ['FOO!\n'] + content_lines[5:])
del tree['extra']
tree['new_dir/subdir/new_file'] = 'A new file\nshould exist.\n'
return tree
def _check_base(self, co, root, git, expected):
read_only = isinstance(co, checkout.ReadOnlyCheckout)
assert not read_only == bool(expected)
if not read_only:
self.FAKE_REPOS.svn_dirty = True
self.assertEquals(root, co.project_path)
self.assertEquals(self.previous_log['revision'], co.prepare())
self.assertEquals('pouet', co.get_settings('bar'))
self.assertTree(self.get_trunk(False), root)
patches = self.get_patches()
co.apply_patch(patches)
self.assertEquals(
['bin_file', 'extra', 'new_dir/subdir/new_file', 'svn_utils_test.txt'],
sorted(patches.filenames))
if git:
# Hackish to verify _branches() internal function.
# pylint: disable=W0212
self.assertEquals(
(['master', 'working_branch'], 'working_branch'),
co.checkout._branches())
# Verify that the patch is applied even for read only checkout.
self.assertTree(self.get_trunk(True), root)
fake_author = self.FAKE_REPOS.USERS[1][0]
revision = co.commit('msg', fake_author)
# Nothing changed.
self.assertTree(self.get_trunk(True), root)
if read_only:
self.assertEquals('FAKE', revision)
self.assertEquals(self.previous_log['revision'], co.prepare())
# Changes should be reverted now.
self.assertTree(self.get_trunk(False), root)
expected = self.previous_log
else:
self.assertEquals(self.previous_log['revision'] + 1, revision)
self.assertEquals(self.previous_log['revision'] + 1, co.prepare())
self.assertTree(self.get_trunk(True), root)
expected = expected.copy()
expected['msg'] = 'msg'
expected['revision'] = self.previous_log['revision'] + 1
expected.setdefault('author', fake_author)
actual = self._log()
self.assertEquals(expected, actual)
def _check_exception(self, co, err_msg):
co.prepare()
try:
co.apply_patch([patch.FilePatchDiff('svn_utils_test.txt', BAD_PATCH, [])])
self.fail()
except checkout.PatchApplicationFailed, e:
self.assertEquals(e.filename, 'svn_utils_test.txt')
self.assertEquals(e.status, err_msg)
def _log(self):
raise NotImplementedError()
class SvnBaseTest(BaseTest):
def setUp(self):
super(SvnBaseTest, self).setUp()
self.enabled = self.FAKE_REPOS.set_up_svn()
self.assertTrue(self.enabled)
self.svn_trunk = 'trunk'
self.svn_url = self.svn_base + self.svn_trunk
self.previous_log = self._log()
def _log(self):
# Don't use the local checkout in case of caching incorrency.
out = subprocess2.check_output(
['svn', 'log', self.svn_url,
'--non-interactive', '--no-auth-cache',
'--username', self.usr, '--password', self.pwd,
'--with-all-revprops', '--xml',
'--limit', '1'])
logentry = ElementTree.XML(out).find('logentry')
if logentry == None:
return {'revision': 0}
data = {
'revision': int(logentry.attrib['revision']),
}
def set_item(name):
item = logentry.find(name)
if item != None:
data[name] = item.text
set_item('author')
set_item('msg')
revprops = logentry.find('revprops')
if revprops != None:
data['revprops'] = []
for prop in revprops.getiterator('property'):
data['revprops'].append((prop.attrib['name'], prop.text))
return data
class SvnCheckout(SvnBaseTest):
def _get_co(self, read_only):
if read_only:
return checkout.ReadOnlyCheckout(
checkout.SvnCheckout(
self.root_dir, self.name, None, None, self.svn_url))
else:
return checkout.SvnCheckout(
self.root_dir, self.name, self.usr, self.pwd, self.svn_url)
def _check(self, read_only, expected):
root = os.path.join(self.root_dir, self.name)
self._check_base(self._get_co(read_only), root, False, expected)
def testAllRW(self):
expected = {
'author': self.FAKE_REPOS.USERS[0][0],
'revprops': [('realauthor', self.FAKE_REPOS.USERS[1][0])]
}
self._check(False, expected)
def testAllRO(self):
self._check(True, None)
def testException(self):
self._check_exception(
self._get_co(True),
'patching file svn_utils_test.txt\n'
'Hunk #1 FAILED at 3.\n'
'1 out of 1 hunk FAILED -- saving rejects to file '
'svn_utils_test.txt.rej\n')
def testSvnProps(self):
co = self._get_co(False)
co.prepare()
try:
# svn:ignore can only be applied to directories.
svn_props = [('svn:ignore', 'foo')]
co.apply_patch(
[patch.FilePatchDiff('svn_utils_test.txt', NAKED_PATCH, svn_props)])
self.fail()
except checkout.PatchApplicationFailed, e:
self.assertEquals(e.filename, 'svn_utils_test.txt')
self.assertEquals(
e.status,
"patching file svn_utils_test.txt\n"
"svn: Cannot set 'svn:ignore' on a file ('svn_utils_test.txt')\n")
co.prepare()
svn_props = [('svn:eol-style', 'LF'), ('foo', 'bar')]
co.apply_patch(
[patch.FilePatchDiff('svn_utils_test.txt', NAKED_PATCH, svn_props)])
filepath = os.path.join(self.root_dir, self.name, 'svn_utils_test.txt')
# Manually verify the properties.
props = subprocess2.check_output(
['svn', 'proplist', filepath],
cwd=self.root_dir).splitlines()[1:]
props = sorted(p.strip() for p in props)
expected_props = dict(svn_props)
self.assertEquals(sorted(expected_props.iterkeys()), props)
for k, v in expected_props.iteritems():
value = subprocess2.check_output(
['svn', 'propget', '--strict', k, filepath],
cwd=self.root_dir).strip()
self.assertEquals(v, value)
def testWithRevPropsSupport(self):
# Add the hook that will commit in a way that removes the race condition.
hook = os.path.join(self.FAKE_REPOS.svn_repo, 'hooks', 'pre-commit')
shutil.copyfile(os.path.join(ROOT_DIR, 'sample_pre_commit_hook'), hook)
os.chmod(hook, 0755)
expected = {
'revprops': [('commit-bot', 'user1@example.com')],
}
self._check(False, expected)
def testWithRevPropsSupportNotCommitBot(self):
# Add the hook that will commit in a way that removes the race condition.
hook = os.path.join(self.FAKE_REPOS.svn_repo, 'hooks', 'pre-commit')
shutil.copyfile(os.path.join(ROOT_DIR, 'sample_pre_commit_hook'), hook)
os.chmod(hook, 0755)
co = checkout.SvnCheckout(
self.root_dir, self.name,
self.FAKE_REPOS.USERS[1][0], self.FAKE_REPOS.USERS[1][1],
self.svn_url)
root = os.path.join(self.root_dir, self.name)
expected = {
'author': self.FAKE_REPOS.USERS[1][0],
}
self._check_base(co, root, False, expected)
def testAutoProps(self):
co = self._get_co(False)
co.svn_config = checkout.SvnConfig(
os.path.join(ROOT_DIR, 'subversion_config'))
co.prepare()
patches = self.get_patches()
co.apply_patch(patches)
self.assertEquals(
['bin_file', 'extra', 'new_dir/subdir/new_file', 'svn_utils_test.txt'],
sorted(patches.filenames))
# *.txt = svn:eol-style=LF in subversion_config/config.
out = subprocess2.check_output(
['svn', 'pget', 'svn:eol-style', 'svn_utils_test.txt'],
cwd=co.project_path)
self.assertEquals('LF\n', out)
class GitSvnCheckout(SvnBaseTest):
name = 'foo.git'
def _get_co(self, read_only):
co = checkout.GitSvnCheckout(
self.root_dir, self.name[:-4],
self.usr, self.pwd,
self.svn_base, self.svn_trunk)
if read_only:
co = checkout.ReadOnlyCheckout(co)
else:
# Hack to simplify testing.
co.checkout = co
return co
def _check(self, read_only, expected):
root = os.path.join(self.root_dir, self.name)
self._check_base(self._get_co(read_only), root, True, expected)
def testAllRO(self):
self._check(True, None)
def testAllRW(self):
expected = {
'author': self.FAKE_REPOS.USERS[0][0],
}
self._check(False, expected)
def testGitSvnPremade(self):
# Test premade git-svn clone. First make a git-svn clone.
git_svn_co = self._get_co(True)
revision = git_svn_co.prepare()
self.assertEquals(self.previous_log['revision'], revision)
# Then use GitSvnClone to clone it to lose the git-svn connection and verify
# git svn init / git svn fetch works.
git_svn_clone = checkout.GitSvnPremadeCheckout(
self.root_dir, self.name[:-4] + '2', 'trunk',
self.usr, self.pwd,
self.svn_base, self.svn_trunk, git_svn_co.project_path)
self.assertEquals(self.previous_log['revision'], git_svn_clone.prepare())
def testException(self):
self._check_exception(
self._get_co(True), 'fatal: corrupt patch at line 12\n')
def testSvnProps(self):
co = self._get_co(False)
co.prepare()
try:
svn_props = [('foo', 'bar')]
co.apply_patch(
[patch.FilePatchDiff('svn_utils_test.txt', NAKED_PATCH, svn_props)])
self.fail()
except patch.UnsupportedPatchFormat, e:
self.assertEquals(e.filename, 'svn_utils_test.txt')
self.assertEquals(
e.status,
'Cannot apply svn property foo to file svn_utils_test.txt.')
co.prepare()
# svn:eol-style is ignored.
svn_props = [('svn:eol-style', 'LF')]
co.apply_patch(
[patch.FilePatchDiff('svn_utils_test.txt', NAKED_PATCH, svn_props)])
class RawCheckout(SvnBaseTest):
def setUp(self):
super(RawCheckout, self).setUp()
# Use a svn checkout as the base.
self.base_co = checkout.SvnCheckout(
self.root_dir, self.name, None, None, self.svn_url)
self.base_co.prepare()
def _get_co(self, read_only):
co = checkout.RawCheckout(self.root_dir, self.name)
if read_only:
return checkout.ReadOnlyCheckout(co)
return co
def _check(self, read_only):
root = os.path.join(self.root_dir, self.name)
co = self._get_co(read_only)
# A copy of BaseTest._check_base()
self.assertEquals(root, co.project_path)
self.assertEquals(None, co.prepare())
self.assertEquals('pouet', co.get_settings('bar'))
self.assertTree(self.get_trunk(False), root)
patches = self.get_patches()
co.apply_patch(patches)
self.assertEquals(
['bin_file', 'extra', 'new_dir/subdir/new_file', 'svn_utils_test.txt'],
sorted(patches.filenames))
# Verify that the patch is applied even for read only checkout.
self.assertTree(self.get_trunk(True), root)
if read_only:
revision = co.commit('msg', self.FAKE_REPOS.USERS[1][0])
self.assertEquals('FAKE', revision)
else:
try:
co.commit('msg', self.FAKE_REPOS.USERS[1][0])
self.fail()
except NotImplementedError:
pass
self.assertTree(self.get_trunk(True), root)
# Verify that prepare() is a no-op.
self.assertEquals(None, co.prepare())
self.assertTree(self.get_trunk(True), root)
def testAllRW(self):
self._check(False)
def testAllRO(self):
self._check(True)
def testException(self):
self._check_exception(
self._get_co(True),
'patching file svn_utils_test.txt\n'
'Hunk #1 FAILED at 3.\n'
'1 out of 1 hunk FAILED -- saving rejects to file '
'svn_utils_test.txt.rej\n')
if __name__ == '__main__':
if '-v' in sys.argv:
DEBUGGING = True
logging.basicConfig(
level=logging.DEBUG,
format='%(levelname)5s %(filename)15s(%(lineno)3d): %(message)s')
else:
logging.basicConfig(
level=logging.ERROR,
format='%(levelname)5s %(filename)15s(%(lineno)3d): %(message)s')
unittest.main()

@ -0,0 +1,187 @@
#!/usr/bin/env python
# Copyright (c) 2011 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 at
# http://src.chromium.org/viewvc/chrome/trunk/src/LICENSE
"""Commit bot fake author svn server hook.
Looks for svn commit --withrevprop realauthor=foo, replaces svn:author with this
author and sets the property commitbot to the commit bot credential to signify
this revision was committed with the commit bot.
It achieves its goal using an undocumented way. This script could use 'svnlook'
to read revprop properties but the code would still be needed to overwrite the
properties.
http://svnbook.red-bean.com/nightly/en/svn.reposadmin.create.html#svn.reposadmin.create.hooks
strongly advise against modifying a transation in a commit because the svn
client caches certain bits of repository data. Upon asking subversion devs,
having the wrong svn:author cached on the commit checkout is the worst that can
happen.
This code doesn't care about this issue because only the commit bot will trigger
this code, which runs in a controlled environment.
The transaction file format is also extremely unlikely to change. If it does,
the hook will throw an UnexpectedFileFormat exception which will be silently
ignored.
"""
import os
import re
import sys
class UnexpectedFileFormat(Exception):
"""The transaction file format is not the format expected."""
def read_svn_dump(filepath):
"""Returns list of (K, V) from a keyed svn file.
Don't use a map so ordering is kept.
raise UnexpectedFileFormat if the file cannot be understood.
"""
class InvalidHeaderLine(Exception):
"""Raised by read_entry when the line read is not the format expected.
"""
try:
f = open(filepath, 'rb')
except EnvironmentError:
raise UnexpectedFileFormat('The transaction file cannot be opened')
try:
out = []
def read_entry(entrytype):
header = f.readline()
match = re.match(r'^' + entrytype + ' (\d+)$', header)
if not match:
raise InvalidHeaderLine(header)
datalen = int(match.group(1))
data = f.read(datalen)
if len(data) != datalen:
raise UnpexpectedFileFormat(
'Data value is not the expected length')
# Reads and ignore \n
if f.read(1) != '\n':
raise UnpexpectedFileFormat('Data value doesn\'t end with \\n')
return data
while True:
try:
key = read_entry('K')
except InvalidHeaderLine, e:
# Check if it's the end of the file.
if e.args[0] == 'END\n':
break
raise UnpexectedFileFormat('Failed to read a key: %s' % e)
try:
value = read_entry('V')
except InvalidHeaderLine, e:
raise UnpexectedFileFormat('Failed to read a value: %s' % e)
out.append([key, value])
return out
finally:
f.close()
def write_svn_dump(filepath, data):
"""Writes a svn keyed file with a list of (K, V)."""
f = open(filepath, 'wb')
try:
def write_entry(entrytype, value):
f.write('%s %d\n' % (entrytype, len(value)))
f.write(value)
f.write('\n')
for k, v in data:
write_entry('K', k)
write_entry('V', v)
f.write('END\n')
finally:
f.close()
def find_key(data, key):
"""Finds the item in a list of tuple where item[0] == key.
asserts if there is more than one item with the key.
"""
items = [i for i in data if i[0] == key]
if not items:
return None
assert len(items) == 1
return items[0]
def handle_commit_bot(repo_path, tx, commit_bot, admin_email):
"""Replaces svn:author with realauthor and sets commit-bot."""
# The file format is described there:
# http://svn.apache.org/repos/asf/subversion/trunk/notes/dump-load-format.txt
propfilepath = os.path.join(
repo_path, 'db', 'transactions', tx + '.txn', 'props')
# Do a lot of checks to make sure everything is in the expected format.
try:
data = read_svn_dump(propfilepath)
except UnexpectedFileFormat:
return (
'Failed to parse subversion server transaction format.\n'
'Please contact %s ASAP with\n'
'this error message.') % admin_email
if not data:
return (
'Failed to load subversion server transaction file.\n'
'Please contact %s ASAP with\n'
'this error message.') % admin_email
realauthor = find_key(data, 'realauthor')
if not realauthor:
# That's fine, there is no author to fake.
return
author = find_key(data, 'svn:author')
if not author or not author[1]:
return (
'Failed to load svn:author from the transaction file.\n'
'Please contact %s ASAP with\n'
'this error message.') % admin_email
if author[1] != commit_bot:
# The author will not be changed and realauthor will be kept as a
# revision property.
return
if len(realauthor[1]) > 50:
return 'Fake author was rejected due to being too long.'
if not re.match(r'^[a-zA-Z0-9\@\-\_\+\%\.]+$', realauthor[1]):
return 'Fake author was rejected due to not passing regexp.'
# Overwrite original author
author[1] = realauthor[1]
# Remove realauthor svn property
data.remove(realauthor)
# Add svn property commit-bot=<commit-bot username>
data.append(('commit-bot', commit_bot))
write_svn_dump(propfilepath, data)
def main():
# Replace with your commit-bot credential.
commit_bot = 'user1@example.com'
admin_email = 'dude@example.com'
ret = handle_commit_bot(sys.argv[1], sys.argv[2], commit_bot, admin_email)
if ret:
print >> sys.stderr, ret
return 1
return 0
if __name__ == '__main__':
sys.exit(main())
# vim: ts=4:sw=4:tw=80:et:

@ -0,0 +1,47 @@
# Chromium-specific config file to put at ~/.subversion/config or %USERPROFILE%\AppData\Roaming\Subversion\config
# Inspired by http://src.chromium.org/svn/trunk/tools/build/slave/config
[auth]
# Warning, this is insecure.
store-passwords=yes
[miscellany]
global-ignores = *.pyc *.user *.suo *.bak *~ #*# *.ncb *.o *.lo *.la .*~ .#* .DS_Store .*.swp *.scons *.mk *.Makefile *.sln *.vcproj *.rules SConstruct *.xcodeproj
enable-auto-props = yes
[auto-props]
*.afm = svn:eol-style=LF
*.bat = svn:eol-style=CRLF
*.c = svn:eol-style=LF
*.cc = svn:eol-style=LF
*.cpp = svn:eol-style=LF
*.css = svn:eol-style=LF
*.def = svn:eol-style=LF
*.dll = svn:executable
*.exe = svn:executable
*.grd = svn:eol-style=LF
*.gyp = svn:eol-style=LF
*.gypi = svn:eol-style=LF
*.h = svn:eol-style=LF
*.htm = svn:eol-style=LF
*.html = svn:eol-style=LF
*.idl = svn:eol-style=LF
*.jpg = svn:mime-type=image/jpeg
*.js = svn:eol-style=LF
*.m = svn:eol-style=LF
*.make = svn:eol-style=LF
*.mm = svn:eol-style=LF
*.mock-http-headers = svn:eol-style=LF
*.obsolete = svn:eol-style=LF
*.pdf = svn:mime-type=application/pdf
*.pl = svn:eol-style=LF
*.pm = svn:eol-style=LF
*.png = svn:mime-type=image/png
*.py = svn:eol-style=LF
*.pyd = svn:executable
*.sh = svn:eol-style=LF;svn:executable
*.txt = svn:eol-style=LF
*.webp = svn:mime-type=image/webp
*.xml = svn:eol-style=LF
*.xtb = svn:eol-style=LF
Makefile = svn:eol-style=LF
Loading…
Cancel
Save