2 # Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
3 # Use of this source code is governed by a BSD-style license that can be
4 # found in the LICENSE file.
6 """Unittests for upload_prebuilts.py."""
8 from __future__ import print_function
13 import multiprocessing
17 sys.path.insert(0, os.path.join(os.path.dirname(os.path.realpath(__file__)),
19 from chromite.scripts import upload_prebuilts as prebuilt
20 from chromite.lib import cros_test_lib
21 from chromite.lib import gs
22 from chromite.lib import binpkg
23 from chromite.lib import osutils
25 # pylint: disable=E1120,W0212,R0904
26 PUBLIC_PACKAGES = [{'CPV': 'gtk+/public1', 'SHA1': '1', 'MTIME': '1'},
27 {'CPV': 'gtk+/public2', 'SHA1': '2',
28 'PATH': 'gtk+/foo.tgz', 'MTIME': '2'}]
29 PRIVATE_PACKAGES = [{'CPV': 'private', 'SHA1': '3', 'MTIME': '3'}]
32 def SimplePackageIndex(header=True, packages=True):
33 pkgindex = binpkg.PackageIndex()
35 pkgindex.header['URI'] = 'gs://example'
37 pkgindex.packages = copy.deepcopy(PUBLIC_PACKAGES + PRIVATE_PACKAGES)
41 class TestUpdateFile(cros_test_lib.TempDirTestCase):
42 """Tests for the UpdateLocalFile function."""
45 self.contents_str = ['# comment that should be skipped',
46 'PKGDIR="/var/lib/portage/pkgs"',
47 'PORTAGE_BINHOST="http://no.thanks.com"',
48 'portage portage-20100310.tar.bz2',
49 'COMPILE_FLAGS="some_value=some_other"',
51 self.version_file = os.path.join(self.tempdir, 'version')
52 osutils.WriteFile(self.version_file, '\n'.join(self.contents_str))
54 def _read_version_file(self, version_file=None):
55 """Read the contents of self.version_file and return as a list."""
57 version_file = self.version_file
59 version_fh = open(version_file)
61 return [line.strip() for line in version_fh.readlines()]
65 def _verify_key_pair(self, key, val):
66 file_contents = self._read_version_file()
67 # ensure key for verify is wrapped on quotes
70 for entry in file_contents:
73 file_key, file_val = entry.split('=')
78 self.fail('Could not find "%s=%s" in version file' % (key, val))
80 def testAddVariableThatDoesNotExist(self):
81 """Add in a new variable that was no present in the file."""
82 key = 'PORTAGE_BINHOST'
84 prebuilt.UpdateLocalFile(self.version_file, value)
85 print(self.version_file)
86 self._read_version_file()
87 self._verify_key_pair(key, value)
88 print(self.version_file)
90 def testUpdateVariable(self):
91 """Test updating a variable that already exists."""
92 key, val = self.contents_str[2].split('=')
93 new_val = 'test_update'
94 self._verify_key_pair(key, val)
95 prebuilt.UpdateLocalFile(self.version_file, new_val)
96 self._verify_key_pair(key, new_val)
98 def testUpdateNonExistentFile(self):
99 key = 'PORTAGE_BINHOST'
101 non_existent_file = tempfile.mktemp()
103 prebuilt.UpdateLocalFile(non_existent_file, value)
104 file_contents = self._read_version_file(non_existent_file)
105 self.assertEqual(file_contents, ['%s="%s"' % (key, value)])
107 if os.path.exists(non_existent_file):
108 os.remove(non_existent_file)
111 class TestPrebuilt(cros_test_lib.MoxTestCase):
112 """Tests for Prebuilt logic."""
114 def testGenerateUploadDict(self):
115 base_local_path = '/b/cbuild/build/chroot/build/x86-dogfood/'
116 gs_bucket_path = 'gs://chromeos-prebuilt/host/version'
117 local_path = os.path.join(base_local_path, 'public1.tbz2')
118 self.mox.StubOutWithMock(prebuilt.os.path, 'exists')
119 prebuilt.os.path.exists(local_path).AndReturn(True)
121 pkgs = [{ 'CPV': 'public1' }]
122 result = prebuilt.GenerateUploadDict(base_local_path, gs_bucket_path, pkgs)
123 expected = { local_path: gs_bucket_path + '/public1.tbz2' }
124 self.assertEqual(result, expected)
126 def testDeterminePrebuiltConfHost(self):
127 """Test that the host prebuilt path comes back properly."""
128 expected_path = os.path.join(prebuilt._PREBUILT_MAKE_CONF['amd64'])
129 self.assertEqual(prebuilt.DeterminePrebuiltConfFile('fake_path', 'amd64'),
133 class TestPkgIndex(cros_test_lib.TestCase):
134 """Helper for tests that update the Packages index file."""
138 self.pkgindex = SimplePackageIndex()
139 self.empty = SimplePackageIndex(packages=False)
141 def assertURIs(self, uris):
142 """Verify that the duplicate DB has the specified URLs."""
143 expected = [v.uri for _, v in sorted(self.db.items())]
144 self.assertEqual(expected, uris)
147 class TestPackagesFileFiltering(TestPkgIndex):
148 """Tests for Packages filtering behavior."""
150 def testFilterPkgIndex(self):
151 """Test filtering out of private packages."""
152 self.pkgindex.RemoveFilteredPackages(lambda pkg: pkg in PRIVATE_PACKAGES)
153 self.assertEqual(self.pkgindex.packages, PUBLIC_PACKAGES)
154 self.assertEqual(self.pkgindex.modified, True)
157 class TestPopulateDuplicateDB(TestPkgIndex):
158 """Tests for the _PopulateDuplicateDB function."""
160 def testEmptyIndex(self):
161 """Test population of the duplicate DB with an empty index."""
162 self.empty._PopulateDuplicateDB(self.db, 0)
163 self.assertEqual(self.db, {})
165 def testNormalIndex(self):
166 """Test population of the duplicate DB with a full index."""
167 self.pkgindex._PopulateDuplicateDB(self.db, 0)
168 self.assertURIs(['gs://example/gtk+/public1.tbz2',
169 'gs://example/gtk+/foo.tgz',
170 'gs://example/private.tbz2'])
172 def testMissingSHA1(self):
173 """Test population of the duplicate DB with a missing SHA1."""
174 del self.pkgindex.packages[0]['SHA1']
175 self.pkgindex._PopulateDuplicateDB(self.db, 0)
176 self.assertURIs(['gs://example/gtk+/foo.tgz',
177 'gs://example/private.tbz2'])
179 def testFailedPopulate(self):
180 """Test failure conditions for the populate method."""
181 headerless = SimplePackageIndex(header=False)
182 self.assertRaises(KeyError, headerless._PopulateDuplicateDB, self.db, 0)
183 del self.pkgindex.packages[0]['CPV']
184 self.assertRaises(KeyError, self.pkgindex._PopulateDuplicateDB, self.db, 0)
187 class TestResolveDuplicateUploads(cros_test_lib.MoxTestCase, TestPkgIndex):
188 """Tests for the ResolveDuplicateUploads function."""
191 self.mox.StubOutWithMock(binpkg.time, 'time')
192 binpkg.time.time().AndReturn(binpkg.TWO_WEEKS)
195 self.dup = SimplePackageIndex()
196 self.expected_pkgindex = SimplePackageIndex()
198 def assertNoDuplicates(self, candidates):
199 """Verify no duplicates are found with the specified candidates."""
200 uploads = self.pkgindex.ResolveDuplicateUploads(candidates)
201 self.assertEqual(uploads, self.pkgindex.packages)
202 self.assertEqual(len(self.pkgindex.packages),
203 len(self.expected_pkgindex.packages))
204 for pkg1, pkg2 in zip(self.pkgindex.packages,
205 self.expected_pkgindex.packages):
206 self.assertNotEqual(pkg1['MTIME'], pkg2['MTIME'])
209 self.assertEqual(self.pkgindex.modified, False)
210 self.assertEqual(self.pkgindex.packages, self.expected_pkgindex.packages)
212 def assertAllDuplicates(self, candidates):
213 """Verify every package is a duplicate in the specified list."""
214 for pkg in self.expected_pkgindex.packages:
215 pkg.setdefault('PATH', pkg['CPV'] + '.tbz2')
216 self.pkgindex.ResolveDuplicateUploads(candidates)
217 self.assertEqual(self.pkgindex.packages, self.expected_pkgindex.packages)
219 def testEmptyList(self):
220 """If no candidates are supplied, no duplicates should be found."""
221 self.assertNoDuplicates([])
223 def testEmptyIndex(self):
224 """If no packages are supplied, no duplicates should be found."""
225 self.assertNoDuplicates([self.empty])
227 def testDifferentURI(self):
228 """If the URI differs, no duplicates should be found."""
229 self.dup.header['URI'] = 'gs://example2'
230 self.assertNoDuplicates([self.dup])
232 def testUpdateModificationTime(self):
233 """When duplicates are found, we should use the latest mtime."""
234 for pkg in self.expected_pkgindex.packages:
236 for pkg in self.dup.packages:
238 self.assertAllDuplicates([self.expected_pkgindex, self.dup])
240 def testCanonicalUrl(self):
241 """If the URL is in a different format, we should still find duplicates."""
242 self.dup.header['URI'] = gs.PUBLIC_BASE_HTTPS_URL + 'example'
243 self.assertAllDuplicates([self.dup])
245 def testMissingSHA1(self):
246 """We should not find duplicates if there is no SHA1."""
247 del self.pkgindex.packages[0]['SHA1']
248 del self.expected_pkgindex.packages[0]['SHA1']
249 for pkg in self.expected_pkgindex.packages[1:]:
250 pkg.setdefault('PATH', pkg['CPV'] + '.tbz2')
251 self.pkgindex.ResolveDuplicateUploads([self.dup])
252 self.assertNotEqual(self.pkgindex.packages[0]['MTIME'],
253 self.expected_pkgindex.packages[0]['MTIME'])
254 del self.pkgindex.packages[0]['MTIME']
255 del self.expected_pkgindex.packages[0]['MTIME']
256 self.assertEqual(self.pkgindex.packages, self.expected_pkgindex.packages)
259 class TestWritePackageIndex(cros_test_lib.MoxTestCase, TestPkgIndex):
260 """Tests for the WriteToNamedTemporaryFile function."""
262 def testSimple(self):
263 """Test simple call of WriteToNamedTemporaryFile()"""
264 self.mox.StubOutWithMock(self.pkgindex, 'Write')
265 self.pkgindex.Write(mox.IgnoreArg())
267 f = self.pkgindex.WriteToNamedTemporaryFile()
268 self.assertEqual(f.read(), '')
271 class TestUploadPrebuilt(cros_test_lib.MoxTestCase):
272 """Tests for the _UploadPrebuilt function."""
275 class MockTemporaryFile(object):
276 """Mock out the temporary file logic."""
277 def __init__(self, name):
279 self.pkgindex = SimplePackageIndex()
280 self.mox.StubOutWithMock(binpkg, 'GrabLocalPackageIndex')
281 binpkg.GrabLocalPackageIndex('/packages').AndReturn(self.pkgindex)
282 self.mox.StubOutWithMock(prebuilt, 'RemoteUpload')
283 self.mox.StubOutWithMock(self.pkgindex, 'ResolveDuplicateUploads')
284 self.pkgindex.ResolveDuplicateUploads([]).AndReturn(PRIVATE_PACKAGES)
285 self.mox.StubOutWithMock(self.pkgindex, 'WriteToNamedTemporaryFile')
286 self.mox.StubOutWithMock(prebuilt, '_GsUpload')
287 fake_pkgs_file = MockTemporaryFile('fake')
288 self.pkgindex.WriteToNamedTemporaryFile().AndReturn(fake_pkgs_file)
290 def testSuccessfulGsUpload(self):
291 uploads = {'/packages/private.tbz2': 'gs://foo/private.tbz2'}
292 self.mox.StubOutWithMock(prebuilt, 'GenerateUploadDict')
293 prebuilt.GenerateUploadDict('/packages', 'gs://foo/suffix',
294 PRIVATE_PACKAGES).AndReturn(uploads)
295 uploads = uploads.copy()
296 uploads['fake'] = 'gs://foo/suffix/Packages'
298 prebuilt.RemoteUpload(mox.IgnoreArg(), acl, uploads)
299 prebuilt._GsUpload(mox.IgnoreArg(), mox.IgnoreArg(),
300 mox.IgnoreArg(), mox.IgnoreArg())
302 uri = self.pkgindex.header['URI']
303 uploader = prebuilt.PrebuiltUploader('gs://foo', acl, uri, [], '/', [],
304 False, 'foo', False, 'x86-foo', [], '')
305 uploader._UploadPrebuilt('/packages', 'suffix')
308 class TestSyncPrebuilts(cros_test_lib.MoxTestCase):
309 """Tests for the SyncHostPrebuilts function."""
312 self.mox.StubOutWithMock(prebuilt, 'DeterminePrebuiltConfFile')
313 self.mox.StubOutWithMock(prebuilt, 'RevGitFile')
314 self.mox.StubOutWithMock(prebuilt, 'UpdateBinhostConfFile')
315 self.build_path = '/trunk'
316 self.upload_location = 'gs://upload/'
318 self.binhost = 'http://prebuilt/'
319 self.key = 'PORTAGE_BINHOST'
320 self.mox.StubOutWithMock(prebuilt.PrebuiltUploader, '_UploadPrebuilt')
322 def testSyncHostPrebuilts(self):
324 target = prebuilt.BuildTarget(board, 'aura')
325 slave_targets = [prebuilt.BuildTarget('x86-bar', 'aura')]
326 package_path = os.path.join(self.build_path,
327 prebuilt._HOST_PACKAGES_PATH)
328 url_suffix = prebuilt._REL_HOST_PATH % {'version': self.version,
329 'host_arch': prebuilt._HOST_ARCH, 'target': target}
330 packages_url_suffix = '%s/packages' % url_suffix.rstrip('/')
331 prebuilt.PrebuiltUploader._UploadPrebuilt(package_path,
332 packages_url_suffix).AndReturn(True)
333 url_value = '%s/%s/' % (self.binhost.rstrip('/'),
334 packages_url_suffix.rstrip('/'))
335 urls = [url_value.replace('foo', 'bar'), url_value]
336 binhost = ' '.join(urls)
337 prebuilt.RevGitFile(mox.IgnoreArg(), {self.key: binhost}, dryrun=False)
338 prebuilt.UpdateBinhostConfFile(mox.IgnoreArg(), self.key, binhost)
340 uploader = prebuilt.PrebuiltUploader(
341 self.upload_location, 'public-read', self.binhost, [],
342 self.build_path, [], False, 'foo', False, target, slave_targets,
344 uploader.SyncHostPrebuilts(self.key, True, True)
346 def testSyncBoardPrebuilts(self):
348 target = prebuilt.BuildTarget(board, 'aura')
349 slave_targets = [prebuilt.BuildTarget('x86-bar', 'aura')]
350 board_path = os.path.join(
351 self.build_path, prebuilt._BOARD_PATH % {'board': board})
352 package_path = os.path.join(board_path, 'packages')
353 url_suffix = prebuilt._REL_BOARD_PATH % {'version': self.version,
355 packages_url_suffix = '%s/packages' % url_suffix.rstrip('/')
356 self.mox.StubOutWithMock(multiprocessing.Process, '__init__')
357 self.mox.StubOutWithMock(multiprocessing.Process, 'exitcode')
358 self.mox.StubOutWithMock(multiprocessing.Process, 'start')
359 self.mox.StubOutWithMock(multiprocessing.Process, 'join')
360 multiprocessing.Process.__init__(
361 target=mox.IgnoreArg(),
362 args=(board_path, url_suffix, None, None, None))
363 multiprocessing.Process.start()
364 prebuilt.PrebuiltUploader._UploadPrebuilt(
365 package_path, packages_url_suffix).AndReturn(True)
366 multiprocessing.Process.join()
367 multiprocessing.Process.exitcode = 0
368 url_value = '%s/%s/' % (self.binhost.rstrip('/'),
369 packages_url_suffix.rstrip('/'))
370 bar_binhost = url_value.replace('foo', 'bar')
371 prebuilt.DeterminePrebuiltConfFile(
372 self.build_path, slave_targets[0]).AndReturn('bar')
373 prebuilt.RevGitFile('bar', {self.key: bar_binhost}, dryrun=False)
374 prebuilt.UpdateBinhostConfFile(mox.IgnoreArg(), self.key, bar_binhost)
375 prebuilt.DeterminePrebuiltConfFile(self.build_path, target).AndReturn('foo')
376 prebuilt.RevGitFile('foo', {self.key: url_value}, dryrun=False)
377 prebuilt.UpdateBinhostConfFile(mox.IgnoreArg(), self.key, url_value)
379 uploader = prebuilt.PrebuiltUploader(
380 self.upload_location, 'public-read', self.binhost, [],
381 self.build_path, [], False, 'foo', False, target, slave_targets,
383 uploader.SyncBoardPrebuilts(self.key, True, True, True, None, None, None)
386 class TestMain(cros_test_lib.MoxTestCase):
387 """Tests for the main() function."""
390 """Test that the main function works."""
391 options = mox.MockObject(object)
392 old_binhost = 'http://prebuilt/1'
393 options.previous_binhost_url = [old_binhost]
394 options.board = 'x86-foo'
395 options.profile = None
396 target = prebuilt.BuildTarget(options.board, options.profile)
397 options.build_path = '/trunk'
398 options.dryrun = False
399 options.private = True
400 options.packages = []
401 options.sync_host = True
402 options.git_sync = True
403 options.upload_board_tarball = True
404 options.prepackaged_tarball = None
405 options.toolchain_tarballs = []
406 options.toolchain_upload_path = ''
407 options.upload = 'gs://upload/'
408 options.binhost_base_url = options.upload
409 options.prepend_version = True
410 options.set_version = None
411 options.skip_upload = False
412 options.filters = True
413 options.key = 'PORTAGE_BINHOST'
414 options.binhost_conf_dir = 'foo'
415 options.sync_binhost_conf = True
416 options.slave_targets = [prebuilt.BuildTarget('x86-bar', 'aura')]
417 self.mox.StubOutWithMock(prebuilt, 'ParseOptions')
418 prebuilt.ParseOptions([]).AndReturn(tuple([options, target]))
419 self.mox.StubOutWithMock(binpkg, 'GrabRemotePackageIndex')
420 binpkg.GrabRemotePackageIndex(old_binhost).AndReturn(True)
421 self.mox.StubOutWithMock(prebuilt.PrebuiltUploader, '__init__')
422 self.mox.StubOutWithMock(prebuilt, 'GetBoardOverlay')
423 fake_overlay_path = '/fake_path'
424 prebuilt.GetBoardOverlay(
425 options.build_path, options.board).AndReturn(fake_overlay_path)
426 expected_gs_acl_path = os.path.join(fake_overlay_path,
427 prebuilt._GOOGLESTORAGE_ACL_FILE)
428 prebuilt.PrebuiltUploader.__init__(options.upload, expected_gs_acl_path,
429 options.upload, mox.IgnoreArg(),
430 options.build_path, options.packages,
431 False, options.binhost_conf_dir, False,
432 target, options.slave_targets,
434 self.mox.StubOutWithMock(prebuilt.PrebuiltUploader, 'SyncHostPrebuilts')
435 prebuilt.PrebuiltUploader.SyncHostPrebuilts(
436 options.key, options.git_sync, options.sync_binhost_conf)
437 self.mox.StubOutWithMock(prebuilt.PrebuiltUploader, 'SyncBoardPrebuilts')
438 prebuilt.PrebuiltUploader.SyncBoardPrebuilts(
439 options.key, options.git_sync,
440 options.sync_binhost_conf, options.upload_board_tarball, None, [], '')
445 class TestSdk(cros_test_lib.MoxTestCase):
446 """Test logic related to uploading SDK binaries"""
449 self.mox.StubOutWithMock(prebuilt, '_GsUpload')
450 self.mox.StubOutWithMock(prebuilt, 'UpdateBinhostConfFile')
452 self.acl = 'magic-acl'
454 # All these args pretty much get ignored. Whee.
455 self.uploader = prebuilt.PrebuiltUploader(
456 'gs://foo', self.acl, 'prebuilt', [], '/', [],
457 False, 'foo', False, 'x86-foo', [], 'chroot-1234')
459 def testSdkUpload(self, cb=lambda:None, tc_tarballs=(),
460 tc_upload_path=None):
461 """Make sure we can upload just an SDK tarball"""
464 vtar = 'cros-sdk-%s.tar.xz' % ver
466 prebuilt._GsUpload(mox.IgnoreArg(), self.acl, '%s.Manifest' % tar,
467 'gs://chromiumos-sdk/%s.Manifest' % vtar)
468 prebuilt._GsUpload(mox.IgnoreArg(), self.acl, tar,
469 'gs://chromiumos-sdk/%s' % vtar)
471 prebuilt._GsUpload(mox.IgnoreArg(), self.acl, mox.IgnoreArg(),
472 'gs://chromiumos-sdk/cros-sdk-latest.conf')
475 self.uploader._UploadSdkTarball('amd64-host', '',
476 tar, tc_tarballs, tc_upload_path)
478 def testTarballUpload(self):
479 """Make sure processing of toolchain tarballs works"""
481 'i686:/some/i686.tar.xz',
482 'arm-none:/some/arm.tar.xz',
484 tc_upload_path = '1994/04/%(target)s-1994.04.02.tar.xz'
486 for tc in tc_tarballs:
489 mox.IgnoreArg(), self.acl, tc[1],
490 ('gs://chromiumos-sdk/' + tc_upload_path) % {'target': tc[0]})
491 self.testSdkUpload(cb, tc_tarballs, tc_upload_path)
494 if __name__ == '__main__':