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 dbus 6 import glob 7 import logging 8 import json 9 import os 10 import tempfile 11 12 from autotest_lib.client.bin import test 13 from autotest_lib.client.bin import utils 14 from autotest_lib.client.common_lib import error 15 from autotest_lib.client.cros.cros_disks import CrosDisksTester 16 from autotest_lib.client.cros.cros_disks import ExceptionSuppressor 17 from autotest_lib.client.cros.cros_disks import VirtualFilesystemImage 18 from autotest_lib.client.cros.cros_disks import DefaultFilesystemTestContent 19 20 21 class 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, ['force']) 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 91 class 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
