xref: /aosp_15_r20/external/autotest/client/cros/storage.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

# 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