123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474 |
- #!/usr/bin/env vpython3
- # Copyright 2020 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 json
- import os
- import sys
- import unittest
- import zipfile
- import six
- import mock # pylint: disable=import-error
- from mock import call # pylint: disable=import-error
- from mock import patch # pylint: disable=import-error
- sys.path.append(
- os.path.join(
- os.path.dirname(__file__), os.pardir, os.pardir, 'third_party',
- 'catapult', 'common', 'py_utils'))
- # pylint: disable=wrong-import-position,import-error
- from py_utils import tempfile_ext
- import update_cts
- import cts_utils
- import cts_utils_test
- from cts_utils_test import CIPD_DATA, CONFIG_DATA, DEPS_DATA
- from cts_utils_test import GENERATE_BUILDBOT_JSON
- def generate_zip_file(path, *files):
- """Create a zip file containing a list of files.
- Args:
- path: Path to generated zip file
- files: one or more file entries in the zip file,
- contents of which will be the same as the file's name
- """
- path_dir = os.path.dirname(path)
- if path_dir and not os.path.isdir(path_dir):
- os.makedirs(path_dir)
- with zipfile.ZipFile(path, 'w') as zf:
- for f in files:
- zf.writestr(f, f)
- def verify_zip_file(path, *files):
- """Verify zip file that was generated.
- Args:
- path: Path to generated zip file
- files: one or more file entries in the zip file,
- contents of which should be the same as the file entry
- """
- with zipfile.ZipFile(path) as zf:
- names = zf.namelist()
- if len(files) != len(names):
- raise AssertionError('Expected ' + len(files) + ' files, found ' +
- len(names) + '.')
- for f in files:
- if f not in names:
- raise AssertionError(f + ' should be in zip file.')
- s = zf.read(f)
- if six.ensure_str(s) != f:
- raise AssertionError('Expected ' + f + ', found ' + s)
- class FakeDownload:
- """Allows test to simulate downloads of CTS zip files."""
- def __init__(self):
- self._files = {}
- def add_fake_zip_files(self, cts_config_json):
- """Associate generated zip files to origin urls in config.
- Each origin url will be associated with a zip file, contents of which will
- be apks specified for that platform. Contents of each apk will just be the
- path of that apk as a string.
- Args:
- cts_config_json: config json string
- """
- config = json.loads(cts_config_json)
- for p in config:
- for a in config[p]['arch']:
- o = config[p]['arch'][a]['_origin']
- for test_run in config[p]['test_runs']:
- self.append_to_zip_file(o, test_run['apk'])
- for additional_apk in test_run.get('additional_apks', []):
- self.append_to_zip_file(o, additional_apk['apk'])
- def append_to_zip_file(self, url, file_name):
- """Append files to any zip files associated with the url.
- If no zip files are associated with the url, one will be created first.
- Contents of each file_name will just be the path of that apk as a string.
- If file_name is already associated with the url, then this is a no-op.
- Args:
- url: Url to associate
- file_name: Path to add to the zip file associated with the url
- """
- if url not in self._files:
- self._files[url] = []
- if file_name not in self._files[url]:
- self._files[url].append(file_name)
- def download(self, url, dest):
- if url not in self._files:
- raise AssertionError('Url should be found: ' + url)
- cts_utils_test.check_tempdir(dest)
- with zipfile.ZipFile(dest, mode='w') as zf:
- for file_name in self._files[url]:
- zf.writestr(file_name, file_name)
- class UpdateCTSTest(unittest.TestCase):
- """Unittests for update_cts.py."""
- @patch('cts_utils.download')
- def testDownloadCTS_onePlatform(self, download_mock):
- with tempfile_ext.NamedTemporaryDirectory() as workDir,\
- tempfile_ext.NamedTemporaryDirectory() as repoRoot:
- cts_utils_test.setup_fake_repo(repoRoot)
- cts_updater = update_cts.UpdateCTS(workDir, repoRoot)
- self.assertEqual(os.path.join(workDir, 'downloaded'),
- cts_updater.download_dir)
- cts_updater.download_cts(platforms=['platform1'])
- download_mock.assert_has_calls([
- call(CONFIG_DATA['origin11'],
- os.path.join(cts_updater.download_dir, CONFIG_DATA['base11'])),
- call(CONFIG_DATA['origin12'],
- os.path.join(cts_updater.download_dir, CONFIG_DATA['base12']))
- ])
- self.assertEqual(2, download_mock.call_count)
- @patch('cts_utils.download')
- def testDownloadCTS_allPlatforms(self, download_mock):
- with tempfile_ext.NamedTemporaryDirectory() as workDir,\
- tempfile_ext.NamedTemporaryDirectory() as repoRoot:
- cts_utils_test.setup_fake_repo(repoRoot)
- cts_updater = update_cts.UpdateCTS(workDir, repoRoot)
- cts_updater.download_cts()
- download_mock.assert_has_calls([
- call(CONFIG_DATA['origin11'],
- os.path.join(cts_updater.download_dir, CONFIG_DATA['base11'])),
- call(CONFIG_DATA['origin12'],
- os.path.join(cts_updater.download_dir, CONFIG_DATA['base12'])),
- call(CONFIG_DATA['origin21'],
- os.path.join(cts_updater.download_dir, CONFIG_DATA['base21'])),
- call(CONFIG_DATA['origin22'],
- os.path.join(cts_updater.download_dir, CONFIG_DATA['base22']))
- ])
- self.assertEqual(4, download_mock.call_count)
- def testFilterCTS(self):
- with tempfile_ext.NamedTemporaryDirectory() as workDir,\
- tempfile_ext.NamedTemporaryDirectory() as repoRoot,\
- cts_utils.chdir(workDir):
- cts_utils_test.setup_fake_repo(repoRoot)
- cts_updater = update_cts.UpdateCTS('.', repoRoot)
- expected_download_dir = os.path.abspath('downloaded')
- self.assertEqual(expected_download_dir, cts_updater.download_dir)
- os.makedirs(expected_download_dir)
- with cts_utils.chdir('downloaded'):
- generate_zip_file(CONFIG_DATA['base11'], CONFIG_DATA['apk1'],
- 'not/a/webview/apk')
- generate_zip_file(CONFIG_DATA['base12'], CONFIG_DATA['apk1'],
- 'not/a/webview/apk')
- cts_updater.filter_downloaded_cts()
- with cts_utils.chdir('filtered'):
- self.assertEqual(2, len(os.listdir('.')))
- verify_zip_file(CONFIG_DATA['base11'], CONFIG_DATA['apk1'])
- verify_zip_file(CONFIG_DATA['base12'], CONFIG_DATA['apk1'])
- @patch('devil.utils.cmd_helper.RunCmd')
- @unittest.skipIf(os.name == "nt", "This fails on Windows because it calls "
- "download_cipd which ultimately calls cipd_ensure which "
- "creates a file with NamedTemporaryFile and then opens it "
- "by name, which hits permission errors.")
- def testDownloadCIPD(self, run_mock):
- with tempfile_ext.NamedTemporaryDirectory() as workDir,\
- tempfile_ext.NamedTemporaryDirectory() as repoRoot,\
- cts_utils.chdir(workDir):
- cts_utils_test.setup_fake_repo(repoRoot)
- fake_cipd = cts_utils_test.FakeCIPD()
- fake_cipd.add_package(
- os.path.join(repoRoot, cts_utils.TOOLS_DIR, cts_utils.CIPD_FILE),
- DEPS_DATA['revision'])
- fake_run_cmd = cts_utils_test.FakeRunCmd(fake_cipd)
- run_mock.side_effect = fake_run_cmd.run_cmd
- cts_updater = update_cts.UpdateCTS('.', repoRoot)
- cts_updater.download_cipd()
- self.assertTrue(os.path.isdir('cipd'))
- for i in [str(x) for x in range(1, 5)]:
- self.assertEqual(
- CIPD_DATA['file' + i],
- cts_utils_test.readfile(
- os.path.join(workDir, 'cipd', CIPD_DATA['file' + i])))
- @unittest.skipIf(os.name == "nt", "This fails on Windows because it calls "
- "download_cipd which ultimately calls cipd_ensure which "
- "creates a file with NamedTemporaryFile and then opens it "
- "by name, which hits permission errors.")
- def testDownloadCIPD_dirExists(self):
- with tempfile_ext.NamedTemporaryDirectory() as workDir,\
- tempfile_ext.NamedTemporaryDirectory() as repoRoot,\
- cts_utils.chdir(workDir):
- cts_utils_test.setup_fake_repo(repoRoot)
- cts_updater = update_cts.UpdateCTS('.', repoRoot)
- os.makedirs('cipd')
- with self.assertRaises(update_cts.DirExistsError):
- cts_updater.download_cipd()
- def testStageCIPDUpdate(self):
- with tempfile_ext.NamedTemporaryDirectory() as workDir,\
- tempfile_ext.NamedTemporaryDirectory() as repoRoot,\
- cts_utils.chdir(workDir):
- cts_utils_test.setup_fake_repo(repoRoot)
- cts_utils_test.writefile('n1',
- os.path.join('filtered', CONFIG_DATA['base11']))
- cts_utils_test.writefile('n3',
- os.path.join('filtered', CONFIG_DATA['base12']))
- for i in [str(i) for i in range(1, 5)]:
- cts_utils_test.writefile('o' + i,
- os.path.join('cipd', CIPD_DATA['file' + i]))
- cts_updater = update_cts.UpdateCTS('.', repoRoot)
- cts_updater.stage_cipd_update()
- self.assertTrue(os.path.isdir('staged'))
- with cts_utils.chdir('staged'):
- self.assertEqual('n1', cts_utils_test.readfile(CONFIG_DATA['file11']))
- self.assertEqual('n3', cts_utils_test.readfile(CONFIG_DATA['file12']))
- self.assertEqual('o2', cts_utils_test.readfile(CONFIG_DATA['file21']))
- self.assertEqual('o4', cts_utils_test.readfile(CONFIG_DATA['file22']))
- self.assertEqual(CIPD_DATA['yaml'],
- cts_utils_test.readfile('cipd.yaml'))
- @patch('devil.utils.cmd_helper.GetCmdOutput')
- def testUpdateCtsConfigFileOrigins(self, cmd_mock):
- with tempfile_ext.NamedTemporaryDirectory() as workDir,\
- tempfile_ext.NamedTemporaryDirectory() as repoRoot,\
- cts_utils.chdir(workDir):
- cmd_mock.return_value = """
- hash refs/tags/platform-1.0_r6
- hash refs/tags/platform-1.0_r7
- hash refs/tags/platform-1.0_r9
- hash refs/tags/platform-2.0_r2
- hash refs/tags/platform-2.0_r3
- """
- expected_config_file = json.loads(CONFIG_DATA['json'])
- expected_config_file['platform1']['arch']['arch1'][
- 'unzip_dir'] = 'arch1/path/platform1_r9'
- expected_config_file['platform1']['arch']['arch2'][
- 'unzip_dir'] = 'arch1/path/platform1_r9'
- expected_config_file['platform2']['arch']['arch1'][
- 'unzip_dir'] = 'arch1/path/platform2_r3'
- expected_config_file['platform2']['arch']['arch2'][
- 'unzip_dir'] = 'arch1/path/platform2_r3'
- cts_utils_test.setup_fake_repo(repoRoot)
- cts_updater = update_cts.UpdateCTS('.', repoRoot)
- cts_updater.update_cts_download_origins_cmd()
- with cts_utils.chdir(repoRoot):
- actual_config_file = json.loads(
- cts_utils_test.readfile(
- os.path.join(cts_utils.TOOLS_DIR, cts_utils.CONFIG_FILE)))
- self.assertEqual(expected_config_file, actual_config_file)
- @patch('cts_utils.update_cipd_package')
- def testCommitStagedCIPD(self, update_mock):
- with tempfile_ext.NamedTemporaryDirectory() as workDir,\
- tempfile_ext.NamedTemporaryDirectory() as repoRoot,\
- cts_utils.chdir(workDir):
- cts_utils_test.setup_fake_repo(repoRoot)
- cts_updater = update_cts.UpdateCTS('.', repoRoot)
- with self.assertRaises(update_cts.MissingDirError):
- cts_updater.commit_staged_cipd()
- cts_utils_test.writefile(CIPD_DATA['yaml'],
- os.path.join('staged', 'cipd.yaml'))
- with cts_utils.chdir('staged'):
- generate_zip_file(CONFIG_DATA['file11'], CONFIG_DATA['apk1'])
- generate_zip_file(CONFIG_DATA['file12'], CONFIG_DATA['apk1'])
- generate_zip_file(CONFIG_DATA['file21'], CONFIG_DATA['apk2a'],
- CONFIG_DATA['apk2b'])
- with self.assertRaises(update_cts.MissingFileError):
- cts_updater.commit_staged_cipd()
- generate_zip_file(CONFIG_DATA['file22'], CONFIG_DATA['apk2a'],
- CONFIG_DATA['apk2b'])
- update_mock.return_value = 'newcipdversion'
- cts_updater.commit_staged_cipd()
- update_mock.assert_called_with(
- os.path.join(workDir, 'staged', 'cipd.yaml'))
- self.assertEqual('newcipdversion',
- cts_utils_test.readfile('cipd_version.txt'))
- @patch('devil.utils.cmd_helper.RunCmd')
- @patch('devil.utils.cmd_helper.GetCmdOutput')
- @unittest.skipIf(os.name == "nt", "This fails on Windows because it calls "
- "update_repository which calls cipd_ensure which creates a "
- "file with NamedTemporaryFile and then opens it by name, "
- "which hits permission errors.")
- def testUpdateRepository(self, cmd_mock, run_mock):
- with tempfile_ext.NamedTemporaryDirectory() as workDir,\
- tempfile_ext.NamedTemporaryDirectory() as repoRoot,\
- cts_utils.chdir(workDir):
- cts_utils_test.setup_fake_repo(repoRoot)
- cts_updater = update_cts.UpdateCTS('.', repoRoot)
- cts_utils_test.writefile('newversion', 'cipd_version.txt')
- cts_utils_test.writefile(CIPD_DATA['yaml'],
- os.path.join('staged', 'cipd.yaml'))
- cts_utils_test.writefile(
- CIPD_DATA['yaml'], os.path.join(os.path.join('staged', 'cipd.yaml')))
- cmd_mock.return_value = ''
- run_mock.return_value = 0
- cts_updater.update_repository()
- self._assertCIPDVersionUpdated(repoRoot, 'newversion')
- repo_cipd_yaml = os.path.join(repoRoot, cts_utils.TOOLS_DIR,
- cts_utils.CIPD_FILE)
- run_mock.assert_any_call(
- ['cp',
- os.path.join(workDir, 'staged', 'cipd.yaml'), repo_cipd_yaml])
- run_mock.assert_any_call([
- 'cipd', 'ensure', '-root',
- os.path.dirname(repo_cipd_yaml), '-ensure-file', mock.ANY
- ])
- run_mock.assert_any_call(['python', GENERATE_BUILDBOT_JSON])
- @patch('devil.utils.cmd_helper.RunCmd')
- @patch('devil.utils.cmd_helper.GetCmdOutput')
- def testUpdateRepository_uncommitedChanges(self, cmd_mock, run_mock):
- with tempfile_ext.NamedTemporaryDirectory() as workDir,\
- tempfile_ext.NamedTemporaryDirectory() as repoRoot,\
- cts_utils.chdir(workDir):
- cts_utils_test.setup_fake_repo(repoRoot)
- cts_updater = update_cts.UpdateCTS('.', repoRoot)
- cts_utils_test.writefile('newversion', 'cipd_version.txt')
- cts_utils_test.writefile(CIPD_DATA['yaml'],
- os.path.join('staged', 'cipd.yaml'))
- cmd_mock.return_value = 'M DEPS'
- run_mock.return_value = 0
- with self.assertRaises(update_cts.UncommittedChangeException):
- cts_updater.update_repository()
- @patch('devil.utils.cmd_helper.RunCmd')
- @patch('devil.utils.cmd_helper.GetCmdOutput')
- def testUpdateRepository_buildbotUpdateError(self, cmd_mock, run_mock):
- with tempfile_ext.NamedTemporaryDirectory() as workDir,\
- tempfile_ext.NamedTemporaryDirectory() as repoRoot,\
- cts_utils.chdir(workDir):
- cts_utils_test.setup_fake_repo(repoRoot)
- cts_updater = update_cts.UpdateCTS('.', repoRoot)
- cts_utils_test.writefile('newversion', 'cipd_version.txt')
- cts_utils_test.writefile(CIPD_DATA['yaml'],
- os.path.join('staged', 'cipd.yaml'))
- cmd_mock.return_value = ''
- run_mock.return_value = 1
- with self.assertRaises(IOError):
- cts_updater.update_repository()
- @patch('devil.utils.cmd_helper.RunCmd')
- @patch('devil.utils.cmd_helper.GetCmdOutput')
- def testUpdateRepository_inconsistentFiles(self, cmd_mock, run_mock):
- with tempfile_ext.NamedTemporaryDirectory() as workDir,\
- tempfile_ext.NamedTemporaryDirectory() as repoRoot,\
- cts_utils.chdir(workDir):
- cts_utils_test.setup_fake_repo(repoRoot)
- cts_updater = update_cts.UpdateCTS('.', repoRoot)
- cts_utils_test.writefile('newversion', 'cipd_version.txt')
- cts_utils_test.writefile(CIPD_DATA['yaml'],
- os.path.join('staged', 'cipd.yaml'))
- cmd_mock.return_value = ''
- run_mock.return_value = 0
- cts_utils_test.writefile(
- CIPD_DATA['template'] %
- ('wrong/package/name', CIPD_DATA['file1'], CIPD_DATA['file2'],
- CIPD_DATA['file3'], CIPD_DATA['file4']),
- os.path.join(os.path.join('staged', 'cipd.yaml')))
- with self.assertRaises(update_cts.InconsistentFilesException):
- cts_updater.update_repository()
- @patch('devil.utils.cmd_helper.RunCmd')
- @patch('devil.utils.cmd_helper.GetCmdOutput')
- @patch.object(cts_utils.ChromiumRepoHelper, 'update_testing_json')
- @patch('urllib.urlretrieve' if six.PY2 else 'urllib.request.urlretrieve')
- @unittest.skipIf(os.name == "nt", "This fails on Windows because it calls "
- "create_cipd_cmd which calls download_cipd which ultimately "
- "calls cipd_ensure which creates a file with "
- "NamedTemporaryFile and then opens it by name, which hits "
- "permission errors.")
- def testCompleteUpdate(self, retrieve_mock, update_json_mock, cmd_mock,
- run_mock):
- with tempfile_ext.NamedTemporaryDirectory() as workDir,\
- tempfile_ext.NamedTemporaryDirectory() as repoRoot,\
- cts_utils.chdir(workDir):
- cts_utils_test.setup_fake_repo(repoRoot)
- fake_cipd = cts_utils_test.FakeCIPD()
- fake_cipd.add_package(
- os.path.join(repoRoot, cts_utils.TOOLS_DIR, cts_utils.CIPD_FILE),
- DEPS_DATA['revision'])
- fake_download = FakeDownload()
- fake_download.add_fake_zip_files(CONFIG_DATA['json'])
- fake_run_cmd = cts_utils_test.FakeRunCmd(fake_cipd)
- retrieve_mock.side_effect = fake_download.download
- run_mock.side_effect = fake_run_cmd.run_cmd
- update_json_mock.return_value = 0
- cmd_mock.return_value = ''
- cts_updater = update_cts.UpdateCTS('.', repoRoot)
- cts_updater.download_cts_cmd()
- cts_updater.create_cipd_cmd()
- cts_updater.update_repository_cmd()
- latest_version = fake_cipd.get_latest_version(
- 'chromium/android_webview/tools/cts_archive')
- self.assertNotEqual(DEPS_DATA['revision'], latest_version)
- self._assertCIPDVersionUpdated(repoRoot, latest_version)
- repo_cipd_yaml = os.path.join(repoRoot, cts_utils.TOOLS_DIR,
- cts_utils.CIPD_FILE)
- run_mock.assert_any_call(
- ['cp',
- os.path.join(workDir, 'staged', 'cipd.yaml'), repo_cipd_yaml])
- run_mock.assert_any_call([
- 'cipd', 'ensure', '-root',
- os.path.dirname(repo_cipd_yaml), '-ensure-file', mock.ANY
- ])
- update_json_mock.assert_called_with()
- def _assertCIPDVersionUpdated(self, repo_root, new_version):
- """Check that cts cipd version in DEPS and test suites were updated.
- Args:
- repo_root: Root directory of checkout
- new_version: Expected version of CTS package
- Raises:
- AssertionError: If contents of DEPS and test suite files were not
- expected.
- """
- self.assertEqual(
- DEPS_DATA['template'] % (CIPD_DATA['package'], new_version),
- cts_utils_test.readfile(os.path.join(repo_root, 'DEPS')))
- self.assertEqual(
- cts_utils_test.SUITES_DATA['template'] % (new_version, new_version),
- cts_utils_test.readfile(
- os.path.join(repo_root, 'testing', 'buildbot', 'test_suites.pyl')))
- if __name__ == '__main__':
- unittest.main()
|