src/third_party/chromite/scripts/autotest_quickmerge_unittest.py

   1 #!/usr/bin/python
   2 # Copyright (c) 2013 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 """Unit tests for autotest_quickmerge."""
   7
   8 from __future__ import print_function
   9
  10 import os
  11 import sys
  12 import types
  13 import unittest
  14 import mox
  15
  16
  17 sys.path.insert(0, os.path.abspath('%s/../..' % os.path.dirname(__file__)))
  18 from chromite.lib import cros_build_lib_unittest
  19 from chromite.lib import cros_test_lib
  20 from chromite.scripts import autotest_quickmerge
  21
  22
  23 RSYNC_TEST_OUTPUT = """.d..t...... ./
  24 >f..t...... touched file with spaces
  25 >f..t...... touched_file
  26 >f.st...... modified_contents_file
  27 .f...p..... modified_permissions_file
  28 .f....o.... modified_owner_file
  29 >f+++++++++ new_file
  30 cL+++++++++ new_symlink -> directory_a/new_file_in_directory
  31 .d..t...... directory_a/
  32 >f+++++++++ directory_a/new_file_in_directory
  33 >f..t...... directory_a/touched_file_in_directory
  34 cd+++++++++ new_empty_directory/
  35 .d..t...... touched_empty_directory/"""
  36 # The output format of rsync's itemized changes has a few unusual cases
  37 # that are ambiguous. For instance, if the operation involved creating a
  38 # symbolic link named "a -> b" to a file named "c", the rsync output would be:
  39 # cL+++++++++ a -> b -> c
  40 # which is indistinguishable from the output for creating a symbolic link named
  41 # "a" to a file named "b -> c".
  42 # Since there is no easy resolution to this ambiguity, and it seems like a case
  43 # that would rarely or never be encountered in the wild, rsync quickmerge
  44 # will exclude all files which contain the substring " -> " in their name.
  45
  46 RSYNC_TEST_OUTPUT_FOR_PACKAGE_UPDATE = \
  47 """>f..t...... client/ardvark.py
  48 .d..t...... client/site_tests/
  49 >f+++++++++ client/site_tests/nothing.py
  50 .d..t...... client/site_tests/factory_Leds/
  51 >f+++++++++ client/site_tests/factory_Leds/factory_Leds2.py
  52 >f..tpog... client/site_tests/login_UserPolicyKeys/control
  53 >f..tpog... client/site_tests/login_UserPolicyKeys/login_UserPolicyKeys.py
  54 >f..t...... client/site_tests/platform_Cryptohome/platform_Cryptohome.py
  55 >f..tpog... server/site_tests/security_DbusFuzzServer/control
  56 >f..t.og... utils/coverage_suite.py
  57 .d..t...... client/site_tests/power_Thermal/
  58 cd+++++++++ client/site_tests/power_Thermal/a/
  59 cd+++++++++ client/site_tests/power_Thermal/a/b/
  60 cd+++++++++ client/site_tests/power_Thermal/a/b/c/
  61 >f+++++++++ client/site_tests/power_Thermal/a/b/c/d.py"""
  62
  63 RSYNC_TEST_DESTINATION_PATH = '/foo/bar/'
  64
  65 TEST_PACKAGE_CP = 'a_cute/little_puppy'
  66 TEST_PACKAGE_CPV = 'a_cute/little_puppy-3.14159'
  67 TEST_PACKAGE_C = 'a_cute'
  68 TEST_PACKAGE_PV = 'little_puppy-3.14159'
  69 TEST_PORTAGE_ROOT = '/bib/bob/'
  70 TEST_PACKAGE_OLDCONTENTS = {
  71   u'/by/the/prickling/of/my/thumbs'   : (u'obj', '1234', '4321'),
  72   u'/something/wicked/this/way/comes' : (u'dir',)
  73 }
  74
  75 class ItemizeChangesFromRsyncOutput(unittest.TestCase):
  76   """Test autotest_quickmerge.ItemizeChangesFromRsyncOutput."""
  77
  78   def testItemizeChangesFromRsyncOutput(self):
  79     """Test that rsync output parser returns correct FileMutations."""
  80     expected_new = set(
  81         [('>f+++++++++', '/foo/bar/new_file'),
  82          ('>f+++++++++', '/foo/bar/directory_a/new_file_in_directory'),
  83          ('cL+++++++++', '/foo/bar/new_symlink')])
  84
  85     expected_mod = set(
  86         [('>f..t......', '/foo/bar/touched file with spaces'),
  87          ('>f..t......', '/foo/bar/touched_file'),
  88          ('>f.st......', '/foo/bar/modified_contents_file'),
  89          ('.f...p.....', '/foo/bar/modified_permissions_file'),
  90          ('.f....o....', '/foo/bar/modified_owner_file'),
  91          ('>f..t......', '/foo/bar/directory_a/touched_file_in_directory')])
  92
  93     expected_dir = set([('cd+++++++++', '/foo/bar/new_empty_directory/')])
  94
  95     report = autotest_quickmerge.ItemizeChangesFromRsyncOutput(
  96         RSYNC_TEST_OUTPUT, RSYNC_TEST_DESTINATION_PATH)
  97
  98     self.assertEqual(expected_new, set(report.new_files))
  99     self.assertEqual(expected_mod, set(report.modified_files))
 100     self.assertEqual(expected_dir, set(report.new_directories))
 101
 102
 103 class PackageNameParsingTest(unittest.TestCase):
 104   """Test autotest_quickmerge.GetStalePackageNames."""
 105
 106   def testGetStalePackageNames(self):
 107     autotest_sysroot = '/an/arbitrary/path/'
 108     change_report = autotest_quickmerge.ItemizeChangesFromRsyncOutput(
 109         RSYNC_TEST_OUTPUT_FOR_PACKAGE_UPDATE, autotest_sysroot)
 110     package_matches = autotest_quickmerge.GetStalePackageNames(
 111         change_report.modified_files + change_report.new_files,
 112         autotest_sysroot)
 113     expected_set = set(['factory_Leds', 'login_UserPolicyKeys',
 114                         'platform_Cryptohome', 'power_Thermal'])
 115     self.assertEqual(set(package_matches), expected_set)
 116
 117
 118 class RsyncCommandTest(cros_build_lib_unittest.RunCommandTestCase):
 119   """Test autotest_quickmerge.RsyncQuickmerge."""
 120
 121   def testRsyncQuickmergeCommand(self):
 122     """Test that RsyncQuickMerge makes correct call to SudoRunCommand"""
 123     include_file_name = 'an_include_file_name'
 124     source_path = 'a_source_path'
 125     sysroot_path = 'a_sysroot_path'
 126
 127     expected_command = ['rsync', '-a', '-n', '-u', '-i',
 128                         '--exclude=**.pyc', '--exclude=**.pyo',
 129                         '--exclude=** -> *',
 130                         '--include-from=%s' % include_file_name,
 131                         '--exclude=*',
 132                         source_path,
 133                         sysroot_path]
 134
 135     autotest_quickmerge.RsyncQuickmerge(source_path, sysroot_path,
 136                                         include_file_name,
 137                                         pretend=True,
 138                                         overwrite=False)
 139
 140     self.assertCommandContains(expected_command)
 141
 142
 143 class PortageManipulationsTest(mox.MoxTestBase):
 144   """Test usage of autotest_quickmerge.portage."""
 145
 146   def testUpdatePackageContents(self):
 147     """Test that UpdatePackageContents makes the correct calls to portage."""
 148     autotest_quickmerge.portage = self.mox.CreateMockAnything('portage')
 149     portage = autotest_quickmerge.portage
 150
 151     portage.root = TEST_PORTAGE_ROOT
 152
 153     mock_vartree = self.mox.CreateMockAnything('vartree')
 154     mock_vartree.settings = {'an arbitrary' : 'dictionary'}
 155     mock_tree = {TEST_PORTAGE_ROOT : {'vartree' : mock_vartree}}
 156     portage.create_trees(TEST_PORTAGE_ROOT,
 157         TEST_PORTAGE_ROOT).AndReturn(mock_tree)
 158
 159     mock_vartree.dbapi = self.mox.CreateMockAnything('dbapi')
 160     mock_vartree.dbapi.cp_list(TEST_PACKAGE_CP).AndReturn([TEST_PACKAGE_CPV])
 161
 162     mock_package = self.mox.CreateMockAnything('dblink')
 163     portage.dblink(TEST_PACKAGE_C, TEST_PACKAGE_PV, #pylint: disable-msg=E1101
 164         settings=mock_vartree.settings,
 165         vartree=mock_vartree).AndReturn(mock_package)
 166     mock_package.getcontents().AndReturn(TEST_PACKAGE_OLDCONTENTS)
 167
 168     EXPECTED_NEW_ENTRIES = {
 169         '/foo/bar/new_empty_directory': (u'dir',),
 170         '/foo/bar/directory_a/new_file_in_directory': (u'obj', '0', '0'),
 171         '/foo/bar/new_file': (u'obj', '0', '0'),
 172         '/foo/bar/new_symlink': (u'obj', '0', '0')
 173     }
 174     RESULT_DICIONARY = TEST_PACKAGE_OLDCONTENTS.copy()
 175     RESULT_DICIONARY.update(EXPECTED_NEW_ENTRIES)
 176
 177     mock_vartree.dbapi.writeContentsToContentsFile(mock_package,
 178       RESULT_DICIONARY)
 179
 180     self.mox.ReplayAll()
 181
 182     change_report = autotest_quickmerge.ItemizeChangesFromRsyncOutput(
 183       RSYNC_TEST_OUTPUT, RSYNC_TEST_DESTINATION_PATH)
 184     autotest_quickmerge.UpdatePackageContents(change_report, TEST_PACKAGE_CP,
 185         TEST_PORTAGE_ROOT)
 186
 187     self.mox.VerifyAll()
 188
 189 class PortageAPITest(unittest.TestCase):
 190   """Ensures that required portage API exists."""
 191   def runTest(self):
 192     try:
 193       # pylint: disable=F0401
 194       import portage
 195     except ImportError:
 196       self.skipTest('Portage not available in test environment. Re-run test '
 197                     'in chroot.')
 198     try:
 199       # pylint: disable-msg=E1101
 200       f = portage.vardbapi.writeContentsToContentsFile
 201     except AttributeError:
 202       self.fail('Required writeContentsToContentsFile function does '
 203                 'not exist.')
 204
 205     self.assertIsInstance(f, types.UnboundMethodType,
 206                           'Required writeContentsToContentsFile is not '
 207                           'a function.')
 208
 209 if __name__ == '__main__':
 210   cros_test_lib.main()