| #!/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( |
| os.path.dirname(os.path.dirname(os.path.abspath(__file__))), |
| '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 bot_update.call(). |
| |
| 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) |
| self.expectations.append(popen) |
| return popen |
| |
| def __call__(self, *args, **kwargs): |
| self.records.append((args, kwargs)) |
| for popen in self.expectations: |
| if popen.check(args, kwargs): |
| self.expectations.remove(popen) |
| 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 self.fake_filesystem.open(args[output_json_index], 'w') as f: |
| json.dump(self.output, f) |
| self.records.append(args) |
| |
| |
| 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, _, __, ___): |
| pass |
| |
| |
| 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 bot_update.call('git', *args, **kwargs) |
| |
| |
| class BotUpdateUnittests(unittest.TestCase): |
| DEFAULT_PARAMS = { |
| 'solutions': [{ |
| 'name': 'somename', |
| 'url': 'https://fake.com' |
| }], |
| 'revisions': {}, |
| 'first_sln': 'somename', |
| 'target_os': None, |
| 'target_os_only': None, |
| 'target_cpu': None, |
| 'patch_root': None, |
| 'patch_refs': [], |
| 'gerrit_rebase_patch_ref': None, |
| 'no_fetch_tags': False, |
| '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() |
| self.call = MockedCall(self.filesystem) |
| self.gclient = MockedGclientSync(self.filesystem) |
| self.call.expect( |
| (sys.executable, '-u', bot_update.GCLIENT_PATH, 'sync') |
| ).returns(self.gclient) |
| self.old_call = getattr(bot_update, 'call') |
| self.params = copy.deepcopy(self.DEFAULT_PARAMS) |
| setattr(bot_update, 'call', self.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.filesystem.open) |
| self.old_codecs_open = codecs.open |
| setattr(codecs, 'open', self.filesystem.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' |
| self.call.expect( |
| (sys.executable, '-u', bot_update.GCLIENT_PATH, 'sync') |
| ).returns(self.gclient) |
| |
| def testBasic(self): |
| bot_update.ensure_checkout(**self.params) |
| return self.call.records |
| |
| def testBasicRevision(self): |
| self.params['revisions'] = { |
| 'src': 'HEAD', 'src/v8': 'deadbeef', 'somename': 'DNE'} |
| bot_update.ensure_checkout(**self.params) |
| 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.assertEqual(args[idx_first_revision+1], 'somename@unmanaged') |
| self.assertEqual(args[idx_second_revision+1], 'src@origin/master') |
| self.assertEqual(args[idx_third_revision+1], 'src/v8@deadbeef') |
| return self.call.records |
| |
| def testTagsByDefault(self): |
| bot_update.ensure_checkout(**self.params) |
| found = False |
| for record in self.call.records: |
| args = record[0] |
| if args[:3] == ('git', 'cache', 'populate'): |
| self.assertFalse('--no-fetch-tags' in args) |
| found = True |
| self.assertTrue(found) |
| return self.call.records |
| |
| def testNoTags(self): |
| params = self.params |
| params['no_fetch_tags'] = True |
| bot_update.ensure_checkout(**params) |
| found = False |
| for record in self.call.records: |
| args = record[0] |
| if args[:3] == ('git', 'cache', 'populate'): |
| self.assertTrue('--no-fetch-tags' in args) |
| found = True |
| self.assertTrue(found) |
| return self.call.records |
| |
| def testApplyPatchOnGclient(self): |
| ref = 'refs/changes/12/345/6' |
| repo = 'https://chromium.googlesource.com/v8/v8' |
| self.params['patch_refs'] = ['%s@%s' % (repo, ref)] |
| bot_update.ensure_checkout(**self.params) |
| 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 bot_update.py |
| for record in self.call.records: |
| self.assertNotIn('git fetch ' + repo, |
| ' '.join(record[0])) |
| |
| def testPatchRefs(self): |
| self.params['patch_refs'] = [ |
| 'https://chromium.googlesource.com/chromium/src@refs/changes/12/345/6', |
| 'https://chromium.googlesource.com/v8/v8@refs/changes/1/234/56'] |
| bot_update.ensure_checkout(**self.params) |
| 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): |
| self.overrideSetupForWindows() |
| bot_update.ensure_checkout(**self.params) |
| gclient_sync_cmd = None |
| for record in self.call.records: |
| 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): |
| self.overrideSetupForWindows() |
| 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) |
| bot_update.ensure_checkout(**self.params) |
| 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': ('https://chromium.googlesource.com/breakpad.git' + |
| '@5f638d532312685548d5033618c8a36f73302d0a') |
| }, |
| "src/": { |
| 'revision': 'f671d3baeb64d9dba628ad582e867cf1aebc0207', |
| 'scm': None, |
| 'url': 'https://chromium.googlesource.com/a/chromium/src.git' |
| }, |
| 'src/overriden': { |
| 'revision': None, |
| 'scm': 'git', |
| 'url': None, |
| }, |
| } |
| } |
| out = bot_update.create_manifest(gclient_output, None) |
| self.assertEqual(len(out['directories']), 2) |
| self.assertEqual( |
| out['directories']['src']['git_checkout']['revision'], |
| 'f671d3baeb64d9dba628ad582e867cf1aebc0207') |
| self.assertEqual( |
| out['directories']['src']['git_checkout']['repo_url'], |
| 'https://chromium.googlesource.com/chromium/src') |
| self.assertEqual( |
| out['directories']['breakpad']['git_checkout']['revision'], |
| '5f638d532312685548d5033618c8a36f73302d0a') |
| self.assertEqual( |
| out['directories']['breakpad']['git_checkout']['repo_url'], |
| 'https://chromium.googlesource.com/breakpad') |
| self.assertNotIn('src/overridden', out['directories']) |
| |
| def testParsesRevisions(self): |
| revisions = [ |
| 'f671d3baeb64d9dba628ad582e867cf1aebc0207', |
| 'src@deadbeef', |
| 'https://foo.googlesource.com/bar@12345', |
| 'bar@refs/experimental/test@example.com/test', |
| ] |
| expected_results = { |
| 'root': 'f671d3baeb64d9dba628ad582e867cf1aebc0207', |
| 'src': 'deadbeef', |
| 'https://foo.googlesource.com/bar.git': '12345', |
| 'bar': 'refs/experimental/test@example.com/test', |
| } |
| actual_results = bot_update.parse_revisions(revisions, 'root') |
| self.assertEqual(expected_results, actual_results) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
