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