Home | History | Annotate | Download | only in platform_CrosDisksFilesystem 
      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 contextlib
      6 import dbus
      7 import glob
      8 import logging
      9 import json
     10 import os
     11 import tempfile
     12 
     13 from autotest_lib.client.bin import test
     14 from autotest_lib.client.bin import utils
     15 from autotest_lib.client.common_lib import error
     16 from autotest_lib.client.cros.cros_disks import CrosDisksTester
     17 from autotest_lib.client.cros.cros_disks import ExceptionSuppressor
     18 from autotest_lib.client.cros.cros_disks import VirtualFilesystemImage
     19 from autotest_lib.client.cros.cros_disks import DefaultFilesystemTestContent
     20 
     21 
     22 class 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 
    110 class 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