client/tests/kvm/tests/usb.py

import logging, os
from autotest_lib.client.common_lib import error


@error.context_aware
def run_usb(test, params, env):
    """
    Test usb device of guest

    1) create a image file by qemu-img
    2) boot up a guest add this file as a usb device
    3) check usb device information by execute monitor/guest command

    @param test: kvm test object
    @param params: Dictionary with the test parameters
    @param env: Dictionary with test environment.
    """
    vm = env.get_vm(params["main_vm"])
    vm.create()

    session = vm.wait_for_login(timeout=int(params.get("login_timeout", 360)))

    output = vm.monitor.cmd("info usb")
    if "Product QEMU USB MSD" not in output:
        logging.debug(output)
        raise error.TestFail("Could not find mass storage device")

    output = session.cmd("lsusb -v")
    # No bus specified, default using "usb.0" for "usb-storage"
    for i in ["ID 0000:0000", "Mass Storage", "SCSI", "QEMU USB HARDDRIVE"]:
        if i not in output:
            logging.debug(output)
            raise error.TestFail("No '%s' in the output of 'lsusb -v'" % i)

    output = session.cmd("fdisk -l")
    if params.get("fdisk_string") not in output:
        for line in output.splitlines():
            logging.debug(line)
        raise error.TestFail("Could not detect the usb device on fdisk output")

    error.context("Formatting USB disk")
    devname = session.cmd("ls /dev/disk/by-path/* | grep usb").strip()
    session.cmd("yes | mkfs %s" % devname,
                timeout=int(params.get("format_timeout")))

    error.context("Mounting USB disk")
    session.cmd("mount %s /mnt" % devname)

    error.context("Creating comparison file")
    c_file = '/tmp/usbfile'
    session.cmd("dd if=/dev/random of=%s bs=1M count=1" % c_file)

    error.context("Copying %s to USB disk" % c_file)
    session.cmd("cp %s /mnt" % c_file)

    error.context("Unmounting USB disk before file comparison")
    session.cmd("umount %s" % devname)

    error.context("Mounting USB disk for file comparison")
    session.cmd("mount %s /mnt" % devname)

    error.context("Determining md5sum for file on root fs and in USB disk")
    md5_root = session.cmd("md5sum %s" % c_file).strip()
    md5_usb = session.cmd("md5sum /mnt/%s" % os.path.basename(c_file)).strip()
    md5_root = md5_root.split()[0]
    md5_usb = md5_usb.split()[0]

    error.context("")
    if md5_root != md5_usb:
        raise error.TestError("MD5 mismatch between file on root fs and on "
                              "USB disk")

    error.context("Unmounting USB disk after file comparison")
    session.cmd("umount %s" % devname)

    error.context("Checking if there are I/O error messages in dmesg")
    output = session.get_command_output("dmesg")
    io_error_msg = []
    for line in output.splitlines():
        if "Buffer I/O error" in line:
            io_error_msg.append(line)

    if io_error_msg:
        e_msg = "IO error found on guest's dmesg when formatting USB device"
        logging.error(e_msg)
        for line in io_error_msg:
            logging.error(line)
        raise error.TestFail(e_msg)

    session.close()