blob: fb6fe73ba2953836d8b0b25c17998cab5f07e1fe [file] [log] [blame]
# 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.
import dbus, gobject, logging, os, stat
from dbus.mainloop.glib import DBusGMainLoop
import common
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib import autotemp, error
from autotest_lib.client.cros import dbus_util
from mainloop import ExceptionForward
from mainloop import GenericTesterMainLoop
"""This module contains several helper classes for writing tests to verify the
CrosDisks DBus interface. In particular, the CrosDisksTester class can be used
to derive functional tests that interact with the CrosDisks server over DBus.
"""
class ExceptionSuppressor(object):
"""A context manager class for suppressing certain types of exception.
An instance of this class is expected to be used with the with statement
and takes a set of exception classes at instantiation, which are types of
exception to be suppressed (and logged) in the code block under the with
statement.
Example:
with ExceptionSuppressor(OSError, IOError):
# An exception, which is a sub-class of OSError or IOError, is
# suppressed in the block code under the with statement.
"""
def __init__(self, *args):
self.__suppressed_exc_types = (args)
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
if exc_type and issubclass(exc_type, self.__suppressed_exc_types):
try:
logging.exception('Suppressed exception: %s(%s)',
exc_type, exc_value)
except Exception:
pass
return True
return False
class DBusClient(object):
""" A base class of a DBus proxy client to test a DBus server.
This class is expected to be used along with a GLib main loop and provides
some convenient functions for testing the DBus API exposed by a DBus server.
"""
def __init__(self, main_loop, bus, bus_name, object_path, timeout=None):
"""Initializes the instance.
Args:
main_loop: The GLib main loop.
bus: The bus where the DBus server is connected to.
bus_name: The bus name owned by the DBus server.
object_path: The object path of the DBus server.
timeout: Maximum time in seconds to wait for the DBus connection.
"""
self.__signal_content = {}
self.main_loop = main_loop
self.signal_timeout_in_seconds = 10
logging.debug('Getting D-Bus proxy object on bus "%s" and path "%s"',
bus_name, object_path)
self.proxy_object = dbus_util.get_dbus_object(bus, bus_name,
object_path, timeout)
def clear_signal_content(self, signal_name):
"""Clears the content of the signal.
Args:
signal_name: The name of the signal.
"""
if signal_name in self.__signal_content:
self.__signal_content[signal_name] = None
def get_signal_content(self, signal_name):
"""Gets the content of a signal.
Args:
signal_name: The name of the signal.
Returns:
The content of a signal or None if the signal is not being handled.
"""
return self.__signal_content.get(signal_name)
def handle_signal(self, interface, signal_name, argument_names=()):
"""Registers a signal handler to handle a given signal.
Args:
interface: The DBus interface of the signal.
signal_name: The name of the signal.
argument_names: A list of argument names that the signal contains.
"""
if signal_name in self.__signal_content:
return
self.__signal_content[signal_name] = None
def signal_handler(*args):
self.__signal_content[signal_name] = dict(zip(argument_names, args))
logging.debug('Handling D-Bus signal "%s(%s)" on interface "%s"',
signal_name, ', '.join(argument_names), interface)
self.proxy_object.connect_to_signal(signal_name, signal_handler,
interface)
def wait_for_signal(self, signal_name):
"""Waits for the reception of a signal.
Args:
signal_name: The name of the signal to wait for.
Returns:
The content of the signal.
"""
if signal_name not in self.__signal_content:
return None
def check_signal_content():
context = self.main_loop.get_context()
while context.iteration(False):
pass
return self.__signal_content[signal_name] is not None
logging.debug('Waiting for D-Bus signal "%s"', signal_name)
utils.poll_for_condition(condition=check_signal_content,
desc='%s signal' % signal_name,
timeout=self.signal_timeout_in_seconds)
content = self.__signal_content[signal_name]
logging.debug('Received D-Bus signal "%s(%s)"', signal_name, content)
self.__signal_content[signal_name] = None
return content
def expect_signal(self, signal_name, expected_content):
"""Waits the the reception of a signal and verifies its content.
Args:
signal_name: The name of the signal to wait for.
expected_content: The expected content of the signal, which can be
partially specified. Only specified fields are
compared between the actual and expected content.
Returns:
The actual content of the signal.
Raises:
error.TestFail: A test failure when there is a mismatch between the
actual and expected content of the signal.
"""
actual_content = self.wait_for_signal(signal_name)
logging.debug("%s signal: expected=%s actual=%s",
signal_name, expected_content, actual_content)
for argument, expected_value in expected_content.iteritems():
if argument not in actual_content:
raise error.TestFail(
('%s signal missing "%s": expected=%s, actual=%s') %
(signal_name, argument, expected_content, actual_content))
if actual_content[argument] != expected_value:
raise error.TestFail(
('%s signal not matched on "%s": expected=%s, actual=%s') %
(signal_name, argument, expected_content, actual_content))
return actual_content
class CrosDisksClient(DBusClient):
"""A DBus proxy client for testing the CrosDisks DBus server.
"""
CROS_DISKS_BUS_NAME = 'org.chromium.CrosDisks'
CROS_DISKS_INTERFACE = 'org.chromium.CrosDisks'
CROS_DISKS_OBJECT_PATH = '/org/chromium/CrosDisks'
DBUS_PROPERTIES_INTERFACE = 'org.freedesktop.DBus.Properties'
FORMAT_COMPLETED_SIGNAL = 'FormatCompleted'
FORMAT_COMPLETED_SIGNAL_ARGUMENTS = (
'status', 'path'
)
MOUNT_COMPLETED_SIGNAL = 'MountCompleted'
MOUNT_COMPLETED_SIGNAL_ARGUMENTS = (
'status', 'source_path', 'source_type', 'mount_path'
)
RENAME_COMPLETED_SIGNAL = 'RenameCompleted'
RENAME_COMPLETED_SIGNAL_ARGUMENTS = (
'status', 'path'
)
def __init__(self, main_loop, bus):
"""Initializes the instance.
Args:
main_loop: The GLib main loop.
bus: The bus where the DBus server is connected to.
"""
super(CrosDisksClient, self).__init__(main_loop, bus,
self.CROS_DISKS_BUS_NAME,
self.CROS_DISKS_OBJECT_PATH)
self.interface = dbus.Interface(self.proxy_object,
self.CROS_DISKS_INTERFACE)
self.properties = dbus.Interface(self.proxy_object,
self.DBUS_PROPERTIES_INTERFACE)
self.handle_signal(self.CROS_DISKS_INTERFACE,
self.FORMAT_COMPLETED_SIGNAL,
self.FORMAT_COMPLETED_SIGNAL_ARGUMENTS)
self.handle_signal(self.CROS_DISKS_INTERFACE,
self.MOUNT_COMPLETED_SIGNAL,
self.MOUNT_COMPLETED_SIGNAL_ARGUMENTS)
self.handle_signal(self.CROS_DISKS_INTERFACE,
self.RENAME_COMPLETED_SIGNAL,
self.RENAME_COMPLETED_SIGNAL_ARGUMENTS)
def enumerate_auto_mountable_devices(self):
"""Invokes the CrosDisks EnumerateAutoMountableDevices method.
Returns:
A list of sysfs paths of devices that are auto-mountable by
CrosDisks.
"""
return self.interface.EnumerateAutoMountableDevices()
def enumerate_devices(self):
"""Invokes the CrosDisks EnumerateMountableDevices method.
Returns:
A list of sysfs paths of devices that are recognized by
CrosDisks.
"""
return self.interface.EnumerateDevices()
def get_device_properties(self, path):
"""Invokes the CrosDisks GetDeviceProperties method.
Args:
path: The device path.
Returns:
The properties of the device in a dictionary.
"""
return self.interface.GetDeviceProperties(path)
def format(self, path, filesystem_type=None, options=None):
"""Invokes the CrosDisks Format method.
Args:
path: The device path to format.
filesystem_type: The filesystem type used for formatting the device.
options: A list of options used for formatting the device.
"""
if filesystem_type is None:
filesystem_type = ''
if options is None:
options = []
self.clear_signal_content(self.FORMAT_COMPLETED_SIGNAL)
self.interface.Format(path, filesystem_type,
dbus.Array(options, signature='s'))
def wait_for_format_completion(self):
"""Waits for the CrosDisks FormatCompleted signal.
Returns:
The content of the FormatCompleted signal.
"""
return self.wait_for_signal(self.FORMAT_COMPLETED_SIGNAL)
def expect_format_completion(self, expected_content):
"""Waits and verifies for the CrosDisks FormatCompleted signal.
Args:
expected_content: The expected content of the FormatCompleted
signal, which can be partially specified.
Only specified fields are compared between the
actual and expected content.
Returns:
The actual content of the FormatCompleted signal.
Raises:
error.TestFail: A test failure when there is a mismatch between the
actual and expected content of the FormatCompleted
signal.
"""
return self.expect_signal(self.FORMAT_COMPLETED_SIGNAL,
expected_content)
def rename(self, path, volume_name=None):
"""Invokes the CrosDisks Rename method.
Args:
path: The device path to rename.
volume_name: The new name used for renaming.
"""
if volume_name is None:
volume_name = ''
self.clear_signal_content(self.RENAME_COMPLETED_SIGNAL)
self.interface.Rename(path, volume_name)
def wait_for_rename_completion(self):
"""Waits for the CrosDisks RenameCompleted signal.
Returns:
The content of the RenameCompleted signal.
"""
return self.wait_for_signal(self.RENAME_COMPLETED_SIGNAL)
def expect_rename_completion(self, expected_content):
"""Waits and verifies for the CrosDisks RenameCompleted signal.
Args:
expected_content: The expected content of the RenameCompleted
signal, which can be partially specified.
Only specified fields are compared between the
actual and expected content.
Returns:
The actual content of the RenameCompleted signal.
Raises:
error.TestFail: A test failure when there is a mismatch between the
actual and expected content of the RenameCompleted
signal.
"""
return self.expect_signal(self.RENAME_COMPLETED_SIGNAL,
expected_content)
def mount(self, path, filesystem_type=None, options=None):
"""Invokes the CrosDisks Mount method.
Args:
path: The device path to mount.
filesystem_type: The filesystem type used for mounting the device.
options: A list of options used for mounting the device.
"""
if filesystem_type is None:
filesystem_type = ''
if options is None:
options = []
self.clear_signal_content(self.MOUNT_COMPLETED_SIGNAL)
self.interface.Mount(path, filesystem_type,
dbus.Array(options, signature='s'))
def unmount(self, path, options=None):
"""Invokes the CrosDisks Unmount method.
Args:
path: The device or mount path to unmount.
options: A list of options used for unmounting the path.
"""
if options is None:
options = []
self.interface.Unmount(path, dbus.Array(options, signature='s'))
def wait_for_mount_completion(self):
"""Waits for the CrosDisks MountCompleted signal.
Returns:
The content of the MountCompleted signal.
"""
return self.wait_for_signal(self.MOUNT_COMPLETED_SIGNAL)
def expect_mount_completion(self, expected_content):
"""Waits and verifies for the CrosDisks MountCompleted signal.
Args:
expected_content: The expected content of the MountCompleted
signal, which can be partially specified.
Only specified fields are compared between the
actual and expected content.
Returns:
The actual content of the MountCompleted signal.
Raises:
error.TestFail: A test failure when there is a mismatch between the
actual and expected content of the MountCompleted
signal.
"""
return self.expect_signal(self.MOUNT_COMPLETED_SIGNAL,
expected_content)
class CrosDisksTester(GenericTesterMainLoop):
"""A base tester class for testing the CrosDisks server.
A derived class should override the get_tests method to return a list of
test methods. The perform_one_test method invokes each test method in the
list to verify some functionalities of CrosDisks server.
"""
def __init__(self, test):
bus_loop = DBusGMainLoop(set_as_default=True)
bus = dbus.SystemBus(mainloop=bus_loop)
self.main_loop = gobject.MainLoop()
super(CrosDisksTester, self).__init__(test, self.main_loop)
self.cros_disks = CrosDisksClient(self.main_loop, bus)
def get_tests(self):
"""Returns a list of test methods to be invoked by perform_one_test.
A derived class should override this method.
Returns:
A list of test methods.
"""
return []
@ExceptionForward
def perform_one_test(self):
"""Exercises each test method in the list returned by get_tests.
"""
tests = self.get_tests()
self.remaining_requirements = set([test.func_name for test in tests])
for test in tests:
test()
self.requirement_completed(test.func_name)
class FilesystemTestObject(object):
"""A base class to represent a filesystem test object.
A filesystem test object can be a file, directory or symbolic link.
A derived class should override the _create and _verify method to implement
how the test object should be created and verified, respectively, on a
filesystem.
"""
def __init__(self, path, content, mode):
"""Initializes the instance.
Args:
path: The relative path of the test object.
content: The content of the test object.
mode: The file permissions given to the test object.
"""
self._path = path
self._content = content
self._mode = mode
def create(self, base_dir):
"""Creates the test object in a base directory.
Args:
base_dir: The base directory where the test object is created.
Returns:
True if the test object is created successfully or False otherwise.
"""
if not self._create(base_dir):
logging.debug('Failed to create filesystem test object at "%s"',
os.path.join(base_dir, self._path))
return False
return True
def verify(self, base_dir):
"""Verifies the test object in a base directory.
Args:
base_dir: The base directory where the test object is expected to be
found.
Returns:
True if the test object is found in the base directory and matches
the expected content, or False otherwise.
"""
if not self._verify(base_dir):
logging.debug('Failed to verify filesystem test object at "%s"',
os.path.join(base_dir, self._path))
return False
return True
def _create(self, base_dir):
return False
def _verify(self, base_dir):
return False
class FilesystemTestDirectory(FilesystemTestObject):
"""A filesystem test object that represents a directory."""
def __init__(self, path, content, mode=stat.S_IRWXU|stat.S_IRGRP| \
stat.S_IXGRP|stat.S_IROTH|stat.S_IXOTH):
super(FilesystemTestDirectory, self).__init__(path, content, mode)
def _create(self, base_dir):
path = os.path.join(base_dir, self._path) if self._path else base_dir
if self._path:
with ExceptionSuppressor(OSError):
os.makedirs(path)
os.chmod(path, self._mode)
if not os.path.isdir(path):
return False
for content in self._content:
if not content.create(path):
return False
return True
def _verify(self, base_dir):
path = os.path.join(base_dir, self._path) if self._path else base_dir
if not os.path.isdir(path):
return False
for content in self._content:
if not content.verify(path):
return False
return True
class FilesystemTestFile(FilesystemTestObject):
"""A filesystem test object that represents a file."""
def __init__(self, path, content, mode=stat.S_IRUSR|stat.S_IWUSR| \
stat.S_IRGRP|stat.S_IROTH):
super(FilesystemTestFile, self).__init__(path, content, mode)
def _create(self, base_dir):
path = os.path.join(base_dir, self._path)
with ExceptionSuppressor(IOError):
with open(path, 'wb+') as f:
f.write(self._content)
with ExceptionSuppressor(OSError):
os.chmod(path, self._mode)
return True
return False
def _verify(self, base_dir):
path = os.path.join(base_dir, self._path)
with ExceptionSuppressor(IOError):
with open(path, 'rb') as f:
return f.read() == self._content
return False
class DefaultFilesystemTestContent(FilesystemTestDirectory):
def __init__(self):
super(DefaultFilesystemTestContent, self).__init__('', [
FilesystemTestFile('file1', '0123456789'),
FilesystemTestDirectory('dir1', [
FilesystemTestFile('file1', ''),
FilesystemTestFile('file2', 'abcdefg'),
FilesystemTestDirectory('dir2', [
FilesystemTestFile('file3', 'abcdefg'),
]),
]),
], stat.S_IRWXU|stat.S_IRGRP|stat.S_IXGRP|stat.S_IROTH|stat.S_IXOTH)
class VirtualFilesystemImage(object):
def __init__(self, block_size, block_count, filesystem_type,
*args, **kwargs):
"""Initializes the instance.
Args:
block_size: The number of bytes of each block in the image.
block_count: The number of blocks in the image.
filesystem_type: The filesystem type to be given to the mkfs
program for formatting the image.
Keyword Args:
mount_filesystem_type: The filesystem type to be given to the
mount program for mounting the image.
mkfs_options: A list of options to be given to the mkfs program.
"""
self._block_size = block_size
self._block_count = block_count
self._filesystem_type = filesystem_type
self._mount_filesystem_type = kwargs.get('mount_filesystem_type')
if self._mount_filesystem_type is None:
self._mount_filesystem_type = filesystem_type
self._mkfs_options = kwargs.get('mkfs_options')
if self._mkfs_options is None:
self._mkfs_options = []
self._image_file = None
self._loop_device = None
self._loop_device_stat = None
self._mount_dir = None
def __del__(self):
with ExceptionSuppressor(Exception):
self.clean()
def __enter__(self):
self.create()
return self
def __exit__(self, exc_type, exc_value, traceback):
self.clean()
return False
def _remove_temp_path(self, temp_path):
"""Removes a temporary file or directory created using autotemp."""
if temp_path:
with ExceptionSuppressor(Exception):
path = temp_path.name
temp_path.clean()
logging.debug('Removed "%s"', path)
def _remove_image_file(self):
"""Removes the image file if one has been created."""
self._remove_temp_path(self._image_file)
self._image_file = None
def _remove_mount_dir(self):
"""Removes the mount directory if one has been created."""
self._remove_temp_path(self._mount_dir)
self._mount_dir = None
@property
def image_file(self):
"""Gets the path of the image file.
Returns:
The path of the image file or None if no image file has been
created.
"""
return self._image_file.name if self._image_file else None
@property
def loop_device(self):
"""Gets the loop device where the image file is attached to.
Returns:
The path of the loop device where the image file is attached to or
None if no loop device is attaching the image file.
"""
return self._loop_device
@property
def mount_dir(self):
"""Gets the directory where the image file is mounted to.
Returns:
The directory where the image file is mounted to or None if no
mount directory has been created.
"""
return self._mount_dir.name if self._mount_dir else None
def create(self):
"""Creates a zero-filled image file with the specified size.
The created image file is temporary and removed when clean()
is called.
"""
self.clean()
self._image_file = autotemp.tempfile(unique_id='fsImage')
try:
logging.debug('Creating zero-filled image file at "%s"',
self._image_file.name)
utils.run('dd if=/dev/zero of=%s bs=%s count=%s' %
(self._image_file.name, self._block_size,
self._block_count))
except error.CmdError as exc:
self._remove_image_file()
message = 'Failed to create filesystem image: %s' % exc
raise RuntimeError(message)
def clean(self):
"""Removes the image file if one has been created.
Before removal, the image file is detached from the loop device that
it is attached to.
"""
self.detach_from_loop_device()
self._remove_image_file()
def attach_to_loop_device(self):
"""Attaches the created image file to a loop device.
Creates the image file, if one has not been created, by calling
create().
Returns:
The path of the loop device where the image file is attached to.
"""
if self._loop_device:
return self._loop_device
if not self._image_file:
self.create()
logging.debug('Attaching image file "%s" to loop device',
self._image_file.name)
utils.run('losetup -f %s' % self._image_file.name)
output = utils.system_output('losetup -j %s' % self._image_file.name)
# output should look like: "/dev/loop0: [000d]:6329 (/tmp/test.img)"
self._loop_device = output.split(':')[0]
logging.debug('Attached image file "%s" to loop device "%s"',
self._image_file.name, self._loop_device)
self._loop_device_stat = os.stat(self._loop_device)
logging.debug('Loop device "%s" (uid=%d, gid=%d, permissions=%04o)',
self._loop_device,
self._loop_device_stat.st_uid,
self._loop_device_stat.st_gid,
stat.S_IMODE(self._loop_device_stat.st_mode))
return self._loop_device
def detach_from_loop_device(self):
"""Detaches the image file from the loop device."""
if not self._loop_device:
return
self.unmount()
logging.debug('Cleaning up remaining mount points of loop device "%s"',
self._loop_device)
utils.run('umount -f %s' % self._loop_device, ignore_status=True)
logging.debug('Restore ownership/permissions of loop device "%s"',
self._loop_device)
os.chmod(self._loop_device,
stat.S_IMODE(self._loop_device_stat.st_mode))
os.chown(self._loop_device,
self._loop_device_stat.st_uid, self._loop_device_stat.st_gid)
logging.debug('Detaching image file "%s" from loop device "%s"',
self._image_file.name, self._loop_device)
utils.run('losetup -d %s' % self._loop_device)
self._loop_device = None
def format(self):
"""Formats the image file as the specified filesystem."""
self.attach_to_loop_device()
try:
logging.debug('Formatting image file at "%s" as "%s" filesystem',
self._image_file.name, self._filesystem_type)
utils.run('yes | mkfs -t %s %s %s' %
(self._filesystem_type, ' '.join(self._mkfs_options),
self._loop_device))
logging.debug('blkid: %s', utils.system_output(
'blkid -c /dev/null %s' % self._loop_device,
ignore_status=True))
except error.CmdError as exc:
message = 'Failed to format filesystem image: %s' % exc
raise RuntimeError(message)
def mount(self, options=None):
"""Mounts the image file to a directory.
Args:
options: An optional list of mount options.
"""
if self._mount_dir:
return self._mount_dir.name
if options is None:
options = []
options_arg = ','.join(options)
if options_arg:
options_arg = '-o ' + options_arg
self.attach_to_loop_device()
self._mount_dir = autotemp.tempdir(unique_id='fsImage')
try:
logging.debug('Mounting image file "%s" (%s) to directory "%s"',
self._image_file.name, self._loop_device,
self._mount_dir.name)
utils.run('mount -t %s %s %s %s' %
(self._mount_filesystem_type, options_arg,
self._loop_device, self._mount_dir.name))
except error.CmdError as exc:
self._remove_mount_dir()
message = ('Failed to mount virtual filesystem image "%s": %s' %
(self._image_file.name, exc))
raise RuntimeError(message)
return self._mount_dir.name
def unmount(self):
"""Unmounts the image file from the mounted directory."""
if not self._mount_dir:
return
try:
logging.debug('Unmounting image file "%s" (%s) from directory "%s"',
self._image_file.name, self._loop_device,
self._mount_dir.name)
utils.run('umount %s' % self._mount_dir.name)
except error.CmdError as exc:
message = ('Failed to unmount virtual filesystem image "%s": %s' %
(self._image_file.name, exc))
raise RuntimeError(message)
finally:
self._remove_mount_dir()
def get_volume_label(self):
"""Gets volume name information of |self._loop_device|
@return a string with volume name if it exists.
"""
# This script is run as root in a normal autotest run,
# so this works: It doesn't have access to the necessary info
# when run as a non-privileged user
cmd = "blkid -c /dev/null -o udev %s" % self._loop_device
output = utils.system_output(cmd, ignore_status=True)
for line in output.splitlines():
udev_key, udev_val = line.split('=')
if udev_key == 'ID_FS_LABEL':
return udev_val
return None