| # Copyright 2012 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Unittests for upload_prebuilts.py.""" |
| |
| import copy |
| import multiprocessing |
| import os |
| import tempfile |
| from unittest import mock |
| |
| from chromite.lib import binpkg |
| from chromite.lib import cros_test_lib |
| from chromite.lib import gs |
| from chromite.lib import osutils |
| from chromite.lib import parallel_unittest |
| from chromite.lib import portage_util |
| from chromite.scripts import upload_prebuilts as prebuilt |
| |
| |
| # pylint: disable=protected-access |
| |
| |
| PUBLIC_PACKAGES = [ |
| {"CPV": "gtk+/public1", "SHA1": "1", "MTIME": "1"}, |
| {"CPV": "gtk+/public2", "SHA1": "2", "PATH": "gtk+/foo.tgz", "MTIME": "2"}, |
| ] |
| PRIVATE_PACKAGES = [{"CPV": "private", "SHA1": "3", "MTIME": "3"}] |
| |
| |
| def SimplePackageIndex(header=True, packages=True): |
| pkgindex = binpkg.PackageIndex() |
| if header: |
| pkgindex.header["URI"] = "gs://example" |
| if packages: |
| pkgindex.packages = copy.deepcopy(PUBLIC_PACKAGES + PRIVATE_PACKAGES) |
| return pkgindex |
| |
| |
| class TestUpdateFile(cros_test_lib.TempDirTestCase): |
| """Tests for the UpdateLocalFile function.""" |
| |
| def setUp(self): |
| self.contents_str = [ |
| "# comment that should be skipped", |
| 'PKGDIR="/var/lib/portage/pkgs"', |
| 'PORTAGE_BINHOST="http://no.thanks.com"', |
| "portage portage-20100310.tar.bz2", |
| 'COMPILE_FLAGS="some_value=some_other"', |
| ] |
| self.version_file = os.path.join(self.tempdir, "version") |
| osutils.WriteFile(self.version_file, "\n".join(self.contents_str)) |
| |
| def _read_version_file(self, version_file=None): |
| """Read the contents of self.version_file and return as a list.""" |
| if not version_file: |
| version_file = self.version_file |
| |
| with open(version_file) as version_fh: |
| return [line.strip() for line in version_fh.readlines()] |
| |
| def _verify_key_pair(self, key, val): |
| file_contents = self._read_version_file() |
| |
| # Make sure 'key' entry is only found once. |
| entry_found = False |
| |
| # Ensure value is wrapped on quotes. |
| if '"' not in val: |
| val = '"%s"' % val |
| |
| # Inspect file contents. |
| for entry in file_contents: |
| if "=" not in entry: |
| continue |
| file_key, file_val = entry.split("=", maxsplit=1) |
| if file_key == key: |
| if val == file_val: |
| if entry_found: |
| self.fail(f"Variable {file_key} appears twice") |
| else: |
| entry_found = True |
| |
| if not entry_found: |
| self.fail('Could not find "%s=%s" in version file' % (key, val)) |
| |
| def testAddVariableThatDoesNotExist(self): |
| """Add in a new variable that was no present in the file.""" |
| key = "PORTAGE_BINHOST" |
| value = "1234567" |
| prebuilt.UpdateLocalFile(self.version_file, value) |
| print(self.version_file) |
| self._read_version_file() |
| self._verify_key_pair(key, value) |
| print(self.version_file) |
| |
| def testUpdateVariable(self): |
| """Test updating a variable that already exists.""" |
| binhost_key, binhost_val = self.contents_str[2].split("=") |
| if binhost_key != "PORTAGE_BINHOST": |
| self.fail( |
| "unexpected test input: expected PORTAGE_BINHOST at line[2]" |
| ) |
| self._verify_key_pair(binhost_key, binhost_val) |
| |
| # Confirm that unrelated variable 'PKGDIR' does not change. |
| pkgdir_key, pkgdir_val = self.contents_str[1].split("=") |
| if pkgdir_key != "PKGDIR": |
| self.fail("unexpected test input: expected PKGDIR at line[1]") |
| self._verify_key_pair(pkgdir_key, pkgdir_val) |
| |
| binhost_new_val = "test_update" |
| prebuilt.UpdateLocalFile(self.version_file, binhost_new_val) |
| self._verify_key_pair(binhost_key, binhost_new_val) |
| self._verify_key_pair(pkgdir_key, pkgdir_val) |
| |
| binhost_new_val = "test_update2" |
| prebuilt.UpdateLocalFile(self.version_file, binhost_new_val) |
| self._verify_key_pair(binhost_key, binhost_new_val) |
| self._verify_key_pair(pkgdir_key, pkgdir_val) |
| |
| def testUpdateNonExistentFile(self): |
| key = "PORTAGE_BINHOST" |
| value = "1234567" |
| non_existent_file = tempfile.mktemp() |
| try: |
| prebuilt.UpdateLocalFile(non_existent_file, value) |
| file_contents = self._read_version_file(non_existent_file) |
| self.assertEqual(file_contents, ['%s="%s"' % (key, value)]) |
| finally: |
| if os.path.exists(non_existent_file): |
| os.remove(non_existent_file) |
| |
| |
| class TestPrebuilt(cros_test_lib.MockTestCase): |
| """Tests for Prebuilt logic.""" |
| |
| def setUp(self): |
| self._base_local_path = "/b/cbuild/build/chroot/build/x86-dogfood/" |
| self._gs_bucket_path = "gs://chromeos-prebuilt/host/version" |
| self._local_path = os.path.join(self._base_local_path, "public1.tbz2") |
| |
| def testGenerateUploadDict(self): |
| self.PatchObject(prebuilt.os.path, "exists", return_true=True) |
| pkgs = [{"CPV": "public1"}] |
| result = prebuilt.GenerateUploadDict( |
| self._base_local_path, self._gs_bucket_path, pkgs |
| ) |
| expected = { |
| self._local_path: self._gs_bucket_path + "/public1.tbz2", |
| } |
| self.assertEqual(result, expected) |
| |
| def testGenerateUploadDictWithDebug(self): |
| self.PatchObject(prebuilt.os.path, "exists", return_true=True) |
| pkgs = [{"CPV": "public1", "DEBUG_SYMBOLS": "yes"}] |
| result = prebuilt.GenerateUploadDict( |
| self._base_local_path, self._gs_bucket_path, pkgs |
| ) |
| expected = { |
| self._local_path: self._gs_bucket_path + "/public1.tbz2", |
| self._local_path.replace( |
| ".tbz2", ".debug.tbz2" |
| ): self._gs_bucket_path |
| + "/public1.debug.tbz2", |
| } |
| self.assertEqual(result, expected) |
| |
| def testDeterminePrebuiltConfHost(self): |
| """Test that the host prebuilt path comes back properly.""" |
| expected_path = os.path.join(prebuilt._PREBUILT_MAKE_CONF["amd64"]) |
| self.assertEqual( |
| prebuilt.DeterminePrebuiltConfFile("fake_path", "amd64"), |
| expected_path, |
| ) |
| |
| |
| class TestPkgIndex(cros_test_lib.TestCase): |
| """Helper for tests that update the Packages index file.""" |
| |
| def setUp(self): |
| self.db = {} |
| self.pkgindex = SimplePackageIndex() |
| self.empty = SimplePackageIndex(packages=False) |
| |
| def assertURIs(self, uris): |
| """Verify that the duplicate DB has the specified URLs.""" |
| expected = [v.uri for _, v in sorted(self.db.items())] |
| self.assertEqual(expected, uris) |
| |
| |
| class TestPackagesFileFiltering(TestPkgIndex): |
| """Tests for Packages filtering behavior.""" |
| |
| def testFilterPkgIndex(self): |
| """Test filtering out of private packages.""" |
| self.pkgindex.RemoveFilteredPackages( |
| lambda pkg: pkg in PRIVATE_PACKAGES |
| ) |
| self.assertEqual(self.pkgindex.packages, PUBLIC_PACKAGES) |
| self.assertEqual(self.pkgindex.modified, True) |
| |
| |
| class TestPopulateDuplicateDB(TestPkgIndex): |
| """Tests for the _PopulateDuplicateDB function.""" |
| |
| def testEmptyIndex(self): |
| """Test population of the duplicate DB with an empty index.""" |
| self.empty._PopulateDuplicateDB(self.db, 0) |
| self.assertEqual(self.db, {}) |
| |
| def testNormalIndex(self): |
| """Test population of the duplicate DB with a full index.""" |
| self.pkgindex._PopulateDuplicateDB(self.db, 0) |
| self.assertURIs( |
| [ |
| "gs://example/gtk+/public1.tbz2", |
| "gs://example/gtk+/foo.tgz", |
| "gs://example/private.tbz2", |
| ] |
| ) |
| |
| def testMissingSHA1(self): |
| """Test population of the duplicate DB with a missing SHA1.""" |
| del self.pkgindex.packages[0]["SHA1"] |
| self.pkgindex._PopulateDuplicateDB(self.db, 0) |
| self.assertURIs( |
| ["gs://example/gtk+/foo.tgz", "gs://example/private.tbz2"] |
| ) |
| |
| def testFailedPopulate(self): |
| """Test failure conditions for the populate method.""" |
| headerless = SimplePackageIndex(header=False) |
| self.assertRaises(KeyError, headerless._PopulateDuplicateDB, self.db, 0) |
| del self.pkgindex.packages[0]["CPV"] |
| self.assertRaises( |
| KeyError, self.pkgindex._PopulateDuplicateDB, self.db, 0 |
| ) |
| |
| |
| class TestResolveDuplicateUploads(cros_test_lib.MockTestCase, TestPkgIndex): |
| """Tests for the ResolveDuplicateUploads function.""" |
| |
| def setUp(self): |
| self.PatchObject(binpkg.time, "time", return_value=binpkg.TWO_WEEKS) |
| self.db = {} |
| self.dup = SimplePackageIndex() |
| self.expected_pkgindex = SimplePackageIndex() |
| |
| def assertNoDuplicates(self, candidates): |
| """Verify no duplicates are found with the specified candidates.""" |
| uploads = self.pkgindex.ResolveDuplicateUploads(candidates) |
| self.assertEqual(uploads, self.pkgindex.packages) |
| self.assertEqual( |
| len(self.pkgindex.packages), len(self.expected_pkgindex.packages) |
| ) |
| for pkg1, pkg2 in zip( |
| self.pkgindex.packages, self.expected_pkgindex.packages |
| ): |
| self.assertNotEqual(pkg1["MTIME"], pkg2["MTIME"]) |
| del pkg1["MTIME"] |
| del pkg2["MTIME"] |
| self.assertEqual(self.pkgindex.modified, False) |
| self.assertEqual( |
| self.pkgindex.packages, self.expected_pkgindex.packages |
| ) |
| |
| def assertAllDuplicates(self, candidates): |
| """Verify every package is a duplicate in the specified list.""" |
| for pkg in self.expected_pkgindex.packages: |
| pkg.setdefault("PATH", pkg["CPV"] + ".tbz2") |
| self.pkgindex.ResolveDuplicateUploads(candidates) |
| self.assertEqual( |
| self.pkgindex.packages, self.expected_pkgindex.packages |
| ) |
| |
| def testEmptyList(self): |
| """If no candidates are supplied, no duplicates should be found.""" |
| self.assertNoDuplicates([]) |
| |
| def testEmptyIndex(self): |
| """If no packages are supplied, no duplicates should be found.""" |
| self.assertNoDuplicates([self.empty]) |
| |
| def testDifferentURI(self): |
| """If the URI differs, no duplicates should be found.""" |
| self.dup.header["URI"] = "gs://example2" |
| self.assertNoDuplicates([self.dup]) |
| |
| def testUpdateModificationTime(self): |
| """When duplicates are found, we should use the latest mtime.""" |
| for pkg in self.expected_pkgindex.packages: |
| pkg["MTIME"] = "10" |
| for pkg in self.dup.packages: |
| pkg["MTIME"] = "4" |
| self.assertAllDuplicates([self.expected_pkgindex, self.dup]) |
| |
| def testCanonicalUrl(self): |
| """If the URL is in a different format, we should still find duplicates.""" |
| self.dup.header["URI"] = gs.PUBLIC_BASE_HTTPS_URL + "example" |
| self.assertAllDuplicates([self.dup]) |
| |
| def testMissingSHA1(self): |
| """We should not find duplicates if there is no SHA1.""" |
| del self.pkgindex.packages[0]["SHA1"] |
| del self.expected_pkgindex.packages[0]["SHA1"] |
| for pkg in self.expected_pkgindex.packages[1:]: |
| pkg.setdefault("PATH", pkg["CPV"] + ".tbz2") |
| self.pkgindex.ResolveDuplicateUploads([self.dup]) |
| self.assertNotEqual( |
| self.pkgindex.packages[0]["MTIME"], |
| self.expected_pkgindex.packages[0]["MTIME"], |
| ) |
| del self.pkgindex.packages[0]["MTIME"] |
| del self.expected_pkgindex.packages[0]["MTIME"] |
| self.assertEqual( |
| self.pkgindex.packages, self.expected_pkgindex.packages |
| ) |
| |
| def testSymbolsAvailable(self): |
| """If symbols are available remotely, re-use them and set DEBUG_SYMBOLS.""" |
| self.dup.packages[0]["DEBUG_SYMBOLS"] = "yes" |
| |
| uploads = self.pkgindex.ResolveDuplicateUploads([self.dup]) |
| self.assertEqual(uploads, []) |
| self.assertEqual(self.pkgindex.packages[0].get("DEBUG_SYMBOLS"), "yes") |
| |
| def testSymbolsAvailableLocallyOnly(self): |
| """If the symbols are only available locally, reupload them.""" |
| self.pkgindex.packages[0]["DEBUG_SYMBOLS"] = "yes" |
| |
| uploads = self.pkgindex.ResolveDuplicateUploads([self.dup]) |
| self.assertEqual(uploads, [self.pkgindex.packages[0]]) |
| |
| |
| class TestWritePackageIndex(cros_test_lib.MockTestCase, TestPkgIndex): |
| """Tests for the WriteToNamedTemporaryFile function.""" |
| |
| def testSimple(self): |
| """Test simple call of WriteToNamedTemporaryFile()""" |
| self.PatchObject(self.pkgindex, "Write") |
| f = self.pkgindex.WriteToNamedTemporaryFile() |
| self.assertEqual(f.read(), "") |
| |
| |
| class TestUploadPrebuilt(cros_test_lib.MockTempDirTestCase): |
| """Tests for the _UploadPrebuilt function.""" |
| |
| def setUp(self): |
| class MockTemporaryFile(object): |
| """Mock out the temporary file logic.""" |
| |
| def __init__(self, name): |
| self.name = name |
| |
| self.pkgindex = SimplePackageIndex() |
| self.PatchObject( |
| binpkg, "GrabLocalPackageIndex", return_value=self.pkgindex |
| ) |
| self.PatchObject( |
| self.pkgindex, |
| "ResolveDuplicateUploads", |
| return_value=PRIVATE_PACKAGES, |
| ) |
| self.PatchObject( |
| self.pkgindex, |
| "WriteToNamedTemporaryFile", |
| return_value=MockTemporaryFile("fake"), |
| ) |
| self.remote_up_mock = self.PatchObject(prebuilt, "RemoteUpload") |
| self.gs_up_mock = self.PatchObject(prebuilt, "_GsUpload") |
| |
| def testSuccessfulGsUpload(self): |
| uploads = { |
| os.path.join(self.tempdir, "private.tbz2"): "gs://foo/private.tbz2" |
| } |
| dev_extras = os.path.join(self.tempdir, "dev-only-extras.tar.xz") |
| osutils.Touch(dev_extras) |
| self.PatchObject(prebuilt, "GenerateUploadDict", return_value=uploads) |
| uploads = uploads.copy() |
| uploads["fake"] = "gs://foo/suffix/Packages" |
| uploads[dev_extras] = "gs://foo/suffix/dev-only-extras.tar.xz" |
| acl = "public-read" |
| uri = self.pkgindex.header["URI"] |
| uploader = prebuilt.PrebuiltUploader( |
| "gs://foo", |
| acl, |
| uri, |
| [], |
| "/", |
| [], |
| False, |
| "foo", |
| False, |
| "x86-foo", |
| [], |
| "", |
| report={}, |
| ) |
| uploader._UploadPrebuilt(self.tempdir, "suffix") |
| self.remote_up_mock.assert_called_once_with(mock.ANY, acl, uploads) |
| self.assertTrue(self.gs_up_mock.called) |
| |
| |
| class TestSyncPrebuilts(cros_test_lib.MockTestCase): |
| """Tests for the SyncHostPrebuilts function.""" |
| |
| def setUp(self): |
| clnum = [1] |
| |
| def mock_rev(_filename, _data, report, *_args, **_kwargs): |
| report.setdefault("created_cls", []).append( |
| f"https://crrev.com/unittest/{clnum[0]}" |
| ) |
| clnum[0] += 1 |
| |
| self.rev_mock = self.PatchObject( |
| prebuilt, |
| "RevGitFile", |
| side_effect=mock_rev, |
| ) |
| self.update_binhost_mock = self.PatchObject( |
| prebuilt, "UpdateBinhostConfFile", return_value=None |
| ) |
| self.build_path = "/trunk" |
| self.upload_location = "gs://upload/" |
| self.version = "1" |
| self.binhost = "http://prebuilt/" |
| self.key = "PORTAGE_BINHOST" |
| self.upload_mock = self.PatchObject( |
| prebuilt.PrebuiltUploader, "_UploadPrebuilt", return_value=True |
| ) |
| |
| def _testSyncHostPrebuilts(self, chroot): |
| board = "x86-foo" |
| target = prebuilt.BuildTarget(board, "aura") |
| slave_targets = [prebuilt.BuildTarget("x86-bar", "aura")] |
| report = {} |
| if chroot is None: |
| package_path = os.path.join( |
| self.build_path, "chroot", prebuilt._HOST_PACKAGES_PATH |
| ) |
| else: |
| package_path = os.path.join(chroot, prebuilt._HOST_PACKAGES_PATH) |
| url_suffix = prebuilt._REL_HOST_PATH % { |
| "version": self.version, |
| "host_arch": prebuilt._HOST_ARCH, |
| "target": target, |
| } |
| packages_url_suffix = "%s/packages" % url_suffix.rstrip("/") |
| url_value = "%s/%s/" % ( |
| self.binhost.rstrip("/"), |
| packages_url_suffix.rstrip("/"), |
| ) |
| urls = [url_value.replace("foo", "bar"), url_value] |
| binhost = " ".join(urls) |
| uploader = prebuilt.PrebuiltUploader( |
| self.upload_location, |
| "public-read", |
| self.binhost, |
| [], |
| self.build_path, |
| [], |
| False, |
| "foo", |
| False, |
| target, |
| slave_targets, |
| self.version, |
| report, |
| chroot=chroot, |
| ) |
| uploader.SyncHostPrebuilts(self.key, True, True) |
| self.assertEqual( |
| report, |
| { |
| "created_cls": ["https://crrev.com/unittest/1"], |
| }, |
| ) |
| self.upload_mock.assert_called_once_with( |
| package_path, packages_url_suffix |
| ) |
| self.rev_mock.assert_called_once_with( |
| mock.ANY, {self.key: binhost}, report, dryrun=False |
| ) |
| self.update_binhost_mock.assert_called_once_with( |
| mock.ANY, self.key, binhost |
| ) |
| |
| def testSyncHostPrebuilts(self): |
| self._testSyncHostPrebuilts(chroot=None) |
| |
| def testSyncHostPrebuiltsWithChroot(self): |
| self._testSyncHostPrebuilts("/test/chroot") |
| |
| def testSyncBoardPrebuilts(self): |
| board = "x86-foo" |
| target = prebuilt.BuildTarget(board, "aura") |
| slave_targets = [prebuilt.BuildTarget("x86-bar", "aura")] |
| board_path = os.path.join( |
| self.build_path, prebuilt._BOARD_PATH % {"board": board} |
| ) |
| package_path = os.path.join(board_path, "packages") |
| url_suffix = prebuilt._REL_BOARD_PATH % { |
| "version": self.version, |
| "target": target, |
| } |
| packages_url_suffix = "%s/packages" % url_suffix.rstrip("/") |
| url_value = "%s/%s/" % ( |
| self.binhost.rstrip("/"), |
| packages_url_suffix.rstrip("/"), |
| ) |
| bar_binhost = url_value.replace("foo", "bar") |
| determine_mock = self.PatchObject( |
| prebuilt, "DeterminePrebuiltConfFile", side_effect=("bar", "foo") |
| ) |
| self.PatchObject(prebuilt.PrebuiltUploader, "_UploadSdkTarball") |
| report = {} |
| with parallel_unittest.ParallelMock(): |
| multiprocessing.Process.exitcode = 0 |
| uploader = prebuilt.PrebuiltUploader( |
| self.upload_location, |
| "public-read", |
| self.binhost, |
| [], |
| self.build_path, |
| [], |
| False, |
| "foo", |
| False, |
| target, |
| slave_targets, |
| self.version, |
| report, |
| ) |
| uploader.SyncBoardPrebuilts( |
| self.key, True, True, True, None, None, None, None, None |
| ) |
| determine_mock.assert_has_calls( |
| [ |
| mock.call(self.build_path, slave_targets[0]), |
| mock.call(self.build_path, target), |
| ] |
| ) |
| self.upload_mock.assert_called_once_with( |
| package_path, packages_url_suffix |
| ) |
| self.rev_mock.assert_has_calls( |
| [ |
| mock.call( |
| "bar", |
| {self.key: bar_binhost}, |
| { |
| "created_cls": [ |
| "https://crrev.com/unittest/1", |
| "https://crrev.com/unittest/2", |
| ], |
| }, |
| dryrun=False, |
| ), |
| mock.call( |
| "foo", |
| {self.key: url_value}, |
| { |
| "created_cls": [ |
| "https://crrev.com/unittest/1", |
| "https://crrev.com/unittest/2", |
| ], |
| }, |
| dryrun=False, |
| ), |
| ] |
| ) |
| self.update_binhost_mock.assert_has_calls( |
| [ |
| mock.call(mock.ANY, self.key, bar_binhost), |
| mock.call(mock.ANY, self.key, url_value), |
| ] |
| ) |
| |
| |
| class TestMain(cros_test_lib.MockTestCase): |
| """Tests for the main() function.""" |
| |
| def testMain(self): |
| """Test that the main function works.""" |
| # Use a real object as returned from ParseOptions as a spec for |
| # the mock options object, so that we don't have any properties |
| # that the real object doesn't have. |
| options_spec, _ = prebuilt.ParseOptions( |
| [ |
| "--dry-run", |
| "--build-path", |
| "/trunk", |
| "-u", |
| "gs://upload", |
| ] |
| ) |
| options = mock.MagicMock(spec=options_spec) |
| old_binhost = "http://prebuilt/1" |
| options.previous_binhost_url = [old_binhost] |
| options.board = "x86-foo" |
| options.profile = None |
| target = prebuilt.BuildTarget(options.board, options.profile) |
| options.build_path = "/trunk" |
| options.chroot = None |
| options.dryrun = False |
| options.private = True |
| options.packages = [] |
| options.sync_host = True |
| options.git_sync = True |
| options.upload_board_tarball = True |
| options.prepackaged_tarball = None |
| options.toolchains_overlay_tarballs = [] |
| options.toolchains_overlay_upload_path = "" |
| options.toolchain_tarballs = [] |
| options.toolchain_upload_path = "" |
| options.upload = "gs://upload/" |
| options.binhost_base_url = options.upload |
| options.prepend_version = True |
| options.set_version = None |
| options.skip_upload = False |
| options.filters = True |
| options.key = "PORTAGE_BINHOST" |
| options.binhost_conf_dir = None |
| options.sync_binhost_conf = True |
| options.slave_targets = [prebuilt.BuildTarget("x86-bar", "aura")] |
| self.PatchObject( |
| prebuilt, "ParseOptions", return_value=tuple([options, target]) |
| ) |
| self.PatchObject(binpkg, "GrabRemotePackageIndex", return_value=True) |
| init_mock = self.PatchObject( |
| prebuilt.PrebuiltUploader, "__init__", return_value=None |
| ) |
| expected_gs_acl_path = os.path.join( |
| "/fake_path", prebuilt._GOOGLESTORAGE_GSUTIL_FILE |
| ) |
| self.PatchObject( |
| portage_util, "FindOverlayFile", return_value=expected_gs_acl_path |
| ) |
| host_mock = self.PatchObject( |
| prebuilt.PrebuiltUploader, "SyncHostPrebuilts", return_value=None |
| ) |
| board_mock = self.PatchObject( |
| prebuilt.PrebuiltUploader, "SyncBoardPrebuilts", return_value=None |
| ) |
| |
| prebuilt.main([]) |
| |
| init_mock.assert_called_once_with( |
| options.upload, |
| expected_gs_acl_path, |
| options.upload, |
| mock.ANY, |
| options.build_path, |
| options.packages, |
| False, |
| None, |
| False, |
| target, |
| options.slave_targets, |
| mock.ANY, |
| {}, |
| None, |
| ) |
| board_mock.assert_called_once_with( |
| options.key, |
| options.git_sync, |
| options.sync_binhost_conf, |
| options.upload_board_tarball, |
| None, |
| [], |
| "", |
| [], |
| "", |
| ) |
| host_mock.assert_called_once_with( |
| options.key, options.git_sync, options.sync_binhost_conf |
| ) |
| |
| |
| class TestSdk(cros_test_lib.MockTestCase): |
| """Test logic related to uploading SDK binaries""" |
| |
| def setUp(self): |
| self.PatchObject( |
| prebuilt, |
| "_GsUpload", |
| side_effect=Exception("should not get called"), |
| ) |
| self.PatchObject( |
| prebuilt, |
| "UpdateBinhostConfFile", |
| side_effect=Exception("should not get called"), |
| ) |
| self.write_file_mock = self.PatchObject(osutils, "WriteFile") |
| self.upload_mock = self.PatchObject( |
| prebuilt.PrebuiltUploader, "_Upload" |
| ) |
| |
| self.acl = "magic-acl" |
| |
| # All these args pretty much get ignored. Whee. |
| self.uploader = prebuilt.PrebuiltUploader( |
| "gs://foo", |
| self.acl, |
| "prebuilt", |
| [], |
| "/", |
| [], |
| False, |
| "foo", |
| False, |
| "x86-foo", |
| [], |
| "chroot-1234", |
| report={}, |
| ) |
| |
| def testSdkUpload( |
| self, |
| to_tarballs=(), |
| to_upload_path=None, |
| tc_tarballs=(), |
| tc_upload_path=None, |
| ): |
| """Make sure we can upload just an SDK tarball""" |
| tar = "sdk.tar.xz" |
| ver = "1234" |
| vtar = "cros-sdk-%s.tar.xz" % ver |
| |
| upload_calls = [ |
| mock.call( |
| "%s.Manifest" % tar, "gs://chromiumos-sdk/%s.Manifest" % vtar |
| ), |
| mock.call(tar, "gs://chromiumos-sdk/%s" % vtar), |
| ] |
| for to in to_tarballs: |
| to = to.split(":") |
| upload_calls.append( |
| mock.call( |
| to[1], |
| ("gs://chromiumos-sdk/" + to_upload_path) |
| % {"toolchains": to[0]}, |
| ) |
| ) |
| for tc in tc_tarballs: |
| tc = tc.split(":") |
| upload_calls.append( |
| mock.call( |
| tc[1], |
| ("gs://chromiumos-sdk/" + tc_upload_path) |
| % {"target": tc[0]}, |
| ) |
| ) |
| upload_calls.append( |
| mock.call(mock.ANY, "gs://chromiumos-sdk/cros-sdk-latest.conf") |
| ) |
| |
| self.uploader._UploadSdkTarball( |
| "amd64-host", |
| "", |
| tar, |
| to_tarballs, |
| to_upload_path, |
| tc_tarballs, |
| tc_upload_path, |
| ) |
| self.upload_mock.assert_has_calls(upload_calls) |
| |
| expected_latest_file_contents = f"""\ |
| # The most recent SDK that is tested and ready for use. |
| LATEST_SDK="{ver}" |
| |
| # The most recently built version. New uprev attempts should target this. |
| # Warning: This version may not be tested yet. |
| LATEST_SDK_UPREV_TARGET=\"None\"""" |
| self.write_file_mock.assert_any_call( |
| mock.ANY, expected_latest_file_contents |
| ) |
| |
| def testBoardOverlayTarballUpload(self): |
| """Make sure processing of board-specific overlay tarballs works.""" |
| to_tarballs = ( |
| ( |
| "i686-pc-linux-gnu:" |
| "/some/path/built-sdk-overlay-toolchains-i686-pc-linux-gnu.tar.xz" |
| ), |
| ( |
| "armv7a-cros-linux-gnueabi-arm-none-eabi:" |
| "/some/path/built-sdk-overlay-toolchains-armv7a-cros-linux-gnueabi-" |
| "arm-none-eabi" |
| ), |
| ) |
| to_upload_path = "1994/04/cros-sdk-overlay-toolchains-%(toolchains)s-1994.04.02.tar.xz" |
| self.testSdkUpload( |
| to_tarballs=to_tarballs, to_upload_path=to_upload_path |
| ) |
| |
| def testToolchainTarballUpload(self): |
| """Make sure processing of toolchain tarballs works.""" |
| tc_tarballs = ( |
| "i686:/some/i686.tar.xz", |
| "arm-none:/some/arm.tar.xz", |
| ) |
| tc_upload_path = "1994/04/%(target)s-1994.04.02.tar.xz" |
| self.testSdkUpload( |
| tc_tarballs=tc_tarballs, tc_upload_path=tc_upload_path |
| ) |