"""
def testListBlockDevices(self):
- """Tests that we can list al block devices correctly."""
+ """Tests that we can list all block devices correctly."""
self.rc.AddCmdResult(partial_mock.Ignore(), output=self.FULL_OUTPUT)
devices = osutils.ListBlockDevices()
self.assertEqual(devices[0].NAME, 'sda')
self.assertEqual(osutils.GetDeviceSize('/dev/sdc'), '7.4G')
+class MountImagePartitionTests(cros_test_lib.MockTestCase):
+ """Tests for MountImagePartition."""
+
+ def setUp(self):
+ self._gpt_table = {
+ 3: cros_build_lib.PartitionInfo(3, 1, 3, 2, 'fs', 'Label', 'flag')
+ }
+
+ def testWithCacheOkay(self):
+ mount_dir = self.PatchObject(osutils, 'MountDir')
+ osutils.MountImagePartition('image_file', 3, 'destination',
+ self._gpt_table)
+ opts = ['loop', 'offset=1', 'sizelimit=2', 'ro']
+ mount_dir.assert_called_with('image_file', 'destination', makedirs=True,
+ skip_mtab=False, sudo=True, mount_opts=opts)
+
+ def testWithCacheFail(self):
+ self.assertRaises(ValueError, osutils.MountImagePartition,
+ 'image_file', 404, 'destination', self._gpt_table)
+
+ def testWithoutCache(self):
+ self.PatchObject(cros_build_lib, 'GetImageDiskPartitionInfo',
+ return_value=self._gpt_table)
+ mount_dir = self.PatchObject(osutils, 'MountDir')
+ osutils.MountImagePartition('image_file', 3, 'destination')
+ opts = ['loop', 'offset=1', 'sizelimit=2', 'ro']
+ mount_dir.assert_called_with(
+ 'image_file', 'destination', makedirs=True, skip_mtab=False,
+ sudo=True, mount_opts=opts
+ )
+
+
+class ChdirTests(cros_test_lib.MockTempDirTestCase):
+ """Tests for ChdirContext."""
+
+ def testChdir(self):
+ current_dir = os.getcwd()
+ self.assertNotEqual(self.tempdir, os.getcwd())
+ with osutils.ChdirContext(self.tempdir):
+ self.assertEqual(self.tempdir, os.getcwd())
+ self.assertEqual(current_dir, os.getcwd())
+
+
+class MountImageTests(cros_test_lib.MockTempDirTestCase):
+ """Tests for MountImageContext."""
+
+ def _testWithParts(self, parts, selectors, check_links=True):
+ self.PatchObject(cros_build_lib, 'GetImageDiskPartitionInfo',
+ return_value=parts)
+ mount_dir = self.PatchObject(osutils, 'MountDir')
+ unmount_dir = self.PatchObject(osutils, 'UmountDir')
+ rmdir = self.PatchObject(osutils, 'RmDir')
+ with osutils.MountImageContext('_ignored', self.tempdir, selectors):
+ for _, part in parts.items():
+ mount_point = os.path.join(self.tempdir, 'dir-%d' % part.number)
+ mount_dir.assert_any_call(
+ '_ignored', mount_point, makedirs=True, skip_mtab=False,
+ sudo=True,
+ mount_opts=['loop', 'offset=0', 'sizelimit=0', 'ro']
+ )
+ if check_links:
+ link = os.path.join(self.tempdir, 'dir-%s' % part.name)
+ self.assertTrue(os.path.islink(link))
+ self.assertEqual(os.path.basename(mount_point),
+ os.readlink(link))
+ for _, part in parts.items():
+ mount_point = os.path.join(self.tempdir, 'dir-%d' % part.number)
+ unmount_dir.assert_any_call(mount_point, cleanup=False)
+ rmdir.assert_any_call(mount_point, sudo=True)
+ if check_links:
+ link = os.path.join(self.tempdir, 'dir-%s' % part.name)
+ self.assertFalse(os.path.lexists(link))
+
+ def testWithPartitionNumber(self):
+ parts = {
+ 1: cros_build_lib.PartitionInfo(1, 0, 0, 0, '', 'my-stateful', ''),
+ 3: cros_build_lib.PartitionInfo(3, 0, 0, 0, '', 'my-root-a', ''),
+ }
+ self._testWithParts(parts, [1, 3])
+
+ def testWithPartitionLabel(self):
+ parts = {
+ 42: cros_build_lib.PartitionInfo(42, 0, 0, 0, '', 'label', ''),
+ }
+ self._testWithParts(parts, ['label'])
+
+ def testInvalidPartSelector(self):
+ parts = {
+ 42: cros_build_lib.PartitionInfo(42, 0, 0, 0, '', 'label', ''),
+ }
+ self.assertRaises(ValueError, self._testWithParts, parts, ['label404'])
+ self.assertRaises(ValueError, self._testWithParts, parts, [404])
+
+ def testFailOnExistingMount(self):
+ parts = {
+ 42: cros_build_lib.PartitionInfo(42, 0, 0, 0, '', 'label', ''),
+ }
+ os.makedirs(os.path.join(self.tempdir, 'dir-42'))
+ self.assertRaises(ValueError, self._testWithParts, parts, [42])
+
+ def testExistingLinkNotCleanedUp(self):
+ parts = {
+ 42: cros_build_lib.PartitionInfo(42, 0, 0, 0, '', 'label', ''),
+ }
+ symlink = os.path.join(self.tempdir, 'dir-label')
+ os.symlink('/tmp', symlink)
+ self.assertEqual('/tmp', os.readlink(symlink))
+ self._testWithParts(parts, [42], check_links=False)
+ self.assertEqual('/tmp', os.readlink(symlink))
+
+
+class IterateMountPointsTests(cros_test_lib.TempDirTestCase):
+ """Test for IterateMountPoints function."""
+
+ def setUp(self):
+ self.proc_mount = os.path.join(self.tempdir, 'mounts')
+ osutils.WriteFile(
+ self.proc_mount,
+ r'''/dev/loop0 /mnt/dir_8 ext4 rw,relatime,data=ordered 0 0
+/dev/loop2 /mnt/dir_1 ext4 rw,relatime,data=ordered 0 0
+/dev/loop1 /mnt/dir_12 vfat rw 0 0
+/dev/loop4 /mnt/dir_3 ext4 ro,relatime 0 0
+weird\040system /mnt/weirdo unknown ro 0 0
+tmpfs /mnt/spaced\040dir tmpfs ro 0 0
+tmpfs /mnt/\134 tmpfs ro 0 0
+'''
+ )
+
+ def testOkay(self):
+ r = list(osutils.IterateMountPoints(self.proc_mount))
+ self.assertEqual(len(r), 7)
+ self.assertEqual(r[0].source, '/dev/loop0')
+ self.assertEqual(r[1].destination, '/mnt/dir_1')
+ self.assertEqual(r[2].filesystem, 'vfat')
+ self.assertEqual(r[3].options, 'ro,relatime')
+
+ def testEscape(self):
+ r = list(osutils.IterateMountPoints(self.proc_mount))
+ self.assertEqual(r[4].source, 'weird system')
+ self.assertEqual(r[5].destination, '/mnt/spaced dir')
+ self.assertEqual(r[6].destination, '/mnt/\\')
+
+
if __name__ == '__main__':
cros_test_lib.main()