Add post_processors override to apply_patch() and add more testing.

Move ReadOnlyCheckout tests into its own fixture.

R=dpranke@chromium.org
BUG=
TEST=


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

git-svn-id: svn://svn.chromium.org/chrome/trunk/tools/depot_tools@103272 0039d316-1c4b-4281-b951-d872f2087c98
experimental/szager/collated-output
maruel@chromium.org 14 years ago
parent bda0503b54
commit b1d1a78c8d

@ -92,7 +92,7 @@ class CheckoutBase(object):
"""
raise NotImplementedError()
def apply_patch(self, patches):
def apply_patch(self, patches, post_processors=None):
"""Applies a patch and returns the list of modified files.
This function should throw patch.UnsupportedPatchFormat or
@ -117,8 +117,9 @@ class RawCheckout(CheckoutBase):
"""Stubbed out."""
pass
def apply_patch(self, patches):
def apply_patch(self, patches, post_processors=None):
"""Ignores svn properties."""
post_processors = post_processors or self.post_processors or []
for p in patches:
try:
stdout = ''
@ -145,7 +146,7 @@ class RawCheckout(CheckoutBase):
elif p.is_new and not os.path.exists(filepath):
# There is only a header. Just create the file.
open(filepath, 'w').close()
for post in (self.post_processors or []):
for post in post_processors:
post(self, p)
except OSError, e:
raise PatchApplicationFailed(p.filename, '%s%s' % (stdout, e))
@ -261,7 +262,8 @@ class SvnCheckout(CheckoutBase, SvnMixIn):
(self.project_name, self.project_path))
return self._revert(revision)
def apply_patch(self, patches):
def apply_patch(self, patches, post_processors=None):
post_processors = post_processors or self.post_processors or []
for p in patches:
try:
# It is important to use credentials=False otherwise credentials could
@ -314,7 +316,7 @@ class SvnCheckout(CheckoutBase, SvnMixIn):
params = value.split('=', 1)
stdout += self._check_output_svn(
['propset'] + params + [p.filename], credentials=False)
for post in (self.post_processors or []):
for post in post_processors:
post(self, p)
except OSError, e:
raise PatchApplicationFailed(p.filename, '%s%s' % (stdout, e))
@ -418,13 +420,14 @@ class GitCheckoutBase(CheckoutBase):
if self.working_branch in branches:
self._call_git(['branch', '-D', self.working_branch])
def apply_patch(self, patches):
def apply_patch(self, patches, post_processors=None):
"""Applies a patch on 'working_branch' and switch to it.
Also commits the changes on the local branch.
Ignores svn properties and raise an exception on unexpected ones.
"""
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:
@ -461,7 +464,7 @@ class GitCheckoutBase(CheckoutBase):
p.filename,
'Cannot apply svn property %s to file %s.' % (
prop[0], p.filename))
for post in (self.post_processors or []):
for post in post_processors:
post(self, p)
except OSError, e:
raise PatchApplicationFailed(p.filename, '%s%s' % (stdout, e))
@ -693,9 +696,11 @@ class GitSvnCheckout(GitSvnCheckoutBase):
class ReadOnlyCheckout(object):
"""Converts a checkout into a read-only one."""
def __init__(self, checkout):
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)
@ -703,8 +708,9 @@ class ReadOnlyCheckout(object):
def get_settings(self, key):
return self.checkout.get_settings(key)
def apply_patch(self, patches):
return self.checkout.apply_patch(patches)
def apply_patch(self, patches, post_processors=None):
return self.checkout.apply_patch(
patches, post_processors or self.post_processors)
def commit(self, message, user): # pylint: disable=R0201
logging.info('Would have committed for %s with message: %s' % (

@ -77,6 +77,7 @@ class FakeRepos(fake_repos.FakeReposBase):
class BaseTest(fake_repos.FakeReposTestBase):
name = 'foo'
FAKE_REPOS_CLASS = FakeRepos
is_read_only = False
def setUp(self):
# Need to enforce subversion_config first.
@ -126,7 +127,8 @@ class BaseTest(fake_repos.FakeReposTestBase):
def _check_base(self, co, root, git, expected):
read_only = isinstance(co, checkout.ReadOnlyCheckout)
assert not read_only == bool(expected)
self.assertEquals(not read_only, bool(expected))
self.assertEquals(read_only, self.is_read_only)
if not read_only:
self.FAKE_REPOS.svn_dirty = True
@ -145,7 +147,7 @@ class BaseTest(fake_repos.FakeReposTestBase):
# pylint: disable=W0212
self.assertEquals(
(['master', 'working_branch'], 'working_branch'),
co.checkout._branches())
co._branches())
# Verify that the patch is applied even for read only checkout.
self.assertTree(self.get_trunk(True), root)
@ -193,7 +195,10 @@ class BaseTest(fake_repos.FakeReposTestBase):
ps = self.get_patches()
results = []
co.apply_patch(ps)
expected = [(co, p) for p in ps.patches]
expected_co = getattr(co, 'checkout', co)
# Because of ReadOnlyCheckout.
expected = [(expected_co, p) for p in ps.patches]
self.assertEquals(len(expected), len(results))
self.assertEquals(expected, results)
@ -238,32 +243,23 @@ class SvnBaseTest(BaseTest):
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 _get_co(self, post_processors):
self.assertNotEqual(False, post_processors)
return checkout.SvnCheckout(
self.root_dir, self.name, self.usr, self.pwd, self.svn_url,
post_processors)
def testAllRW(self):
def testAll(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)
root = os.path.join(self.root_dir, self.name)
self._check_base(self._get_co(None), root, False, expected)
def testException(self):
self._check_exception(
self._get_co(True),
self._get_co(None),
'While running patch -p1 --forward --force;\n'
'patching file chrome/file.cc\n'
'Hunk #1 FAILED at 3.\n'
@ -271,7 +267,7 @@ class SvnCheckout(SvnBaseTest):
'chrome/file.cc.rej\n')
def testSvnProps(self):
co = self._get_co(False)
co = self._get_co(None)
co.prepare(None)
try:
# svn:ignore can only be applied to directories.
@ -313,7 +309,8 @@ class SvnCheckout(SvnBaseTest):
expected = {
'revprops': [('commit-bot', 'user1@example.com')],
}
self._check(False, expected)
root = os.path.join(self.root_dir, self.name)
self._check_base(self._get_co(None), root, False, expected)
def testWithRevPropsSupportNotCommitBot(self):
# Add the hook that will commit in a way that removes the race condition.
@ -331,7 +328,7 @@ class SvnCheckout(SvnBaseTest):
self._check_base(co, root, False, expected)
def testAutoProps(self):
co = self._get_co(False)
co = self._get_co(None)
co.svn_config = checkout.SvnConfig(
os.path.join(ROOT_DIR, 'subversion_config'))
co.prepare(None)
@ -347,52 +344,32 @@ class SvnCheckout(SvnBaseTest):
self.assertEquals('LF\n', out)
def testProcess(self):
co = lambda x: checkout.SvnCheckout(
self.root_dir, self.name,
None, None,
self.svn_url,
x)
self._test_process(co)
self._test_process(self._get_co)
def testPrepare(self):
co = checkout.SvnCheckout(
self.root_dir, self.name,
None, None,
self.svn_url)
self._test_prepare(co)
self._test_prepare(self._get_co(None))
class GitSvnCheckout(SvnBaseTest):
name = 'foo.git'
def _get_co(self, read_only):
co = checkout.GitSvnCheckout(
def _get_co(self, post_processors):
self.assertNotEqual(False, post_processors)
return 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)
self.svn_base, self.svn_trunk, post_processors)
def testAllRO(self):
self._check(True, None)
def testAllRW(self):
def testAll(self):
expected = {
'author': self.FAKE_REPOS.USERS[0][0],
}
self._check(False, expected)
root = os.path.join(self.root_dir, self.name)
self._check_base(self._get_co(None), root, True, expected)
def testGitSvnPremade(self):
# Test premade git-svn clone. First make a git-svn clone.
git_svn_co = self._get_co(True)
git_svn_co = self._get_co(None)
revision = git_svn_co.prepare(None)
self.assertEquals(self.previous_log['revision'], revision)
# Then use GitSvnClone to clone it to lose the git-svn connection and verify
@ -406,10 +383,10 @@ class GitSvnCheckout(SvnBaseTest):
def testException(self):
self._check_exception(
self._get_co(True), 'fatal: corrupt patch at line 12\n')
self._get_co(None), 'fatal: corrupt patch at line 12\n')
def testSvnProps(self):
co = self._get_co(False)
co = self._get_co(None)
co.prepare(None)
try:
svn_props = [('foo', 'bar')]
@ -428,17 +405,18 @@ class GitSvnCheckout(SvnBaseTest):
[patch.FilePatchDiff('chrome/file.cc', RAW.PATCH, svn_props)])
def testProcess(self):
co = lambda x: checkout.SvnCheckout(
self.root_dir, self.name,
None, None,
self.svn_url, x)
self._test_process(co)
self._test_process(self._get_co)
def testPrepare(self):
co = checkout.SvnCheckout(
self.root_dir, self.name,
None, None,
self.svn_url)
co = self._get_co(None)
# TODO(maruel): Cheat here until prepare(revision != None) implemented.
co.old_prepare = co.prepare
def prepare(rev):
# Basically, test that it is broken.
self.assertEquals(1, rev)
self.assertEquals(2, co.old_prepare(None))
return 1
co.prepare = prepare
self._test_prepare(co)
@ -450,15 +428,14 @@ class RawCheckout(SvnBaseTest):
self.root_dir, self.name, None, None, self.svn_url)
self.base_co.prepare(None)
def _get_co(self, read_only):
co = checkout.RawCheckout(self.root_dir, self.name, None)
if read_only:
return checkout.ReadOnlyCheckout(co)
return co
def _get_co(self, post_processors):
self.assertNotEqual(False, post_processors)
return checkout.RawCheckout(self.root_dir, self.name, post_processors)
def _check(self, read_only):
def testAll(self):
# Can't use self._check_base() since it's too different.
root = os.path.join(self.root_dir, self.name)
co = self._get_co(read_only)
co = self._get_co(None)
# A copy of BaseTest._check_base()
self.assertEquals(root, co.project_path)
@ -473,47 +450,68 @@ class RawCheckout(SvnBaseTest):
# Verify that the patch is applied even for read only checkout.
self.assertTree(self.get_trunk(True), root)
if read_only:
revision = co.commit(u'msg', self.FAKE_REPOS.USERS[1][0])
self.assertEquals('FAKE', revision)
else:
try:
co.commit(u'msg', self.FAKE_REPOS.USERS[1][0])
self.fail()
except NotImplementedError:
pass
try:
co.commit(u'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(None))
self.assertTree(self.get_trunk(True), root)
def testAllRW(self):
self._check(False)
def testException(self):
self._check_exception(
self._get_co(None),
'patching file chrome/file.cc\n'
'Hunk #1 FAILED at 3.\n'
'1 out of 1 hunk FAILED -- saving rejects to file '
'chrome/file.cc.rej\n')
def testProcess(self):
self._test_process(self._get_co)
def testPrepare(self):
# RawCheckout doesn't support prepare() but emulate it.
co = self._get_co(None)
revs = [1]
def prepare(asked):
self.assertEquals(1, asked)
return revs.pop(0)
co.prepare = prepare
self._test_prepare(co)
self.assertEquals([], revs)
def testAllRO(self):
self._check(True)
class ReadOnlyCheckout(SvnBaseTest):
# Use SvnCheckout as the backed since it support read-only checkouts too.
is_read_only = True
def _get_co(self, post_processors):
self.assertNotEqual(False, post_processors)
return checkout.ReadOnlyCheckout(
checkout.SvnCheckout(
self.root_dir, self.name, None, None, self.svn_url, None),
post_processors)
def testAll(self):
root = os.path.join(self.root_dir, self.name)
self._check_base(self._get_co(None), root, False, None)
def testException(self):
self._check_exception(
self._get_co(True),
self._get_co(None),
'While running patch -p1 --forward --force;\n'
'patching file chrome/file.cc\n'
'Hunk #1 FAILED at 3.\n'
'1 out of 1 hunk FAILED -- saving rejects to file '
'chrome/file.cc.rej\n')
def testProcess(self):
co = lambda x: checkout.SvnCheckout(
self.root_dir, self.name,
None, None,
self.svn_url, x)
self._test_process(co)
self._test_process(self._get_co)
def testPrepare(self):
co = checkout.SvnCheckout(
self.root_dir, self.name,
None, None,
self.svn_url)
self._test_prepare(co)
self._test_prepare(self._get_co(None))
if __name__ == '__main__':

Loading…
Cancel
Save