• 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 dbus
6import glob
7import logging
8import json
9import os
10import tempfile
11
12from autotest_lib.client.bin import test
13from autotest_lib.client.bin import utils
14from autotest_lib.client.common_lib import error
15from autotest_lib.client.cros.cros_disks import CrosDisksTester
16from autotest_lib.client.cros.cros_disks import ExceptionSuppressor
17from autotest_lib.client.cros.cros_disks import VirtualFilesystemImage
18from autotest_lib.client.cros.cros_disks import DefaultFilesystemTestContent
19
20
21class CrosDisksFilesystemTester(CrosDisksTester):
22    """A tester to verify filesystem support in CrosDisks.
23    """
24    def __init__(self, test, test_configs):
25        super(CrosDisksFilesystemTester, self).__init__(test)
26        self._test_configs = test_configs
27
28    def _run_test_config(self, config):
29        logging.info('Testing "%s"', config['description'])
30        test_mount_filesystem_type = config.get('test_mount_filesystem_type')
31        test_mount_options = config.get('test_mount_options')
32        is_write_test = config.get('is_write_test', False)
33
34        # Create a virtual filesystem image based on the specified parameters in
35        # the test configuration and use it to verify that CrosDisks can
36        # recognize and mount the filesystem properly.
37        with VirtualFilesystemImage(
38                block_size=config['block_size'],
39                block_count=config['block_count'],
40                filesystem_type=config['filesystem_type'],
41                mount_filesystem_type=config.get('mount_filesystem_type'),
42                mkfs_options=config.get('mkfs_options')) as image:
43            image.format()
44            image.mount(options=['sync'])
45            test_content = DefaultFilesystemTestContent()
46
47            # If it is not a write test, create the test content on the virtual
48            # filesystem so that they can be read later after CrosDisks mounts
49            # the filesystem.
50            if not is_write_test and not test_content.create(image.mount_dir):
51                raise error.TestFail("Failed to create filesystem test content")
52            image.unmount()
53
54            device_file = image.loop_device
55            self.cros_disks.mount(device_file, test_mount_filesystem_type,
56                                  test_mount_options)
57            expected_mount_completion = {
58                'status': config['expected_mount_status'],
59                'source_path': device_file,
60            }
61            if 'expected_mount_path' in config:
62                expected_mount_completion['mount_path'] = \
63                    config['expected_mount_path']
64            result = self.cros_disks.expect_mount_completion(
65                    expected_mount_completion)
66
67            actual_mount_path = result['mount_path']
68
69            # If it is a write test, create the test content on the filesystem
70            # after it is mounted by CrosDisks.
71            if is_write_test and not test_content.create(actual_mount_path):
72                raise error.TestFail("Failed to create filesystem test content")
73
74            if not test_content.verify(actual_mount_path):
75                raise error.TestFail("Failed to verify filesystem test content")
76            self.cros_disks.unmount(device_file, ['lazy'])
77
78    def test_using_virtual_filesystem_image(self):
79        try:
80            for config in self._test_configs:
81                self._run_test_config(config)
82        except RuntimeError:
83            cmd = 'ls -la %s' % tempfile.gettempdir()
84            logging.debug(utils.run(cmd))
85            raise
86
87    def get_tests(self):
88        return [self.test_using_virtual_filesystem_image]
89
90
91class platform_CrosDisksFilesystem(test.test):
92    version = 1
93
94    def run_once(self, *args, **kwargs):
95        test_configs = []
96        config_file = '%s/%s' % (self.bindir, kwargs['config_file'])
97        with open(config_file, 'rb') as f:
98            test_configs.extend(json.load(f))
99
100        tester = CrosDisksFilesystemTester(self, test_configs)
101        tester.run(*args, **kwargs)
102