| # Lint as: python2, python3 |
| # Copyright (c) 2012 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Storage device utilities to be used in storage device based tests |
| """ |
| |
| import logging, re, os, time, hashlib |
| |
| from autotest_lib.client.bin import test, utils |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.cros import liststorage |
| |
| |
| class StorageException(error.TestError): |
| """Indicates that a storage/volume operation failed. |
| It is fatal to the test unless caught. |
| """ |
| pass |
| |
| |
| class StorageScanner(object): |
| """Scan device for storage points. |
| |
| It also performs basic operations on found storage devices as mount/umount, |
| creating file with randomized content or checksum file content. |
| |
| Each storage device is defined by a dictionary containing the following |
| keys: |
| |
| device: the device path (e.g. /dev/sdb1) |
| bus: the bus name (e.g. usb, ata, etc) |
| model: the kind of device (e.g. Multi-Card, USB_DISK_2.0, SanDisk) |
| size: the size of the volume/partition ib bytes (int) |
| fs_uuid: the UUID for the filesystem (str) |
| fstype: filesystem type |
| is_mounted: wether the FS is mounted (0=False,1=True) |
| mountpoint: where the FS is mounted (if mounted=1) or a suggestion where to |
| mount it (if mounted=0) |
| |
| Also |filter()| and |scan()| will use the same dictionary keys associated |
| with regular expression in order to filter a result set. |
| Multiple keys act in an AND-fashion way. The absence of a key in the filter |
| make the filter matching all the values for said key in the storage |
| dictionary. |
| |
| Example: {'device':'/dev/sd[ab]1', 'is_mounted':'0'} will match all the |
| found devices which block device file is either /dev/sda1 or /dev/sdb1, AND |
| are not mounted, excluding all other devices from the matched result. |
| """ |
| storages = None |
| |
| |
| def __init__(self): |
| self.__mounted = {} |
| |
| |
| def filter(self, storage_filter={}): |
| """Filters a stored result returning a list of matching devices. |
| |
| The passed dictionary represent the filter and its values are regular |
| expressions (str). If an element of self.storage matches the regex |
| defined in all the keys for a filter, the item will be part of the |
| returning value. |
| |
| Calling this method does not change self.storages, thus can be called |
| several times against the same result set. |
| |
| @param storage_filter: a dictionary representing the filter. |
| |
| @return a list of dictionaries representing the found devices after the |
| application of the filter. The list can be empty if no device |
| has been found. |
| """ |
| ret = [] |
| |
| for storage in self.storages: |
| matches = True |
| for key in storage_filter: |
| if not re.match(storage_filter[key], storage[key]): |
| matches = False |
| break |
| if matches: |
| ret.append(storage.copy()) |
| |
| return ret |
| |
| |
| def scan(self, storage_filter={}): |
| """Scan the current storage devices. |
| |
| If no parameter is given, it will return all the storage devices found. |
| Otherwise it will internally call self.filter() with the passed |
| filter. |
| The result (being it filtered or not) will be saved in self.storages. |
| |
| Such list can be (re)-filtered using self.filter(). |
| |
| @param storage_filter: a dict representing the filter, default is |
| matching anything. |
| |
| @return a list of found dictionaries representing the found devices. |
| The list can be empty if no device has been found. |
| """ |
| self.storages = liststorage.get_all() |
| |
| if storage_filter: |
| self.storages = self.filter(storage_filter) |
| |
| return self.storages |
| |
| |
| def mount_volume(self, index=None, storage_dict=None, args=''): |
| """Mount the passed volume. |
| |
| Either index or storage_dict can be set, but not both at the same time. |
| If neither is passed, it will mount the first volume found in |
| self.storage. |
| |
| @param index: (int) the index in self.storages for the storage |
| device/volume to be mounted. |
| @param storage_dict: (dict) the storage dictionary representing the |
| storage device, the dictionary should be obtained from |
| self.storage or using self.scan() or self.filter(). |
| @param args: (str) args to be passed to the mount command, if needed. |
| e.g., "-o foo,bar -t ext3". |
| """ |
| if index is None and storage_dict is None: |
| storage_dict = self.storages[0] |
| elif isinstance(index, int): |
| storage_dict = self.storages[index] |
| elif not isinstance(storage_dict, dict): |
| raise TypeError('Either index or storage_dict passed ' |
| 'with the wrong type') |
| |
| if storage_dict['is_mounted']: |
| logging.debug('Volume "%s" is already mounted, skipping ' |
| 'mount_volume().') |
| return |
| |
| logging.info('Mounting %(device)s in %(mountpoint)s.', storage_dict) |
| |
| try: |
| # Create the dir in case it does not exist. |
| os.mkdir(storage_dict['mountpoint']) |
| except OSError as e: |
| # If it's not "file exists", report the exception. |
| if e.errno != 17: |
| raise e |
| cmd = 'mount %s' % args |
| cmd += ' %(device)s %(mountpoint)s' % storage_dict |
| utils.system(cmd) |
| storage_dict['is_mounted'] = True |
| self.__mounted[storage_dict['mountpoint']] = storage_dict |
| |
| |
| def umount_volume(self, index=None, storage_dict=None, args=''): |
| """Un-mount the passed volume, by index or storage dictionary. |
| |
| Either index or storage_dict can be set, but not both at the same time. |
| If neither is passed, it will mount the first volume found in |
| self.storage. |
| |
| @param index: (int) the index in self.storages for the storage |
| device/volume to be mounted. |
| @param storage_dict: (dict) the storage dictionary representing the |
| storage device, the dictionary should be obtained from |
| self.storage or using self.scan() or self.filter(). |
| @param args: (str) args to be passed to the umount command, if needed. |
| e.g., '-f -t' for force+lazy umount. |
| """ |
| if index is None and storage_dict is None: |
| storage_dict = self.storages[0] |
| elif isinstance(index, int): |
| storage_dict = self.storages[index] |
| elif not isinstance(storage_dict, dict): |
| raise TypeError('Either index or storage_dict passed ' |
| 'with the wrong type') |
| |
| |
| if not storage_dict['is_mounted']: |
| logging.debug('Volume "%s" is already unmounted: skipping ' |
| 'umount_volume().') |
| return |
| |
| logging.info('Unmounting %(device)s from %(mountpoint)s.', |
| storage_dict) |
| cmd = 'umount %s' % args |
| cmd += ' %(device)s' % storage_dict |
| utils.system(cmd) |
| # We don't care if it fails, it might be busy for a /proc/mounts issue. |
| # See BUG=chromium-os:32105 |
| try: |
| os.rmdir(storage_dict['mountpoint']) |
| except OSError as e: |
| logging.debug('Removing %s failed: %s: ignoring.', |
| storage_dict['mountpoint'], e) |
| storage_dict['is_mounted'] = False |
| # If we previously mounted it, remove it from our internal list. |
| if storage_dict['mountpoint'] in self.__mounted: |
| del self.__mounted[storage_dict['mountpoint']] |
| |
| |
| def unmount_all(self): |
| """Unmount all volumes mounted by self.mount_volume(). |
| """ |
| # We need to copy it since we are iterating over a dict which will |
| # change size. |
| for volume in self.__mounted.copy(): |
| self.umount_volume(storage_dict=self.__mounted[volume]) |
| |
| |
| class StorageTester(test.test): |
| """This is a class all tests about Storage can use. |
| |
| It has methods to |
| - create random files |
| - compute a file's md5 checksum |
| - look/wait for a specific device (specified using StorageScanner |
| dictionary format) |
| |
| Subclasses can override the _prepare_volume() method in order to disable |
| them or change their behaviours. |
| |
| Subclasses should take care of unmount all the mounted filesystems when |
| needed (e.g. on cleanup phase), calling self.umount_volume() or |
| self.unmount_all(). |
| """ |
| scanner = None |
| |
| |
| def initialize(self, filter_dict={'bus':'usb'}, filesystem='ext2'): |
| """Initialize the test. |
| |
| Instantiate a StorageScanner instance to be used by tests and prepare |
| any volume matched by |filter_dict|. |
| Volume preparation is done by the _prepare_volume() method, which can be |
| overriden by subclasses. |
| |
| @param filter_dict: a dictionary to filter attached USB devices to be |
| initialized. |
| @param filesystem: the filesystem name to format the attached device. |
| """ |
| super(StorageTester, self).initialize() |
| |
| self.scanner = StorageScanner() |
| |
| self._prepare_volume(filter_dict, filesystem=filesystem) |
| |
| # Be sure that if any operation above uses self.scanner related |
| # methods, its result is cleaned after use. |
| self.storages = None |
| |
| |
| def _prepare_volume(self, filter_dict, filesystem='ext2'): |
| """Prepare matching volumes for test. |
| |
| Prepare all the volumes matching |filter_dict| for test by formatting |
| the matching storages with |filesystem|. |
| |
| This method is called by StorageTester.initialize(), a subclass can |
| override this method to change its behaviour. |
| Setting it to None (or a not callable) will disable it. |
| |
| @param filter_dict: a filter for the storages to be prepared. |
| @param filesystem: filesystem with which volumes will be formatted. |
| """ |
| if not os.path.isfile('/sbin/mkfs.%s' % filesystem): |
| raise error.TestError('filesystem not supported by mkfs installed ' |
| 'on this device') |
| |
| try: |
| storages = self.wait_for_devices(filter_dict, cycles=1, |
| mount_volume=False)[0] |
| |
| for storage in storages: |
| logging.debug('Preparing volume on %s.', storage['device']) |
| cmd = 'mkfs.%s %s' % (filesystem, storage['device']) |
| utils.system(cmd) |
| except StorageException as e: |
| logging.warning("%s._prepare_volume() didn't find any device " |
| "attached: skipping volume preparation: %s", |
| self.__class__.__name__, e) |
| except error.CmdError as e: |
| logging.warning("%s._prepare_volume() couldn't format volume: %s", |
| self.__class__.__name__, e) |
| |
| logging.debug('Volume preparation finished.') |
| |
| |
| def wait_for_devices(self, storage_filter, time_to_sleep=1, cycles=10, |
| mount_volume=True): |
| """Cycles |cycles| times waiting |time_to_sleep| seconds each cycle, |
| looking for a device matching |storage_filter| |
| |
| @param storage_filter: a dictionary holding a set of storage device's |
| keys which are used as filter, to look for devices. |
| @see StorageDevice class documentation. |
| @param time_to_sleep: time (int) to wait after each |cycles|. |
| @param cycles: number of tentatives. Use -1 for infinite. |
| |
| @raises StorageException if no device can be found. |
| |
| @return (storage_dict, waited_time) tuple. storage_dict is the found |
| device list and waited_time is the time spent waiting for the |
| device to be found. |
| """ |
| msg = ('Scanning for %s for %d times, waiting each time ' |
| '%d secs' % (storage_filter, cycles, time_to_sleep)) |
| if mount_volume: |
| logging.debug('%s and mounting each matched volume.', msg) |
| else: |
| logging.debug('%s, but not mounting each matched volume.', msg) |
| |
| if cycles == -1: |
| logging.info('Waiting until device is inserted, ' |
| 'no timeout has been set.') |
| |
| cycle = 0 |
| while cycles == -1 or cycle < cycles: |
| ret = self.scanner.scan(storage_filter) |
| if ret: |
| logging.debug('Found %s (mount_volume=%d).', ret, mount_volume) |
| if mount_volume: |
| for storage in ret: |
| self.scanner.mount_volume(storage_dict=storage) |
| |
| return (ret, cycle*time_to_sleep) |
| else: |
| logging.debug('Storage %s not found, wait and rescan ' |
| '(cycle %d).', storage_filter, cycle) |
| # Wait a bit and rescan storage list. |
| time.sleep(time_to_sleep) |
| cycle += 1 |
| |
| # Device still not found. |
| msg = ('Could not find anything matching "%s" after %d seconds' % |
| (storage_filter, time_to_sleep*cycles)) |
| raise StorageException(msg) |
| |
| |
| def wait_for_device(self, storage_filter, time_to_sleep=1, cycles=10, |
| mount_volume=True): |
| """Cycles |cycles| times waiting |time_to_sleep| seconds each cycle, |
| looking for a device matching |storage_filter|. |
| |
| This method needs to match one and only one device. |
| @raises StorageException if no device can be found or more than one is |
| found. |
| |
| @param storage_filter: a dictionary holding a set of storage device's |
| keys which are used as filter, to look for devices |
| The filter has to be match a single device, a multiple matching |
| filter will lead to StorageException to e risen. Use |
| self.wait_for_devices() if more than one device is allowed to |
| be found. |
| @see StorageDevice class documentation. |
| @param time_to_sleep: time (int) to wait after each |cycles|. |
| @param cycles: number of tentatives. Use -1 for infinite. |
| |
| @return (storage_dict, waited_time) tuple. storage_dict is the found |
| device list and waited_time is the time spent waiting for the |
| device to be found. |
| """ |
| storages, waited_time = self.wait_for_devices(storage_filter, |
| time_to_sleep=time_to_sleep, |
| cycles=cycles, |
| mount_volume=mount_volume) |
| if len(storages) > 1: |
| msg = ('filter matched more than one storage volume, use ' |
| '%s.wait_for_devices() if you need more than one match' % |
| self.__class__) |
| raise StorageException(msg) |
| |
| # Return the first element if only this one has been matched. |
| return (storages[0], waited_time) |
| |
| |
| # Some helpers not present in utils.py to abstract normal file operations. |
| |
| def create_file(path, size): |
| """Create a file using /dev/urandom. |
| |
| @param path: the path of the file. |
| @param size: the file size in bytes. |
| """ |
| logging.debug('Creating %s (size %d) from /dev/urandom.', path, size) |
| with open('/dev/urandom', 'rb') as urandom: |
| utils.open_write_close(path, urandom.read(size)) |
| |
| |
| def checksum_file(path): |
| """Compute the MD5 Checksum for a file. |
| |
| @param path: the path of the file. |
| |
| @return a string with the checksum. |
| """ |
| chunk_size = 1024 |
| |
| m = hashlib.md5() |
| with open(path, 'rb') as f: |
| for chunk in f.read(chunk_size): |
| m.update(chunk) |
| |
| logging.debug("MD5 checksum for %s is %s.", path, m.hexdigest()) |
| |
| return m.hexdigest() |
| |
| |
| def args_to_storage_dict(args): |
| """Map args into storage dictionaries. |
| |
| This function is to be used (likely) in control files to obtain a storage |
| dictionary from command line arguments. |
| |
| @param args: a list of arguments as passed to control file. |
| |
| @return a tuple (storage_dict, rest_of_args) where storage_dict is a |
| dictionary for storage filtering and rest_of_args is a dictionary |
| of keys which do not match storage dict keys. |
| """ |
| args_dict = utils.args_to_dict(args) |
| storage_dict = {} |
| |
| # A list of all allowed keys and their type. |
| key_list = ('device', 'bus', 'model', 'size', 'fs_uuid', 'fstype', |
| 'is_mounted', 'mountpoint') |
| |
| def set_if_exists(src, dst, key): |
| """If |src| has |key| copies its value to |dst|. |
| |
| @return True if |key| exists in |src|, False otherwise. |
| """ |
| if key in src: |
| dst[key] = src[key] |
| return True |
| else: |
| return False |
| |
| for key in key_list: |
| if set_if_exists(args_dict, storage_dict, key): |
| del args_dict[key] |
| |
| # Return the storage dict and the leftovers of the args to be evaluated |
| # later. |
| return storage_dict, args_dict |
