1 # Copyright (c) 2011 The Chromium OS Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 5 import os 6 import shutil 7 8 from autotest_lib.client.bin import test 9 from autotest_lib.client.common_lib import autotemp, error 10 from autotest_lib.client.cros.cros_disks import CrosDisksTester 11 from autotest_lib.client.cros.cros_disks import VirtualFilesystemImage 12 from autotest_lib.client.cros.cros_disks import DefaultFilesystemTestContent 13 from collections import deque 14 15 16 class CrosDisksArchiveTester(CrosDisksTester): 17 """A tester to verify archive support in CrosDisks. 18 """ 19 def __init__(self, test, archive_types): 20 super(CrosDisksArchiveTester, self).__init__(test) 21 self._data_dir = os.path.join(test.bindir, 'data') 22 self._archive_types = archive_types 23 24 def _find_all_files(self, root_dir): 25 """Returns all files under a directory and its sub-directories. 26 27 This is a generator that performs a breadth-first-search of 28 all files under a specified directory and its sub-directories. 29 30 Args: 31 root_dir: The root directory where the search starts from. 32 Yields: 33 Path of any found file relative to the root directory. 34 """ 35 dirs_to_explore = deque(['']) 36 while len(dirs_to_explore) > 0: 37 current_dir = dirs_to_explore.popleft() 38 for path in os.listdir(os.path.join(root_dir, current_dir)): 39 expanded_path = os.path.join(root_dir, current_dir, path) 40 relative_path = os.path.join(current_dir, path) 41 if os.path.isdir(expanded_path): 42 dirs_to_explore.append(relative_path) 43 else: 44 yield relative_path 45 46 def _make_rar_archive(self, archive_path, root_dir): 47 """Archives a specified directory into a RAR file. 48 49 The created RAR file contains all files and sub-directories 50 under the specified root directory, but not the root directory 51 itself. 52 53 Args: 54 archive_path: Path of the output archive. 55 root_dir: The root directory to archive. 56 """ 57 # DESPICABLE HACK: app-arch/rar provides only pre-compiled rar binaries 58 # for x86/amd64. As a workaround, we pretend the RAR creation here 59 # using a precanned RAR file. 60 shutil.copyfile(os.path.join(self._data_dir, 'test.rar'), archive_path) 61 62 def _make_archive(self, archive_type, archive_path, root_dir): 63 """Archives a specified directory into an archive of specified type. 64 65 The created archive file contains all files and sub-directories 66 under the specified root directory, but not the root directory 67 itself. 68 69 Args: 70 archive_type: Type of the output archive. 71 archive_path: Path of the output archive. 72 root_dir: The root directory to archive. 73 """ 74 if archive_type in ['rar']: 75 self._make_rar_archive(archive_path, root_dir) 76 else: 77 raise error.TestFail("Unsupported archive type " + archive_type) 78 79 def _test_archive(self, archive_type): 80 # Create the archive file content in a temporary directory. 81 archive_dir = autotemp.tempdir(unique_id='CrosDisks') 82 test_content = DefaultFilesystemTestContent() 83 if not test_content.create(archive_dir.name): 84 raise error.TestFail("Failed to create archive test content") 85 86 # Create a FAT-formatted virtual filesystem image containing an 87 # archive file to help stimulate mounting an archive file on a 88 # removable drive. 89 with VirtualFilesystemImage( 90 block_size=1024, 91 block_count=65536, 92 filesystem_type='vfat', 93 mkfs_options=[ '-F', '32', '-n', 'ARCHIVE' ]) as image: 94 image.format() 95 image.mount(options=['sync']) 96 # Create the archive file on the virtual filesystem image. 97 archive_name = 'test.' + archive_type 98 archive_path = os.path.join(image.mount_dir, archive_name) 99 self._make_archive(archive_type, archive_path, archive_dir.name) 100 image.unmount() 101 102 # Mount the virtual filesystem image via CrosDisks. 103 device_file = image.loop_device 104 self.cros_disks.mount(device_file, '', 105 [ "ro", "nodev", "noexec", "nosuid" ]) 106 result = self.cros_disks.expect_mount_completion({ 107 'status': 0, 108 'source_path': device_file 109 }) 110 111 # Mount the archive file on the mounted filesystem via CrosDisks. 112 archive_path = os.path.join(result['mount_path'], archive_name) 113 expected_mount_path = os.path.join('/media/archive', archive_name) 114 self.cros_disks.mount(archive_path) 115 result = self.cros_disks.expect_mount_completion({ 116 'status': 0, 117 'source_path': archive_path, 118 'mount_path': expected_mount_path 119 }) 120 121 # Verify the content of the mounted archive file. 122 if not test_content.verify(expected_mount_path): 123 raise error.TestFail("Failed to verify filesystem test content") 124 125 self.cros_disks.unmount(expected_mount_path, ['lazy']) 126 self.cros_disks.unmount(device_file, ['lazy']) 127 128 def test_archives(self): 129 for archive_type in self._archive_types: 130 self._test_archive(archive_type) 131 132 def get_tests(self): 133 return [self.test_archives] 134 135 136 class platform_CrosDisksArchive(test.test): 137 version = 1 138 139 def run_once(self, *args, **kwargs): 140 tester = CrosDisksArchiveTester(self, kwargs['archive_types']) 141 tester.run(*args, **kwargs) 142
