Merge pull request #1068 from hereischen/hub1324310

Add usb device check cases

qemu/tests/cfg/usb_device_check.cfg

+- usb_device_check:
+    virt_test_type = qemu
+    no Host_RHEL.m6.u1
+    not_preprocess = yes
+    kill_vm_on_error = yes
+    usbs = "usbtest"
+    usb_bus = "usbtest.0"
+    cmd_timeout = 300
+    chk_usb_info_cmd = "lsusb -v"
+    type = usb_device_check
+    # usb device info name
+    usb-hub = "QEMU USB Hub"
+
+    # usb controllers
+    variants:
+        - piix3-uhci:
+            usb_type_usbtest = piix3-usb-uhci
+            usb_controller = uhci
+            usb_max_port_usbtest = 2
+            no ppc64 ppc64le
+        - piix4-uhci:
+            usb_type_usbtest = piix4-usb-uhci
+            usb_controller = uhci
+            usb_max_port_usbtest = 2
+            no ppc64 ppc64le
+        - ich9-uhci:
+            usb_type_usbtest = ich9-usb-uhci6
+            usb_controller = uhci
+            usb_max_port_usbtest = 2
+            no ppc64 ppc64le
+            no Host_RHEL.m6
+        - ich9-ehci:
+            usb_type_usbtest = ich9-usb-ehci1
+            usb_controller = ehci
+            usb_max_port_usbtest = 6
+        - usb-ehci:
+            usb_type_usbtest = usb-ehci
+            usb_controller = ehci
+            usb_max_port_usbtest = 6
+        - nec-xhci:
+            no RHEL.3
+            no RHEL.4
+            no RHEL.5
+            no Host_RHEL.m6
+            usb_type_usbtest = nec-usb-xhci
+            usb_controller = xhci
+            usb_max_port_usbtest = 4
+            Host_RHEL:
+                no Win2000, WinXP, Win2003, WinVista, Win7, Win2008
+
+    # usb topology
+    variants:
+        - multi_hubs_in_series:
+            no usb-ehci
+            usb_topology = '{"usb-hub":5}'

qemu/tests/usb_common.py

