• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
5import os
6import shutil
7
8from autotest_lib.client.bin import test
9from autotest_lib.client.common_lib import autotemp, error
10from autotest_lib.client.cros.cros_disks import CrosDisksTester
11from autotest_lib.client.cros.cros_disks import VirtualFilesystemImage
12from autotest_lib.client.cros.cros_disks import DefaultFilesystemTestContent
13from collections import deque
14
15
16class 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
136class 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