Upstream version 10.39.225.0
[platform/framework/web/crosswalk.git] / src / third_party / chromite / scripts / upload_prebuilts_unittest.py
1 #!/usr/bin/python
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.
5
6 """Unittests for upload_prebuilts.py."""
7
8 from __future__ import print_function
9
10 import copy
11 import mox
12 import os
13 import multiprocessing
14 import sys
15 import tempfile
16
17 sys.path.insert(0, os.path.join(os.path.dirname(os.path.realpath(__file__)),
18                                 '..', '..'))
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
24
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'}]
30
31
32 def SimplePackageIndex(header=True, packages=True):
33   pkgindex = binpkg.PackageIndex()
34   if header:
35     pkgindex.header['URI'] = 'gs://example'
36   if packages:
37     pkgindex.packages = copy.deepcopy(PUBLIC_PACKAGES + PRIVATE_PACKAGES)
38   return pkgindex
39
40
41 class TestUpdateFile(cros_test_lib.TempDirTestCase):
42   """Tests for the UpdateLocalFile function."""
43
44   def setUp(self):
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"',
50                          ]
51     self.version_file = os.path.join(self.tempdir, 'version')
52     osutils.WriteFile(self.version_file, '\n'.join(self.contents_str))
53
54   def _read_version_file(self, version_file=None):
55     """Read the contents of self.version_file and return as a list."""
56     if not version_file:
57       version_file = self.version_file
58
59     version_fh = open(version_file)
60     try:
61       return [line.strip() for line in version_fh.readlines()]
62     finally:
63       version_fh.close()
64
65   def _verify_key_pair(self, key, val):
66     file_contents = self._read_version_file()
67     # ensure key for verify is wrapped on quotes
68     if '"' not in val:
69       val = '"%s"' % val
70     for entry in file_contents:
71       if '=' not in entry:
72         continue
73       file_key, file_val = entry.split('=')
74       if file_key == key:
75         if val == file_val:
76           break
77     else:
78       self.fail('Could not find "%s=%s" in version file' % (key, val))
79
80   def testAddVariableThatDoesNotExist(self):
81     """Add in a new variable that was no present in the file."""
82     key = 'PORTAGE_BINHOST'
83     value = '1234567'
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)
89
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)
97
98   def testUpdateNonExistentFile(self):
99     key = 'PORTAGE_BINHOST'
100     value = '1234567'
101     non_existent_file = tempfile.mktemp()
102     try:
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)])
106     finally:
107       if os.path.exists(non_existent_file):
108         os.remove(non_existent_file)
109
110
111 class TestPrebuilt(cros_test_lib.MoxTestCase):
112   """Tests for Prebuilt logic."""
113
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)
120     self.mox.ReplayAll()
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)
125
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'),
130                      expected_path)
131
132
133 class TestPkgIndex(cros_test_lib.TestCase):
134   """Helper for tests that update the Packages index file."""
135
136   def setUp(self):
137     self.db = {}
138     self.pkgindex = SimplePackageIndex()
139     self.empty = SimplePackageIndex(packages=False)
140
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)
145
146
147 class TestPackagesFileFiltering(TestPkgIndex):
148   """Tests for Packages filtering behavior."""
149
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)
155
156
157 class TestPopulateDuplicateDB(TestPkgIndex):
158   """Tests for the _PopulateDuplicateDB function."""
159
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, {})
164
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'])
171
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'])
178
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)
185
186
187 class TestResolveDuplicateUploads(cros_test_lib.MoxTestCase, TestPkgIndex):
188   """Tests for the ResolveDuplicateUploads function."""
189
190   def setUp(self):
191     self.mox.StubOutWithMock(binpkg.time, 'time')
192     binpkg.time.time().AndReturn(binpkg.TWO_WEEKS)
193     self.mox.ReplayAll()
194     self.db = {}
195     self.dup = SimplePackageIndex()
196     self.expected_pkgindex = SimplePackageIndex()
197
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'])
207       del pkg1['MTIME']
208       del pkg2['MTIME']
209     self.assertEqual(self.pkgindex.modified, False)
210     self.assertEqual(self.pkgindex.packages, self.expected_pkgindex.packages)
211
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)
218
219   def testEmptyList(self):
220     """If no candidates are supplied, no duplicates should be found."""
221     self.assertNoDuplicates([])
222
223   def testEmptyIndex(self):
224     """If no packages are supplied, no duplicates should be found."""
225     self.assertNoDuplicates([self.empty])
226
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])
231
232   def testUpdateModificationTime(self):
233     """When duplicates are found, we should use the latest mtime."""
234     for pkg in self.expected_pkgindex.packages:
235       pkg['MTIME'] = '10'
236     for pkg in self.dup.packages:
237       pkg['MTIME'] = '4'
238     self.assertAllDuplicates([self.expected_pkgindex, self.dup])
239
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])
244
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)
257
258
259 class TestWritePackageIndex(cros_test_lib.MoxTestCase, TestPkgIndex):
260   """Tests for the WriteToNamedTemporaryFile function."""
261
262   def testSimple(self):
263     """Test simple call of WriteToNamedTemporaryFile()"""
264     self.mox.StubOutWithMock(self.pkgindex, 'Write')
265     self.pkgindex.Write(mox.IgnoreArg())
266     self.mox.ReplayAll()
267     f = self.pkgindex.WriteToNamedTemporaryFile()
268     self.assertEqual(f.read(), '')
269
270
271 class TestUploadPrebuilt(cros_test_lib.MoxTestCase):
272   """Tests for the _UploadPrebuilt function."""
273
274   def setUp(self):
275     class MockTemporaryFile(object):
276       """Mock out the temporary file logic."""
277       def __init__(self, name):
278         self.name = 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)
289
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'
297     acl = 'public-read'
298     prebuilt.RemoteUpload(mox.IgnoreArg(), acl, uploads)
299     prebuilt._GsUpload(mox.IgnoreArg(), mox.IgnoreArg(),
300                        mox.IgnoreArg(), mox.IgnoreArg())
301     self.mox.ReplayAll()
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')
306
307
308 class TestSyncPrebuilts(cros_test_lib.MoxTestCase):
309   """Tests for the SyncHostPrebuilts function."""
310
311   def setUp(self):
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/'
317     self.version = '1'
318     self.binhost = 'http://prebuilt/'
319     self.key = 'PORTAGE_BINHOST'
320     self.mox.StubOutWithMock(prebuilt.PrebuiltUploader, '_UploadPrebuilt')
321
322   def testSyncHostPrebuilts(self):
323     board = 'x86-foo'
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)
339     self.mox.ReplayAll()
340     uploader = prebuilt.PrebuiltUploader(
341         self.upload_location, 'public-read', self.binhost, [],
342         self.build_path, [], False, 'foo', False, target, slave_targets,
343         self.version)
344     uploader.SyncHostPrebuilts(self.key, True, True)
345
346   def testSyncBoardPrebuilts(self):
347     board = 'x86-foo'
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,
354         'target': target}
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)
378     self.mox.ReplayAll()
379     uploader = prebuilt.PrebuiltUploader(
380         self.upload_location, 'public-read', self.binhost, [],
381         self.build_path, [], False, 'foo', False, target, slave_targets,
382         self.version)
383     uploader.SyncBoardPrebuilts(self.key, True, True, True, None, None, None)
384
385
386 class TestMain(cros_test_lib.MoxTestCase):
387   """Tests for the main() function."""
388
389   def testMain(self):
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,
433                                        mox.IgnoreArg())
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, [], '')
441     self.mox.ReplayAll()
442     prebuilt.main([])
443
444
445 class TestSdk(cros_test_lib.MoxTestCase):
446   """Test logic related to uploading SDK binaries"""
447
448   def setUp(self):
449     self.mox.StubOutWithMock(prebuilt, '_GsUpload')
450     self.mox.StubOutWithMock(prebuilt, 'UpdateBinhostConfFile')
451
452     self.acl = 'magic-acl'
453
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')
458
459   def testSdkUpload(self, cb=lambda:None, tc_tarballs=(),
460                     tc_upload_path=None):
461     """Make sure we can upload just an SDK tarball"""
462     tar = 'sdk.tar.xz'
463     ver = '1234'
464     vtar = 'cros-sdk-%s.tar.xz' % ver
465
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)
470     cb()
471     prebuilt._GsUpload(mox.IgnoreArg(), self.acl, mox.IgnoreArg(),
472                        'gs://chromiumos-sdk/cros-sdk-latest.conf')
473     self.mox.ReplayAll()
474
475     self.uploader._UploadSdkTarball('amd64-host', '',
476                                     tar, tc_tarballs, tc_upload_path)
477
478   def testTarballUpload(self):
479     """Make sure processing of toolchain tarballs works"""
480     tc_tarballs = (
481         'i686:/some/i686.tar.xz',
482         'arm-none:/some/arm.tar.xz',
483     )
484     tc_upload_path = '1994/04/%(target)s-1994.04.02.tar.xz'
485     def cb():
486       for tc in tc_tarballs:
487         tc = tc.split(':')
488         prebuilt._GsUpload(
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)
492
493
494 if __name__ == '__main__':
495   cros_test_lib.main()