+import json
+
+from collections import OrderedDict, Counter
+
+from virttest import utils_misc
+
+
+def parse_usb_topology(params):
+    """
+    Parse the usb devices topology to the params.
+
+    :param params: Dictionary with the test parameters.
+    :return: A list of dictionary ({usb_type_d0: usb_type}) of specified
+             usb topology.
+    """
+    params["usb_devices"] = ""
+    # usb topology
+    usb_topology = json.loads(params["usb_topology"],
+                              object_pairs_hook=OrderedDict)
+    parsed_devs = []
+    for key, value in usb_topology.iteritems():
+        for i in xrange(value):
+            params["usb_devices"] += " d%s" % len(parsed_devs)
+            usb_type = '{"usb_type_d%s": "%s"}' % (len(parsed_devs), key)
+            params.update(json.loads(usb_type))
+            parsed_devs.append(json.loads(usb_type))
+    return parsed_devs
+
+
+def collect_usb_dev(params, vm, parsed_devs):
+    """
+    Collect usb device information of parsed_devs.
+
+    :param params: Dictionary with the test parameters.
+    :param vm: Virtual machine object.
+    :param parsed_devs: A list of parsed usb devices.
+    :return: A list of tuple contains (id, type, port) of a
+             usb device.
+    """
+    devs = []
+    for parsed_dev in parsed_devs:
+        key = list(parsed_dev.keys())[0]
+        usb_dev_id = "usb-%s" % key[9:]
+        usb_dev_type = params[parsed_dev[key]]
+        usb_dev_port = str(vm.devices.get(usb_dev_id).get_param("port"))
+        devs.append((usb_dev_id, usb_dev_type, usb_dev_port))
+    return devs
+
+
+def verify_usb_device_in_monitor(vm, devs):
+    """
+    Verify usb device information in the qemu monitor.
+
+    This function is using "info usb", it is much more strict verification,
+    as it compares the output and devs one by one. However,RHEL does not
+    support it.
+
+    :param vm: Virtual machine object.
+    :param devs: A list of detailed device information.
+    :return: A tuple (status, output) where status is the verification result
+             and output is the detail information.
+    """
+    output = str(vm.monitor.info("usb")).splitlines()
+    for dev in devs:
+        for chk_str in dev:
+            result = next((True for info in output if chk_str in info),
+                          False)
+            if result is False:
+                return (False, "[%s] is not in the monitor info" % chk_str)
+    return (True, "all given devices in the monitor info")
+
+
+def verify_usb_device_in_monitor_qtree(vm, devs):
+    """
+    Verify usb device information in the qemu monitor.
+
+    This function is using "info qtree" to compatible with RHEL.
+
+    :param vm: Virtual machine object.
+    :param devs: A list of detailed device information.
+    :return: A tuple (status, output) where status is the verification result
+             and output is the detail information.
+    """
+    output = str(vm.monitor.info("qtree"))
+    for dev in devs:
+        for chk_str in dev:
+            if chk_str not in output:
+                return (False, "[%s] is not in the monitor info" % chk_str)
+    return (True, "all given devices are verified in the monitor info")
+
+
+def verify_usb_device_in_guest(params, session, devs):
+    """
+    Verify usb device information in the guest.
+
+    :param params: Dictionary with the test parameters.
+    :param session: Session object.
+    :return: A tuple (status, output) where status is the verification result
+             and output is the detail information
+    """
+    def _verify_guest_usb():
+        output = session.cmd_output(params["chk_usb_info_cmd"],
+                                    float(params["cmd_timeout"]))
+        # each dev must in the output
+        for dev in devs:
+            if dev[1] not in output:
+                return False
+        # match number of devices
+        counter = Counter(dev[1] for dev in devs)
+        for k, v in counter.items():
+            if output.count(k) != v:
+                return False
+        return True
+
+    res = utils_misc.wait_for(_verify_guest_usb,
+                              float(params["cmd_timeout"]),
+                              text="wait for getting guest usb devices info")
+
+    if res:
+        return (True, "all given devices are verified in the guest")
+    else:
+        return (False, "failed to verify usb devices in guest")

qemu/tests/usb_device_check.py

+import logging
+
+from virttest import env_process
+from virttest import error_context
+from qemu.tests.usb_common import (parse_usb_topology,
+                                   collect_usb_dev,
+                                   verify_usb_device_in_monitor_qtree,
+                                   verify_usb_device_in_guest)
+
+
+@error_context.context_aware
+def run(test, params, env):
+    """
+    Check the usb devices.
+
+    1) Boot up guest with usb devices
+    2) verify usb devices in monitor
+    3) verify usb devices in guest
+
+    :param test:   QEMU test object.
+    :param params: Dictionary with the test parameters.
+    :param env:    Dictionary with test environment.
+    """
+    def _check_test_step_result(result, output):
+        if result:
+            logging.info(output)
+        else:
+            test.fail(output)
+
+    # parse the usb toplogy from cfg
+    parsed_devs = parse_usb_topology(params)
+
+    logging.info("starting vm according to the usb toplogy")
+    env_process.process(test, params, env,
+                        env_process.preprocess_image,
+                        env_process.preprocess_vm)
+    vm = env.get_vm(params["main_vm"])
+    vm.verify_alive()
+
+    devs = collect_usb_dev(params, vm, parsed_devs)
+
+    error_context.context("verify usb devices information in monitor...",
+                          logging.info)
+    result, output = verify_usb_device_in_monitor_qtree(vm, devs)
+    _check_test_step_result(result, output)
+
+    login_timeout = int(params.get("login_timeout", 360))
+    session = vm.wait_for_login(timeout=login_timeout)
+
+    error_context.context("verify usb devices information in guest...",
+                          logging.info)
+    result, output = verify_usb_device_in_guest(params, session, devs)
+    _check_test_step_result(result, output)
+
+    session.close()