blob: cdb69b028f318973c81aa5757d3635e5133fc85b [file] [log] [blame]
#!/usr/bin/env python
# Copyright (c) 2015 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 codecs
import copy
import json
import os
import sys
import unittest
#import test_env # pylint: disable=relative-import,unused-import
sys.path.insert(0, os.path.join(
'recipes', 'recipe_modules', 'bot_update', 'resources'))
import bot_update
class MockedPopen(object):
"""A fake instance of a called subprocess.
This is meant to be used in conjunction with MockedCall.
def __init__(self, args=None, kwargs=None):
self.args = args or []
self.kwargs = kwargs or {}
self.return_value = None
self.fails = False
def returns(self, rv):
"""Set the return value when this popen is called.
rv can be a string, or a callable (eg function).
self.return_value = rv
return self
def check(self, args, kwargs):
"""Check to see if the given args/kwargs call match this instance.
This does a partial match, so that a call to "git clone foo" will match
this instance if this instance was recorded as "git clone"
if any(input_arg != expected_arg
for (input_arg, expected_arg) in zip(args, self.args)):
return False
return self.return_value
def __call__(self, args, kwargs):
"""Actually call this popen instance."""
if hasattr(self.return_value, '__call__'):
return self.return_value(*args, **kwargs)
return self.return_value
class MockedCall(object):
"""A fake instance of
This object is pre-seeded with "answers" in self.expectations. The type
is a MockedPopen object, or any object with a __call__() and check() method.
The check() method is used to check to see if the correct popen object is
chosen (can be a partial match, eg a "git clone" popen module would match
a "git clone foo" call).
By default, if no answers have been pre-seeded, the call() returns successful
with an empty string.
def __init__(self, fake_filesystem):
self.expectations = []
self.records = []
def expect(self, args=None, kwargs=None):
args = args or []
kwargs = kwargs or {}
popen = MockedPopen(args, kwargs)
return popen
def __call__(self, *args, **kwargs):
self.records.append((args, kwargs))
for popen in self.expectations:
if popen.check(args, kwargs):
return popen(args, kwargs)
return ''
class MockedGclientSync():
"""A class producing a callable instance of gclient sync.
Because for bot_update, gclient sync also emits an output json file, we need
a callable object that can understand where the output json file is going, and
emit a (albite) fake file for bot_update to consume.
def __init__(self, fake_filesystem):
self.output = {}
self.fake_filesystem = fake_filesystem
self.records = []
def __call__(self, *args, **_):
output_json_index = args.index('--output-json') + 1
with[output_json_index], 'w') as f:
json.dump(self.output, f)
class FakeFile():
def __init__(self):
self.contents = ''
def write(self, buf):
self.contents += buf
def read(self):
return self.contents
def __enter__(self):
return self
def __exit__(self, _, __, ___):
class FakeFilesystem():
def __init__(self):
self.files = {}
def open(self, target, mode='r', encoding=None):
if 'w' in mode:
self.files[target] = FakeFile()
return self.files[target]
return self.files[target]
def fake_git(*args, **kwargs):
return'git', *args, **kwargs)
class BotUpdateUnittests(unittest.TestCase):
'solutions': [{
'name': 'somename',
'url': ''
'revisions': {},
'first_sln': 'somename',
'target_os': None,
'target_os_only': None,
'target_cpu': None,
'patch_root': None,
'patch_refs': [],
'gerrit_rebase_patch_ref': None,
'refs': [],
'git_cache_dir': '',
'cleanup_dir': None,
'gerrit_reset': None,
'disable_syntax_validation': False,
def setUp(self):
sys.platform = 'linux2' # For consistency, ya know?
self.filesystem = FakeFilesystem() = MockedCall(self.filesystem)
self.gclient = MockedGclientSync(self.filesystem)
(sys.executable, '-u', bot_update.GCLIENT_PATH, 'sync')
self.old_call = getattr(bot_update, 'call')
self.params = copy.deepcopy(self.DEFAULT_PARAMS)
setattr(bot_update, 'call',
setattr(bot_update, 'git', fake_git)
self.old_os_cwd = os.getcwd
setattr(os, 'getcwd', lambda: '/b/build/slave/foo/build')
setattr(bot_update, 'open',
self.old_codecs_open =
setattr(codecs, 'open',
def tearDown(self):
setattr(bot_update, 'call', self.old_call)
setattr(os, 'getcwd', self.old_os_cwd)
delattr(bot_update, 'open')
setattr(codecs, 'open', self.old_codecs_open)
def overrideSetupForWindows(self):
sys.platform = 'win'
(sys.executable, '-u', bot_update.GCLIENT_PATH, 'sync')
def testBasic(self):
def testBasicRevision(self):
self.params['revisions'] = {
'src': 'HEAD', 'src/v8': 'deadbeef', 'somename': 'DNE'}
args = self.gclient.records[0]
idx_first_revision = args.index('--revision')
idx_second_revision = args.index(
'--revision', idx_first_revision+1)
idx_third_revision = args.index('--revision', idx_second_revision+1)
self.assertEquals(args[idx_first_revision+1], 'somename@unmanaged')
self.assertEquals(args[idx_second_revision+1], 'src@origin/master')
self.assertEquals(args[idx_third_revision+1], 'src/v8@deadbeef')
def testApplyPatchOnGclient(self):
ref = 'refs/changes/12/345/6'
repo = ''
self.params['patch_refs'] = ['%s@%s' % (repo, ref)]
args = self.gclient.records[0]
idx = args.index('--patch-ref')
self.assertEqual(args[idx+1], self.params['patch_refs'][0])
self.assertNotIn('--patch-ref', args[idx+1:])
# Assert we're not patching in
for record in
self.assertNotIn('git fetch ' + repo,
' '.join(record[0]))
def testPatchRefs(self):
self.params['patch_refs'] = [
args = self.gclient.records[0]
patch_refs = set(
args[i+1] for i in xrange(len(args))
if args[i] == '--patch-ref' and i+1 < len(args))
self.assertIn(self.params['patch_refs'][0], patch_refs)
self.assertIn(self.params['patch_refs'][1], patch_refs)
def testBreakLocks(self):
gclient_sync_cmd = None
for record in
args = record[0]
if args[:4] == (sys.executable, '-u', bot_update.GCLIENT_PATH, 'sync'):
gclient_sync_cmd = args
self.assertTrue('--break_repo_locks' in gclient_sync_cmd)
def testGitCheckoutBreaksLocks(self):
path = '/b/build/slave/foo/build/.git'
lockfile = 'index.lock'
removed = []
old_os_walk = os.walk
old_os_remove = os.remove
setattr(os, 'walk', lambda _: [(path, None, [lockfile])])
setattr(os, 'remove', removed.append)
setattr(os, 'walk', old_os_walk)
setattr(os, 'remove', old_os_remove)
self.assertTrue(os.path.join(path, lockfile) in removed)
def testGenerateManifestsBasic(self):
gclient_output = {
'solutions': {
'breakpad/': {
'revision': None,
'scm': None,
'url': ('' +
"src/": {
'revision': 'f671d3baeb64d9dba628ad582e867cf1aebc0207',
'scm': None,
'url': ''
'src/overriden': {
'revision': None,
'scm': 'git',
'url': None,
out = bot_update.create_manifest(gclient_output, None)
self.assertEquals(len(out['directories']), 2)
self.assertNotIn('src/overridden', out['directories'])
def testParsesRevisions(self):
revisions = [
expected_results = {
'root': 'f671d3baeb64d9dba628ad582e867cf1aebc0207',
'src': 'deadbeef',
'': '12345',
'bar': 'refs/experimental/',
actual_results = bot_update.parse_revisions(revisions, 'root')
self.assertEqual(expected_results, actual_results)
if __name__ == '__main__':