"""
mbed SDK
Copyright (c) 2018-2019 ARM Limited
SPDX-License-Identifier: Apache-2.0

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from __future__ import print_function

from mbed_host_tests import BaseHostTest
from argparse import ArgumentParser
import time
import sys
import inspect
from threading import Thread, Event, Timer
import array
import random
import os
import traceback

import usb.core
from usb.util import build_request_type
from usb.util import CTRL_OUT, CTRL_IN
from usb.util import CTRL_TYPE_STANDARD, CTRL_TYPE_CLASS, CTRL_TYPE_VENDOR
from usb.util import (CTRL_RECIPIENT_DEVICE, CTRL_RECIPIENT_INTERFACE,
                      CTRL_RECIPIENT_ENDPOINT, CTRL_RECIPIENT_OTHER)
from usb.util import (DESC_TYPE_DEVICE, DESC_TYPE_CONFIG, DESC_TYPE_STRING,
                      DESC_TYPE_INTERFACE, DESC_TYPE_ENDPOINT)

import struct

if sys.platform.startswith('win'):
    # Use libusb0 on Windows. libusb1 implementation for Windows
    # does not support all features necessary for testing.
    import usb.backend.libusb0
    USB_BACKEND = usb.backend.libusb0.get_backend()
else:
    # Use a default backend on other platforms.
    USB_BACKEND = None

def get_interface(dev, interface, alternate=0):
    intf = None
    for active_if in dev.get_active_configuration():
        if active_if.bInterfaceNumber == interface and active_if.bAlternateSetting == alternate:
            assert intf is None, "duplicate interface"
            intf = active_if
    return intf

VENDOR_TEST_CTRL_IN = 1
VENDOR_TEST_CTRL_OUT = 2
VENDOR_TEST_CTRL_NONE = 3
VENDOR_TEST_CTRL_IN_DELAY = 4
VENDOR_TEST_CTRL_OUT_DELAY = 5
VENDOR_TEST_CTRL_NONE_DELAY = 6
VENDOR_TEST_CTRL_IN_STATUS_DELAY = 7
VENDOR_TEST_CTRL_OUT_STATUS_DELAY = 8
VENDOR_TEST_CTRL_IN_SIZES = 9
VENDOR_TEST_CTRL_OUT_SIZES = 10
VENDOR_TEST_RW_RESTART = 11
VENDOR_TEST_ABORT_BUFF_CHECK = 12
VENDOR_TEST_UNSUPPORTED_REQUEST = 32

REQUEST_GET_STATUS = 0
REQUEST_CLEAR_FEATURE = 1
REQUEST_SET_FEATURE = 3
REQUEST_SET_ADDRESS = 5
REQUEST_GET_DESCRIPTOR = 6
REQUEST_SET_DESCRIPTOR = 7
REQUEST_GET_CONFIGURATION = 8
REQUEST_SET_CONFIGURATION = 9
REQUEST_GET_INTERFACE = 10
REQUEST_SET_INTERFACE = 11
REQUEST_SYNCH_FRAME = 12

FEATURE_ENDPOINT_HALT = 0
FEATURE_DEVICE_REMOTE_WAKEUP = 1

DEVICE_QUALIFIER_DESC_SIZE = 10
DESC_TYPE_DEVICE_QUALIFIER = 0x06

DEVICE_DESC_SIZE = 18
device_descriptor_parser = struct.Struct('BBHBBBBHHHBBBB')
device_descriptor_keys = ['bLength', 'bDescriptorType', 'bcdUSB', 'bDeviceClass',
                          'bDeviceSubClass', 'bDeviceProtocol', 'bMaxPacketSize0',
                          'idVendor', 'idProduct', 'bcdDevice', 'iManufacturer',
                          'iProduct', 'iSerialNumber', 'bNumConfigurations']

CONFIGURATION_DESC_SIZE = 9
configuration_descriptor_parser = struct.Struct('BBHBBBBB')
configuration_descriptor_keys = ['bLength', 'bDescriptorType', 'wTotalLength',
                                 'bNumInterfaces', 'bConfigurationValue',
                                 'iConfiguration', 'bmAttributes', 'bMaxPower']

INTERFACE_DESC_SIZE = 9
interface_descriptor_parser = struct.Struct('BBBBBBBBB')
interface_descriptor_keys = ['bLength', 'bDescriptorType', 'bInterfaceNumber',
                             'bAlternateSetting', 'bNumEndpoints',
                             'bInterfaceClass', 'bInterfaceSubClass',
                             'bInterfaceProtocol', 'iInterface']

ENDPOINT_DESC_SIZE = 7
interface_descriptor_parser = struct.Struct('BBBBBHB')
interface_descriptor_keys = ['bLength', 'bDescriptorType', 'bEndpointAddress',
                             'bmAttributes', 'wMaxPacketSize', 'bInterval']

ENDPOINT_TYPE_NAMES = {
    usb.ENDPOINT_TYPE_BULK: 'BULK',
    usb.ENDPOINT_TYPE_CONTROL: 'CONTROL',
    usb.ENDPOINT_TYPE_INTERRUPT: 'INTERRUPT',
    usb.ENDPOINT_TYPE_ISOCHRONOUS: 'ISOCHRONOUS'}

# Greentea message keys used to notify DUT of test status
MSG_KEY_TEST_CASE_FAILED = 'fail'
MSG_KEY_TEST_CASE_PASSED = 'pass'
MSG_VALUE_DUMMY = '0'


def format_local_error_msg(fmt):
    """Return an error message formatted with the last traceback entry from this file.

    The message is formatted according to fmt with data from the last traceback
    enrty internal to this file. There are 4 arguments supplied to the format
    function: filename, line_number, exc_type and exc_value.

    Returns None if formatting fails.
    """
    try:
        exc_type, exc_value, exc_traceback = sys.exc_info()
        # A list of 4-tuples (filename, line_number, function_name, text).
        tb_entries = traceback.extract_tb(exc_traceback)
        # Reuse the filename from the first tuple instead of relying on __file__:
        # 1. No need for path handling.
        # 2. No need for file extension handling (i.e. .py vs .pyc).
        name_of_this_file = tb_entries[0][0]
        last_internal_tb_entry = [tb for tb in tb_entries if tb[0] == name_of_this_file][-1]
        msg = fmt.format(
            filename=last_internal_tb_entry[0],
            line_number=last_internal_tb_entry[1],
            exc_type=str(exc_type).strip(),
            exc_value=str(exc_value).strip(),
            )
    except (IndexError, KeyError):
        msg = None
    return msg


class PyusbBasicTest(BaseHostTest):

    def test_usb_device(self, usb_dev_serial_number, test_fun, **test_fun_kwargs):
        """Find a USB device and execute a testing function.

        Search is based on usb_dev_serial_number. If the device is found, the
        test_fun is executed with its dev argument set to the device found and
        all other kwargs set as specified by test_fun_kwargs.

        The DUT is notified with either success, failure or error status.
        """
        usb_device = self.find_device(usb_dev_serial_number)
        if usb_device is None:
            self.notify_error('USB device (SN={}) not found.'.format(usb_dev_serial_number))
            return
        try:
            test_fun(usb_device, **test_fun_kwargs)
            self.notify_success()
        except RuntimeError as exc:
            self.notify_failure(exc)
        except usb.core.USBError as exc:
            error_msg = format_local_error_msg('[{filename}]:{line_number}, Dev-host transfer error ({exc_value}).')
            self.notify_failure(error_msg if error_msg is not None else exc)

    def _callback_control_basic_test(self, key, value, timestamp):
        serial_number, vendor_id, product_id = value.split(' ')
        self.test_usb_device(
            usb_dev_serial_number=serial_number,
            test_fun=control_basic_test,
            log=print,
            vendor_id=int(vendor_id),
            product_id=int(product_id)
        )

    def _callback_control_stall_test(self, key, value, timestamp):
        self.test_usb_device(
            usb_dev_serial_number=value,
            test_fun=control_stall_test,
            log=print
        )

    def _callback_control_sizes_test(self, key, value, timestamp):
        self.test_usb_device(
            usb_dev_serial_number=value,
            test_fun=control_sizes_test,
            log=print
        )

    def _callback_control_stress_test(self, key, value, timestamp):
        self.test_usb_device(
            usb_dev_serial_number=value,
            test_fun=control_stress_test,
            log=print
        )

    def _callback_device_reset_test(self, key, value, timestamp):
        self.test_usb_device(
            usb_dev_serial_number=value,
            # Advance the coroutine to the next yield statement
            # and send the usb_device to use.
            test_fun=self.device_reset_test.send
        )

    def _callback_device_soft_reconnection_test(self, key, value, timestamp):
        self.test_usb_device(
            usb_dev_serial_number=value,
            # Advance the coroutine to the next yield statement
            # and send the usb_device to use.
            test_fun=self.device_soft_reconnection_test.send
        )

    def _callback_device_suspend_resume_test(self, key, value, timestamp):
        self.test_usb_device(
            usb_dev_serial_number=value,
            # Advance the coroutine to the next yield statement
            # and send the usb_device to use.
            test_fun=self.device_suspend_resume_test.send
        )

    def _callback_repeated_construction_destruction_test(self, key, value, timestamp):
        self.test_usb_device(
            usb_dev_serial_number=value,
            # Advance the coroutine to the next yield statement
            # and send the usb_device to use.
            test_fun=self.repeated_construction_destruction_test.send
        )

    def _callback_ep_test_data_correctness(self, key, value, timestamp):
        self.test_usb_device(
            usb_dev_serial_number=value,
            test_fun=ep_test_data_correctness,
            log=print
        )

    def _callback_ep_test_halt(self, key, value, timestamp):
        self.test_usb_device(
            usb_dev_serial_number=value,
            test_fun=ep_test_halt,
            log=print
        )

    def _callback_ep_test_parallel_transfers(self, key, value, timestamp):
        self.test_usb_device(
            usb_dev_serial_number=value,
            test_fun=ep_test_parallel_transfers,
            log=print
        )

    def _callback_ep_test_parallel_transfers_ctrl(self, key, value, timestamp):
        self.test_usb_device(
            usb_dev_serial_number=value,
            test_fun=ep_test_parallel_transfers_ctrl,
            log=print
        )

    def _callback_ep_test_abort(self, key, value, timestamp):
        self.test_usb_device(
            usb_dev_serial_number=value,
            test_fun=ep_test_abort,
            log=print
        )

    def _callback_ep_test_data_toggle(self, key, value, timestamp):
        self.test_usb_device(
            usb_dev_serial_number=value,
            test_fun=ep_test_data_toggle,
            log=print
        )

    def _callback_reset_support(self, key, value, timestamp):
        status = "false" if sys.platform == "darwin" else "true"
        self.log("Reset supported: %s" % status)
        self.send_kv("placeholder", status)

    def find_device(self, serial_number):
        # to make it more reliable, 20 retries in 2[s]
        for _ in range(20):
            dev = usb.core.find(custom_match=TestMatch(serial_number), backend=USB_BACKEND)
            if dev is not None:
                break
            time.sleep(0.1)
        return dev

    def notify_success(self, value=None, msg=''):
        """Report a host side test success to the DUT."""
        if msg:
            self.log('TEST PASSED: {}'.format(msg))
        if value is None:
            value = MSG_VALUE_DUMMY
        self.send_kv(MSG_KEY_TEST_CASE_PASSED, value)

    def notify_failure(self, msg):
        """Report a host side test failure to the DUT."""
        self.log('TEST FAILED: {}'.format(msg))
        self.send_kv(MSG_KEY_TEST_CASE_FAILED, MSG_VALUE_DUMMY)

    def notify_error(self, msg):
        """Terminate the test with an error msg."""
        self.log('TEST ERROR: {}'.format(msg))
        self.notify_complete(None)

    def setup(self):
        self.__result = False

        self.device_reset_test = device_reset_test(log=print)
        self.device_reset_test.send(None)

        self.device_soft_reconnection_test = device_soft_reconnection_test(log=print)
        self.device_soft_reconnection_test.send(None)

        self.device_suspend_resume_test = device_suspend_resume_test(log=print)
        self.device_suspend_resume_test.send(None)

        self.repeated_construction_destruction_test = repeated_construction_destruction_test(log=print)
        self.repeated_construction_destruction_test.send(None)

        self.register_callback('control_basic_test', self._callback_control_basic_test)
        self.register_callback('control_stall_test', self._callback_control_stall_test)
        self.register_callback('control_sizes_test', self._callback_control_sizes_test)
        self.register_callback('control_stress_test', self._callback_control_stress_test)
        self.register_callback('device_reset_test', self._callback_device_reset_test)
        self.register_callback('device_soft_reconnection_test', self._callback_device_soft_reconnection_test)
        self.register_callback('device_suspend_resume_test', self._callback_device_suspend_resume_test)
        self.register_callback('repeated_construction_destruction_test', self._callback_repeated_construction_destruction_test)

        self.register_callback('ep_test_data_correctness', self._callback_ep_test_data_correctness)
        self.register_callback('ep_test_halt', self._callback_ep_test_halt)
        self.register_callback('ep_test_parallel_transfers', self._callback_ep_test_parallel_transfers)
        self.register_callback('ep_test_parallel_transfers_ctrl', self._callback_ep_test_parallel_transfers_ctrl)
        self.register_callback('ep_test_abort', self._callback_ep_test_abort)
        self.register_callback('ep_test_data_toggle', self._callback_ep_test_data_toggle)

        self.register_callback('reset_support', self._callback_reset_support)

    def result(self):
        return self.__result

    def teardown(self):
        pass


class TestMatch(object):

    def __init__(self, serial):
        self.serial = serial

    def __call__(self, dev):
        try:
            return dev.serial_number == self.serial
        except ValueError:
            return False


def lineno():
    """Returns the current line number in our program."""
    return inspect.currentframe().f_back.f_lineno


def raise_if_different(expected, actual, line, text=''):
    """Raise a RuntimeError if actual is different than expected."""
    if expected != actual:
        raise RuntimeError('[{}]:{}, {} Got {!r}, expected {!r}'.format(__file__, line, text, actual, expected))


def raise_unconditionally(line, text=''):
    """Raise a RuntimeError unconditionally."""
    raise RuntimeError('[{}]:{}, {}'.format(__file__, line, text))


def control_basic_test(dev, vendor_id, product_id, log):
    get_set_configuration_test(dev, log)
    get_set_interface_test(dev, log)
    get_status_test(dev, log)
    set_clear_feature_test(dev, log)
    get_descriptor_test(dev, vendor_id, product_id, log)
    set_descriptor_test(dev, log)


def get_set_configuration_test(dev, log):
    """
    Test device configuration/deconfiguration

    Given an initialized USB (HOST <---> DUT connection established)
    When device configuration is checked just after initialization
    Then get_configuration returns 1 (default configuration is set)
    When device is deconfigured
    Then get_configuration returns 0 (no configuration is set)
    When each from supported configurations is set
    Then the configuration is set correctly
    """

    print("<<< get_set_configuration_test >>>")
    # check if dafault(1) configuration set
    try:
        ret = usb.control.get_configuration(dev)
        raise_if_different(1, ret, lineno(), 'Invalid configuration.')
    except usb.core.USBError as error:
        raise_unconditionally(lineno(), 'get_configuration request failed ({}).'.format(str(error).strip()))

    cfg = dev.get_active_configuration()
    for intf in cfg:
        usb.util.release_interface(dev, intf)

    # deconfigure the device
    try:
        ret = dev.set_configuration(0)
    except usb.core.USBError as error:
        raise_unconditionally(lineno(), 'set_configuration request (deconfigure) failed ({}).'.format(str(error).strip()))

    # check if deconfigured
    try:
        ret = usb.control.get_configuration(dev)
        raise_if_different(0, ret, lineno(), 'Invalid configuration.')
        print("device deconfigured - OK")
    except usb.core.USBError as error:
        raise_unconditionally(lineno(), 'get_configuration request failed ({}).'.format(str(error).strip()))

    # for every configuration
    for cfg in dev:
        try:
            # set configuration
            ret = cfg.set()
        except usb.core.USBError as error:
            raise_unconditionally(lineno(), 'set_configuration request failed ({}).'.format(str(error).strip()))

        # check if configured
        try:
            ret = usb.control.get_configuration(dev)
            raise_if_different(cfg.bConfigurationValue, ret, lineno(), 'Invalid configuration.')
            print("configuration {} set - OK ".format(cfg.bConfigurationValue))
        except usb.core.USBError as error:
            raise_unconditionally(lineno(), 'get_configuration request failed ({}).'.format(str(error).strip()))
        # test control data transfer after configuration set
        control_data_test(dev, [64, 256], log)
    print("") # new line


def get_set_interface_test(dev, log):
    """
    Test device interface setting

    Given an initialized USB (HOST <---> DUT connection established)
    When each altsetting from every supported configuration is set
    Then the interface altsetting is set correctly
    """

    print("<<< get_set_interface_test >>>")
    # for every configuration
    for cfg in dev:
        cfg.set()
        # for every interface
        for intf in cfg:
            intf.set_altsetting()
            altsett = usb.control.get_interface(dev, intf.bInterfaceNumber)
            raise_if_different(intf.bAlternateSetting, altsett, lineno(), text='Wrong alternate setting for interface {}'.format(intf.bInterfaceNumber))
            print("cfg({}) inteface {}.{} set - OK".format(cfg.bConfigurationValue, intf.bInterfaceNumber, intf.bAlternateSetting))
            control_data_test(dev, [64, 256], log)

        release_interfaces(dev)

    restore_default_configuration(dev)

    # test control data transfer after default interface restoring
    control_data_test(dev, [64, 256], log)
    print("") # new line


def get_status_test(dev, log):
    """
    Test device/interface/endpoint status

    Given an initialized USB (HOST <---> DUT connection established)
    When device status is checked
    Then status is within allowed values (see status bits description below)
    When control endpoint status is checked
    Then control endpoint status is 0
    When status of each interface from every supported configuration is checked
    Then interface status is 0
    When status of each endpoint in every allowed device interface/configuration combination is checked
    Then endpoint status is 0 (not halted)
    """

    print("<<< get_status_test >>>")
    # check device status
    ret = get_status(dev, CTRL_RECIPIENT_DEVICE)
    # Status bits
    # ret == 0b01    (D0)Self Powered
    # ret == 0b10    (D1)Remote Wakeup
    # (D2 - D15 reserved) Must be set to 0
    if(ret < 0 or ret > 3):
        raise_unconditionally(lineno(), "GET_STATUS on DEVICE failed")

    # check endpoint 0 status
    ret = get_status(dev, CTRL_RECIPIENT_ENDPOINT, 0)
    # Status bits
    # ret == 0b1    (D0)endpoint Halt
    # (D1 - D15 reserved) Must be set to 0
    # endpoint 0 can't be halted ret == 0
    raise_if_different(0, ret, lineno(), "GET_STATUS on ENDPOINT 0 should return 0")

    # for every configuration
    for cfg in dev:
        cfg.set()
        raise_if_different(cfg.bConfigurationValue, usb.control.get_configuration(dev), lineno(), "Configuration {} set failed".format(cfg.bConfigurationValue))

        for intf in cfg:
            intf.set_altsetting()
            # check interface status
            ret = get_status(dev, CTRL_RECIPIENT_INTERFACE, intf.bInterfaceNumber)
            # Status bits
            # ret == 0b0
            # (D0 - D15 reserved) Must be set to 0
            if(ret != 0):
                raise_unconditionally(lineno(), "GET_STATUS on INTERFACE ({},{}) failed".format(intf.bInterfaceNumber, intf.bAlternateSetting))
            print("cfg({}) interface {}.{} status - OK".format(cfg.bConfigurationValue, intf.bInterfaceNumber, intf.bAlternateSetting))

            # on every ENDPOINT in this altsetting
            for ep in intf:
                ret = usb.control.get_status(dev, ep)
                # Status bits
                # ret == 0b1    (D0)endpoint Halt
                # (D1 - D15 reserved) Must be set to 0
                if(ret >= 1):
                    raise_unconditionally(lineno(), "GET_STATUS on ENDPOINT {} failed - endpoint halted".format(ep.bEndpointAddress))
                print("cfg({}) intf({}.{}) endpoint {} status - OK".format(cfg.bConfigurationValue, intf.bInterfaceNumber, intf.bAlternateSetting, ep.bEndpointAddress))

        release_interfaces(dev)
    restore_default_configuration(dev)
    print("") # new line


def set_clear_feature_test(dev, log):
    """
    Test set/clear feature on device/interface/endpoint

    Given an initialized USB (HOST <---> DUT connection established)
    When for each endpoint in every allowed interface/configuration combination the feature is set and then cleared
    Then selected feature is set/cleared accordingly
    """

    print("<<< set_clear_feature_test >>>")
    # TODO:
    # test set_feature on device (Remote wakeup feature not supported on DUT side)
    # test set_feature on interface (not supported at all)

    # for every configuration
    for cfg in dev:
        cfg.set()
        raise_if_different(cfg.bConfigurationValue, usb.control.get_configuration(dev), lineno(), "Configuration {} set failed".format(cfg.bConfigurationValue))

        for intf in cfg:
            intf.set_altsetting()
            # on every ENDPOINT
            for ep in intf:
                # halt endpoint
                try:
                    usb.control.set_feature(dev, FEATURE_ENDPOINT_HALT, ep)
                except usb.core.USBError as err:
                    raise_unconditionally(lineno(), 'set_feature request (halt) failed for endpoint {} ({}).'.format(ep.bEndpointAddress, str(err).strip()))

                # check if endpoint was halted
                try:
                    ret = usb.control.get_status(dev, ep)
                except usb.core.USBError as err:
                    raise_unconditionally(lineno(), 'get_status request failed for endpoint {} ({}).'.format(ep.bEndpointAddress, str(err).strip()))
                if(ret != 1):
                    raise_unconditionally(lineno(), "endpoint {} was not halted".format(ep.bEndpointAddress))
                print("cfg({}) intf({}.{})   ep {} halted - OK".format(cfg.bConfigurationValue, intf.bInterfaceNumber, intf.bAlternateSetting, ep.bEndpointAddress))

                # Control OUT CLEAR_FEATURE on endpoint - unhalt
                try:
                    usb.control.clear_feature(dev, FEATURE_ENDPOINT_HALT, ep)
                except usb.core.USBError as err:
                    raise_unconditionally(lineno(), "clear_feature request (unhalt) failed for endpoint {} ({})".format(ep.bEndpointAddress, str(err).strip()))

                # check if endpoint was unhalted
                ret = usb.control.get_status(dev, ep)
                if(ret != 0):
                    raise_unconditionally(lineno(), "endpoint {} was not unhalted".format(ep.bEndpointAddress))
                print("cfg({}) intf({}.{})   ep {} unhalted - OK".format(cfg.bConfigurationValue, intf.bInterfaceNumber, intf.bAlternateSetting, ep.bEndpointAddress))

        release_interfaces(dev)

    restore_default_configuration(dev)
    print("") # new line


def get_descriptor_test(dev, vendor_id, product_id, log):
    """
    Test device/configuration/interface/endpoint descriptors

    Given an initialized USB (HOST <---> DUT connection established)
    When device descriptor is read
    Then the descriptor content is valid
    When configuration descriptor is read
    Then the descriptor content is valid
    When interface descriptor is read
    Then the error is thrown since it is not directly accessible
    When endpoint descriptor is read
    Then the error is thrown since it is not directly accessible
    """

    print("<<< get_descriptor_test >>>")
    # device descriptor
    try:
        ret = get_descriptor(dev, (DESC_TYPE_DEVICE << 8) | (0 << 0), 0, DEVICE_DESC_SIZE)
        dev_desc = dict(zip(device_descriptor_keys, device_descriptor_parser.unpack(ret)))
        raise_if_different(DEVICE_DESC_SIZE, dev_desc['bLength'], lineno(), text='Wrong device descriptor size.')
        raise_if_different(vendor_id, dev_desc['idVendor'], lineno(), text='Wrong vendor ID.')
        raise_if_different(product_id, dev_desc['idProduct'], lineno(), text='Wrong product ID.')
    except usb.core.USBError:
        raise_unconditionally(lineno(), "Requesting device descriptor failed")

    # configuration descriptor
    try:
        ret = get_descriptor(dev, (DESC_TYPE_CONFIG << 8) | (0 << 0), 0, CONFIGURATION_DESC_SIZE)
        conf_desc = dict(zip(configuration_descriptor_keys, configuration_descriptor_parser.unpack(ret)))
        raise_if_different(CONFIGURATION_DESC_SIZE, conf_desc['bLength'], lineno(), text='Wrong configuration descriptor size.')
    except usb.core.USBError:
        raise_unconditionally(lineno(), "Requesting configuration descriptor failed")

    # interface descriptor
    try:
        ret = get_descriptor(dev, (DESC_TYPE_INTERFACE << 8) | (0 << 0), 0, INTERFACE_DESC_SIZE)
        raise_unconditionally(lineno(), "Requesting interface descriptor should fail since it is not directly accessible.")
    except usb.core.USBError:
        log("interface descriptor is not directly accessible - OK")

    # endpoint descriptor
    try:
        ret = get_descriptor(dev, (DESC_TYPE_ENDPOINT << 8) | (0 << 0), 0, ENDPOINT_DESC_SIZE)
        raise_unconditionally(lineno(), "Requesting endpoint descriptor should fail since it is not directly accessible.")
    except usb.core.USBError:
        log("endpoint descriptor is not directly accessible - OK")
    print("") # new line


def set_descriptor_test(dev, log):
    """
    Test descriptor setting

    Given an initialized USB (HOST <---> DUT connection established)
    When device descriptor is to be set
    Then error is thrown since descriptor setting command is not supported by Mbed
    """

    print("<<< set_descriptor_test >>>")
    # SET_DESCRIPTOR is optional and not implemented in Mbed
    # command should fail with no action on device side

    # Control OUT SET_DESCRIPTOR
    request_type = build_request_type(CTRL_OUT, CTRL_TYPE_STANDARD,
                                      CTRL_RECIPIENT_DEVICE)
    request = REQUEST_SET_DESCRIPTOR
    value = (DESC_TYPE_DEVICE << 8) | (0 << 0)  # Descriptor Type (H) and Descriptor Index (L)
    index = 0                                   # 0 or Language ID for this request
    data = bytearray(DEVICE_DESC_SIZE)          # Descriptor data
    try:
        dev.ctrl_transfer(request_type, request, value, index, data)
        raise_unconditionally(lineno(), "set_descriptor request should fail since it is not implemented")
    except usb.core.USBError:
        log("SET_DESCRIPTOR is unsupported - OK")
    print("") # new line


def synch_frame_test(dev, log):
    """
    Test sync frame request

    Given an initialized USB (HOST <---> DUT connection established)
    When ...
    Then ...
    """

    print("<<< synch_frame_test >>>")
    # only for isochronous endpoints
    request_type = build_request_type(CTRL_IN, CTRL_TYPE_STANDARD,
                                      CTRL_RECIPIENT_ENDPOINT)
    request = REQUEST_SYNCH_FRAME
    value = 0                                   # Always 0 for this request
    index = 1                                   # Endpoint index
    length = 2                                  # Always 2 for this request (size of return data)
    try:
        ret = dev.ctrl_transfer(request_type, request, value, index, length)
        ret = ret[0] | (ret[1] << 8)
        log("synch frame ret: %d" % (ret))
    except usb.core.USBError:
        raise_unconditionally(lineno(), "SYNCH_FRAME request failed")
    print("") # new line


def control_stall_test(dev, log):
    """
    Test control endpoint stall on invalid request

    Given an initialized USB (HOST <---> DUT connection established)
    When unsupported request to control endpoint is to be sent
    Then the endpoint is stalled and error is thrown
    """

    print("<<< control_stall_test >>>")
    # Control OUT stall
    try:
        request_type = build_request_type(CTRL_OUT, CTRL_TYPE_VENDOR,
                                          CTRL_RECIPIENT_DEVICE)
        request = VENDOR_TEST_UNSUPPORTED_REQUEST
        value = 0                           # Always 0 for this request
        index = 0                           # Communication interface
        data = bytearray(64)                # Dummy data
        dev.ctrl_transfer(request_type, request, value, index, data, 5000)
        raise_unconditionally(lineno(), "Invalid request not stalled")
    except usb.core.USBError:
        log("Invalid request stalled - OK")

    # Control request with no data stage (Device-to-host)
    try:
        request_type = build_request_type(CTRL_IN, CTRL_TYPE_VENDOR,
                                          CTRL_RECIPIENT_DEVICE)
        request = VENDOR_TEST_UNSUPPORTED_REQUEST
        value = 0                           # Always 0 for this request
        index = 0                           # Communication interface
        length = 0
        dev.ctrl_transfer(request_type, request, value, index, length, 5000)
        raise_unconditionally(lineno(), "Invalid request not stalled")
    except usb.core.USBError:
        log("Invalid request stalled - OK")

    # Control request with no data stage (Host-to-device)
    try:
        request_type = build_request_type(CTRL_OUT, CTRL_TYPE_VENDOR,
                                          CTRL_RECIPIENT_DEVICE)
        request = VENDOR_TEST_UNSUPPORTED_REQUEST
        value = 0                           # Always 0 for this request
        index = 0                           # Communication interface
        length = 0
        dev.ctrl_transfer(request_type, request, value, index, length, 5000)
        raise_unconditionally(lineno(), "Invalid request not stalled")
    except usb.core.USBError:
        log("Invalid request stalled - OK")

    # Control IN stall
    try:
        request_type = build_request_type(CTRL_IN, CTRL_TYPE_VENDOR,
                                          CTRL_RECIPIENT_DEVICE)
        request = VENDOR_TEST_UNSUPPORTED_REQUEST
        value = 0                           # Always 0 for this request
        index = 0                           # Communication interface
        length = 255
        dev.ctrl_transfer(request_type, request, value, index, length, 5000)
        raise_unconditionally(lineno(), "Invalid request not stalled")
    except usb.core.USBError:
        log("Invalid request stalled - OK")

    for i in (3, 4, 5):
        try:
            request_type = build_request_type(CTRL_IN, CTRL_TYPE_STANDARD,
                                              CTRL_RECIPIENT_DEVICE)
            request = 0x6                   # GET_DESCRIPTOR
            value = (0x03 << 8) | (i << 0)  # String descriptor index
            index = 0                       # Communication interface
            length = 255
            resp = dev.ctrl_transfer(request_type, request, value, index, length, 5000)
        except usb.core.USBError:
            raise_unconditionally(lineno(), "Requesting string failed i: " + str(i))

    for i in (6, 7):
        try:
            request_type = build_request_type(CTRL_IN, CTRL_TYPE_STANDARD,
                                              CTRL_RECIPIENT_DEVICE)
            request = 0x6                   # GET_DESCRIPTOR
            value = (0x03 << 8) | (i << 0)  # String descriptor index
            index = 0                       # Communication interface
            length = 255
            resp = dev.ctrl_transfer(request_type, request, value, index, length, 5000)
            raise_unconditionally(lineno(), "Requesting string passed i: " + str(i))
        except usb.core.USBError:
            log("Requesting string %s failed - OK" % i)
    print("") # new line


def control_sizes_test(dev, log):
    """
    Test various data sizes in control transfer

    Given an initialized USB (HOST <---> DUT connection established)
    When control data in each tested size is sent
    Then read data should match sent data
    """

    list = [1, 2, 3, 7, 8, 9, 15, 16, 17, 31, 32, 33, 63, 64, 65, 127, 128, 129, 255, 256, 257, 511, 512, 513, 1023, 1024, 1025, 2047, 2048]
    control_data_test(dev, list, log)


def control_data_test(dev, sizes_list, log):
    # Test control requests of various data stage sizes (1,8,16,32,64,255,256,...)
    count = 1
    for i in sizes_list:
        request_type = build_request_type(CTRL_OUT, CTRL_TYPE_VENDOR,
                                              CTRL_RECIPIENT_DEVICE)
        request = VENDOR_TEST_CTRL_OUT_SIZES
        value = i                           # Size of data the device should actually read
        index = 0                           # Unused - set for debugging only
        data = bytearray(os.urandom(i))     # Dummy data

        try:
            dev.ctrl_transfer(request_type, request, value, index, data, 5000)
        except usb.core.USBError:
            raise_unconditionally(lineno(), "VENDOR_TEST_CTRL_OUT_SIZES failed ")

        request_type = build_request_type(CTRL_IN, CTRL_TYPE_VENDOR,
                                              CTRL_RECIPIENT_DEVICE)
        request = VENDOR_TEST_CTRL_IN_SIZES
        value = 0                           # Size of data the device should actually send
        index = 0                           # Unused - set for debugging only
        length = i
        try:
            ret = dev.ctrl_transfer(request_type, request, value, index, length, 5000)
            raise_if_different(i, len(ret), lineno(), "send/receive data is the wrong size")
            for j in range(0, i):
                raise_if_different(data[j], ret[j], lineno(), "send/receive data mismatch")
        except usb.core.USBError:
            raise_unconditionally(lineno(), "VENDOR_TEST_CTRL_IN_SIZES failed")
        count += 1


def control_stress_test(dev, log):
    """
    Test various patterns of control transfers

    Given an initialized USB (HOST <---> DUT connection established)
    When stress control transfer with a data in stage is performed
    Then transfer ends with success
    When stress control transfer with a data out stage followed by a control transfer with a data in stage is performed
    Then transfer ends with success
    When stress control transfer with a data out stage is performed
    Then transfer ends with success
    """

    # Some devices have had problems with back-to-back
    # control transfers. Intentionally send these sequences
    # to make sure they are properly handled.
    count = 0
    for _ in range(100):
        # Control transfer with a data in stage
        request_type = build_request_type(CTRL_IN, CTRL_TYPE_VENDOR,
                                          CTRL_RECIPIENT_DEVICE)
        request = VENDOR_TEST_CTRL_IN
        value = 8                           # Size of data the device should actually send
        index = count                       # Unused - set for debugging only
        length = 255
        dev.ctrl_transfer(request_type, request, value, index, length, 5000)
        count += 1

    for _ in range(100):
        # Control transfer with a data out stage followed
        # by a control transfer with a data in stage
        request_type = build_request_type(CTRL_OUT, CTRL_TYPE_VENDOR,
                                          CTRL_RECIPIENT_DEVICE)
        request = VENDOR_TEST_CTRL_OUT
        value = 8                           # Size of data the device should actually read
        index = count                       # Unused - set for debugging only
        data = bytearray(8)                 # Dummy data
        dev.ctrl_transfer(request_type, request, value, index, data, 5000)
        count += 1

        request_type = build_request_type(CTRL_IN, CTRL_TYPE_VENDOR,
                                          CTRL_RECIPIENT_DEVICE)
        request = VENDOR_TEST_CTRL_IN
        value = 8                           # Size of data the device should actually send
        index = count                       # Unused - set for debugging only
        length = 255
        dev.ctrl_transfer(request_type, request, value, index, length, 5000)
        count += 1

    for _ in range(100):
        # Control transfer with a data out stage
        request_type = build_request_type(CTRL_OUT, CTRL_TYPE_VENDOR,
                                          CTRL_RECIPIENT_DEVICE)
        request = VENDOR_TEST_CTRL_OUT
        value = 8                           # Size of data the device should actually read
        index = count                       # Unused - set for debugging only
        data = bytearray(8)                 # Dummy data
        dev.ctrl_transfer(request_type, request, value, index, data, 5000)
        count += 1


def find_ep_pair(intf, endpoint_type):
    """Find an OUT and IN endpoint pair.

    Raise a RuntimeError if any endpoint could not be found
    or wMaxPacketSize is not equal for both endpoints.
    """
    ep_out = usb.util.find_descriptor(
        intf, custom_match=lambda e:
        usb.util.endpoint_type(e.bmAttributes) == endpoint_type and
        usb.util.endpoint_direction(e.bEndpointAddress) == usb.ENDPOINT_OUT)
    ep_in = usb.util.find_descriptor(
        intf, custom_match=lambda e:
        usb.util.endpoint_type(e.bmAttributes) == endpoint_type and
        usb.util.endpoint_direction(e.bEndpointAddress) == usb.ENDPOINT_IN)
    if not all((ep_out, ep_in)):
        raise_unconditionally(lineno(), 'Unable to find {} endpoint pair.'
                              .format(ENDPOINT_TYPE_NAMES[endpoint_type]))
    raise_if_different(ep_out.wMaxPacketSize, ep_in.wMaxPacketSize, lineno(),
                       'wMaxPacketSize not equal for OUT and IN {} endpoints.'
                       .format(ENDPOINT_TYPE_NAMES[endpoint_type]))
    return ep_out, ep_in


def loopback_ep_test(ep_out, ep_in, payload_size):
    """Send and receive random data using OUT/IN endpoint pair.

    Verify that data received from IN endpoint is equal to
    data sent to OUT endpoint.
    Raise a RuntimeError if data does not match.
    """
    payload_out = array.array('B', (random.randint(0x00, 0xff) for _ in range(payload_size)))
    ep_out.write(payload_out)
    payload_in = ep_in.read(ep_in.wMaxPacketSize)
    raise_if_different(payload_out, payload_in, lineno(), 'Payloads mismatch.')


def random_size_loopback_ep_test(ep_out, ep_in, failure, error, seconds, log, min_payload_size=1):
    """Repeat data transfer test for OUT/IN endpoint pair for a given time.

    Set a failure Event if OUT/IN data verification fails.
    Set an error Event if unexpected USB error occurs.
    """
    end_ts = time.time() + seconds
    while time.time() < end_ts and not failure.is_set() and not error.is_set():
        payload_size = random.randint(min_payload_size, ep_out.wMaxPacketSize)
        try:
            loopback_ep_test(ep_out, ep_in, payload_size)
        except RuntimeError as err:
            log(err)
            failure.set()
            return
        except usb.USBError as err:
            log(USB_ERROR_FMT.format(err, ep_out, ep_in, payload_size))
            error.set()
            return
        time.sleep(0.01)


def halt_ep_test(dev, ep_out, ep_in, log):
    """OUT/IN endpoint halt test.

    Verify that halting an endpoint at a random point of OUT or IN transfer
    raises a USBError.
    Raise a RuntimeError if halt fails or any unexpected error occurs.
    """
    MIN_HALT_DELAY = 0.01
    MAX_HALT_DELAY = 0.1
    POST_HALT_DELAY = 0.1
    ctrl_error = Event()

    for ep in (ep_out, ep_in):
        try:
            if (usb.control.get_status(dev, ep) == 1):
                raise_unconditionally(lineno(), 'Endpoints must NOT be halted at the start of this test')
        except usb.core.USBError as err:
            raise_unconditionally(lineno(), 'Unable to get endpoint status ({!r}).'.format(err))

    ep_to_halt = random.choice([ep_out, ep_in])

    def timer_handler():
        """Halt an endpoint using a USB control request."""
        try:
            usb.control.set_feature(dev, FEATURE_ENDPOINT_HALT, ep_to_halt)
            if (usb.control.get_status(dev, ep_to_halt) != 1):
                raise RuntimeError('Invalid endpoint status after halt operation')
        except Exception as err:
            ctrl_error.set()
            log('Endpoint {:#04x} halt failed ({!r}).'.format(ep_to_halt.bEndpointAddress, err))
        # Whether the halt operation was successful or not,
        # wait a bit so the main thread has a chance to run into a USBError
        # or report the failure of halt operation.
        time.sleep(POST_HALT_DELAY)

    delay = random.uniform(MIN_HALT_DELAY, MAX_HALT_DELAY)
    delayed_halt = Timer(delay, timer_handler)
    delayed_halt.start()
    # Keep transferring data to and from the device until one of the endpoints
    # is halted.
    try:
        while delayed_halt.is_alive():
            if ctrl_error.is_set():
                break
            try:
                loopback_ep_test(ep_out, ep_in, ep_out.wMaxPacketSize)
            except usb.core.USBError as err:
                if ctrl_error.is_set():
                    break
                try:
                    ep_status = usb.control.get_status(dev, ep_to_halt)
                except usb.core.USBError as err:
                    if ctrl_error.is_set():
                        break
                    raise_unconditionally(lineno(), 'Unable to get endpoint status ({!r}).'.format(err))
                if ep_status == 1:
                    # OK, got USBError because of endpoint halt
                    return
                else:
                    raise_unconditionally(lineno(), 'Unexpected error ({!r}).'.format(err))
        if ctrl_error.is_set():
            raise_unconditionally(lineno(), 'Halting endpoint {0.bEndpointAddress:#04x} failed'
                                  .format(ep_to_halt))
    finally:
        # Always wait for the Timer thread created above.
        delayed_halt.join()
        if not ctrl_error.is_set():
            ep_out.clear_halt()
            ep_in.clear_halt()
    raise_unconditionally(lineno(), 'Halting endpoint {0.bEndpointAddress:#04x}'
                          ' during transmission did not raise USBError.'
                          .format(ep_to_halt))


def request_endpoint_loops_restart(dev):
    ctrl_kwargs = {
        'bmRequestType': build_request_type(CTRL_OUT, CTRL_TYPE_VENDOR, CTRL_RECIPIENT_DEVICE),
        'bRequest': VENDOR_TEST_RW_RESTART,
        'wValue': 0,
        'wIndex': 0}
    dev.ctrl_transfer(**ctrl_kwargs)


def request_abort_buff_check(dev, ep):
    ctrl_kwargs = {
        'bmRequestType': build_request_type(CTRL_IN, CTRL_TYPE_VENDOR, CTRL_RECIPIENT_ENDPOINT),
        'bRequest': VENDOR_TEST_ABORT_BUFF_CHECK,
        'wValue': 0,
        'wIndex': ep.bEndpointAddress,
        'data_or_wLength': 1}
    return bool(dev.ctrl_transfer(**ctrl_kwargs)[0])


USB_ERROR_FMT = str('Got {0!r} while testing endpoints '
                    '{1.bEndpointAddress:#04x}({1.wMaxPacketSize:02}) and '
                    '{2.bEndpointAddress:#04x}({2.wMaxPacketSize:02}) with '
                    'a random payload of {3} B.')


def ep_test_data_correctness(dev, log, verbose=False):
    """Test data correctness for every OUT/IN endpoint pair.

    Given a USB device with multiple OUT/IN endpoint pairs
    When the host sends random payloads up to wMaxPacketSize in size
        to an OUT endpoint of the device,
        and then the device sends data back to host using an IN endpoint
    Then data sent and received by host is equal for every endpoint pair
    """
    cfg = dev.get_active_configuration()
    for intf in cfg:
        log('interface {}, alt {} -- '.format(intf.bInterfaceNumber, intf.bAlternateSetting), end='')
        if intf.bAlternateSetting == 0:
            log('skipping the default AlternateSetting')
            continue
        log('running tests')
        intf.set_altsetting()

        bulk_out, bulk_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_BULK)
        interrupt_out, interrupt_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_INTERRUPT)
        iso_out, iso_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_ISOCHRONOUS)

        if verbose:
            log('\tbulk_out      {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(bulk_out))
            log('\tbulk_in       {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(bulk_in))
            log('\tinterrupt_out {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(interrupt_out))
            log('\tinterrupt_in  {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(interrupt_in))
            log('\tiso_out       {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(iso_out))
            log('\tiso_in        {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(iso_in))

        if verbose:
            log('Testing OUT/IN data correctness for bulk endpoint pair.')
        for payload_size in range(bulk_out.wMaxPacketSize + 1):
            try:
                loopback_ep_test(bulk_out, bulk_in, payload_size)
            except usb.USBError as err:
                raise_unconditionally(lineno(), USB_ERROR_FMT.format(err, bulk_out, bulk_in, payload_size))

        if verbose:
            log('Testing OUT/IN data correctness for interrupt endpoint pair.')
        for payload_size in range(interrupt_out.wMaxPacketSize + 1):
            try:
                loopback_ep_test(interrupt_out, interrupt_in, payload_size)
            except usb.USBError as err:
                raise_unconditionally(lineno(), USB_ERROR_FMT.format(err, interrupt_out, interrupt_in, payload_size))

#         if verbose:
#             log('Testing OUT/IN data correctness for isochronous endnpoint pair.')
#         payload_size = 128 # range(1, iso_out.wMaxPacketSize + 1):
#         try:
#             loopback_ep_test(iso_out, iso_in, payload_size)
#         except usb.USBError as err:
#             log(err)
#             raise_unconditionally(lineno(), USB_ERROR_FMT.format(err, iso_out, iso_in, payload_size))


def ep_test_halt(dev, log, verbose=False):
    """Test endpoint halt for every OUT/IN endpoint pair.

    Given a USB device with multiple OUT/IN endpoint pairs
    When the host issues an endpoint halt control request at a random point
        of OUT or IN transfer
    Then the endpoint is stalled and all further transfers fail
    """
    cfg = dev.get_active_configuration()
    for intf in cfg:
        log('interface {}, alt {} -- '.format(intf.bInterfaceNumber, intf.bAlternateSetting), end='')
        if intf.bAlternateSetting == 0:
            log('skipping the default AlternateSetting')
            continue
        log('running tests')
        intf.set_altsetting()

        bulk_out, bulk_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_BULK)
        interrupt_out, interrupt_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_INTERRUPT)

        if verbose:
            log('\tbulk_out      {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(bulk_out))
            log('\tbulk_in       {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(bulk_in))
            log('\tinterrupt_out {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(interrupt_out))
            log('\tinterrupt_in  {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(interrupt_in))

        if verbose:
            log('Testing endpoint halt at a random point of bulk transmission.')
        end_ts = time.time() + 1.0
        while time.time() < end_ts:
            halt_ep_test(dev, bulk_out, bulk_in, log)
            request_endpoint_loops_restart(dev)

        if verbose:
            log('Testing endpoint halt at a random point of interrupt transmission.')
        end_ts = time.time() + 1.0
        while time.time() < end_ts:
            halt_ep_test(dev, interrupt_out, interrupt_in, log)
            request_endpoint_loops_restart(dev)


def ep_test_parallel_transfers(dev, log, verbose=False):
    """Test simultaneous data transfers for multiple OUT/IN endpoint pairs.

    Given a USB device with multiple OUT/IN endpoint pairs
    When multiple OUT and IN endpoints are used to transfer random test data
    Then all transfers succeed
        and data received equals data sent for every endpoint pair
    """
    cfg = dev.get_active_configuration()
    for intf in cfg:
        log('interface {}, alt {} -- '.format(intf.bInterfaceNumber, intf.bAlternateSetting), end='')
        if intf.bAlternateSetting == 0:
            log('skipping the default AlternateSetting')
            continue
        log('running tests')
        intf.set_altsetting()

        bulk_out, bulk_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_BULK)
        interrupt_out, interrupt_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_INTERRUPT)
        iso_out, iso_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_ISOCHRONOUS)

        if verbose:
            log('\tbulk_out      {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(bulk_out))
            log('\tbulk_in       {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(bulk_in))
            log('\tinterrupt_out {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(interrupt_out))
            log('\tinterrupt_in  {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(interrupt_in))
            log('\tiso_out       {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(iso_out))
            log('\tiso_in        {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(iso_in))

        if verbose:
            log('Testing simultaneous transfers through bulk and interrupt endpoint pairs.')
        test_error = Event()
        test_failure = Event()
        test_kwargs_bulk_ep = {
            'ep_out': bulk_out,
            'ep_in': bulk_in,
            'failure': test_failure,
            'error': test_error,
            'seconds': 1.0,
            'log': log}
        test_kwargs_interrupt_ep = {
            'ep_out': interrupt_out,
            'ep_in': interrupt_in,
            'failure': test_failure,
            'error': test_error,
            'seconds': 1.0,
            'log': log}
        ep_test_threads = []
        for kwargs in (test_kwargs_bulk_ep, test_kwargs_interrupt_ep):
            ep_test_threads.append(Thread(target=random_size_loopback_ep_test, kwargs=kwargs))
        for t in ep_test_threads:
            t.start()
        for t in ep_test_threads:
            t.join()
        if test_failure.is_set():
            raise_unconditionally(lineno(), 'Payload mismatch')
        if test_error.is_set():
            raise_unconditionally(lineno(), 'USBError')


def ep_test_parallel_transfers_ctrl(dev, log, verbose=False):
    """Test simultaneous data transfers in parallel with control transfers.

    Given a USB device with multiple OUT/IN endpoint pairs
    When multiple OUT and IN endpoints are used to transfer random data
        and control requests are processed in parallel
    Then all transfers succeed
        and for every endpoint pair, data received by host equals data sent by host
    """
    cfg = dev.get_active_configuration()
    for intf in cfg:
        log('interface {}, alt {} -- '.format(intf.bInterfaceNumber, intf.bAlternateSetting), end='')
        if intf.bAlternateSetting == 0:
            log('skipping the default AlternateSetting')
            continue
        log('running tests')
        intf.set_altsetting()

        bulk_out, bulk_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_BULK)
        interrupt_out, interrupt_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_INTERRUPT)
        iso_out, iso_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_ISOCHRONOUS)

        if verbose:
            log('\tbulk_out      {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(bulk_out))
            log('\tbulk_in       {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(bulk_in))
            log('\tinterrupt_out {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(interrupt_out))
            log('\tinterrupt_in  {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(interrupt_in))
            log('\tiso_out       {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(iso_out))
            log('\tiso_in        {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(iso_in))

        if verbose:
            log('Testing parallel data transfers through bulk, interrupt & control endpoint pairs.')
        test_error = Event()
        test_failure = Event()
        test_kwargs_bulk_ep = {
            'ep_out': bulk_out,
            'ep_in': bulk_in,
            'failure': test_failure,
            'error': test_error,
            'seconds': 1.0,
            'log': log}
        test_kwargs_interrupt_ep = {
            'ep_out': interrupt_out,
            'ep_in': interrupt_in,
            'failure': test_failure,
            'error': test_error,
            'seconds': 1.0,
            'log': log}
        ep_test_threads = []
        for kwargs in (test_kwargs_bulk_ep, test_kwargs_interrupt_ep):
            ep_test_threads.append(Thread(target=random_size_loopback_ep_test, kwargs=kwargs))
        for t in ep_test_threads:
            t.start()
        while any(t.is_alive() for t in ep_test_threads):
            control_stress_test(dev, log)
            control_sizes_test(dev, log)
        for t in ep_test_threads:
            t.join()
        if test_failure.is_set():
            raise_unconditionally(lineno(), 'Payload mismatch')
        if test_error.is_set():
            raise_unconditionally(lineno(), 'USBError')


def ep_test_abort(dev, log, verbose=False):
    """Test aborting data transfer for every OUT/IN endpoint pair.

    Given a USB device with multiple OUT/IN endpoint pairs
    When a device aborts an in progress data transfer
    Then no more data is transmitted
        and endpoint buffer is correctly released on the device end
    """
    NUM_PACKETS_UNTIL_ABORT = 2
    NUM_PACKETS_AFTER_ABORT = 8

    # If the host ever receives a payload with any byte set to this value,
    # the device does not handle abort operation correctly. The buffer
    # passed to aborted operation must not be used after call to abort().
    FORBIDDEN_PAYLOAD_VALUE = NUM_PACKETS_AFTER_ABORT + 1

    cfg = dev.get_active_configuration()
    for intf in cfg:
        log('interface {}, alt {} -- '.format(intf.bInterfaceNumber, intf.bAlternateSetting), end='')
        if intf.bAlternateSetting == 0:
            log('skipping the default AlternateSetting')
            continue
        log('running tests')
        intf.set_altsetting()

        bulk_out, bulk_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_BULK)
        interrupt_out, interrupt_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_INTERRUPT)

        if verbose:
            log('\tbulk_out      {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(bulk_out))
            log('\tbulk_in       {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(bulk_in))
            log('\tinterrupt_out {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(interrupt_out))
            log('\tinterrupt_in  {0.bEndpointAddress:#04x}, {0.wMaxPacketSize:02} B'.format(interrupt_in))

        if verbose:
            log('Testing aborting an in progress transfer for IN endpoints.')
        for ep_in in (bulk_in, interrupt_in):
            payload_size = (NUM_PACKETS_UNTIL_ABORT + NUM_PACKETS_AFTER_ABORT) * ep_in.wMaxPacketSize
            payload_in = array.array('B')
            while len(payload_in) < payload_size:
                try:
                    packet = ep_in.read(ep_in.wMaxPacketSize)
                    payload_in.extend(packet)
                except usb.core.USBError as err:
                    break
            if FORBIDDEN_PAYLOAD_VALUE in payload_in:
                raise_unconditionally(
                    lineno(), 'Endpoint buffer not released when aborting the '
                    'write operation on endpoint {0.bEndpointAddress:#04x}.'
                    .format(ep_in))
            if verbose:
                log('The size of data successfully received from endpoint {0.bEndpointAddress:#04x}: {1} B.'
                    .format(ep_in, len(payload_in)))
            too_little = bool(len(payload_in) < (NUM_PACKETS_UNTIL_ABORT * ep_in.wMaxPacketSize))
            too_much = bool(len(payload_in) >= payload_size)
            if too_little or too_much:
                raise_unconditionally(
                    lineno(), 'Invalid size of data successfully received from endpoint '
                    '{0.bEndpointAddress:#04x} before aborting the transfer. '
                    'Value {1} B out of range [{2}, {3}).'
                    .format(ep_in, len(payload_in),
                            NUM_PACKETS_UNTIL_ABORT * ep_in.wMaxPacketSize, payload_size))

        if verbose:
            log('Testing aborting an in progress transfer for OUT endpoints.')
        for ep_out in (bulk_out, interrupt_out):
            payload_size = (NUM_PACKETS_UNTIL_ABORT + NUM_PACKETS_AFTER_ABORT) * ep_out.wMaxPacketSize
            num_bytes_written = 0
            while num_bytes_written < payload_size:
                payload_out = array.array('B', (num_bytes_written//ep_out.wMaxPacketSize
                                                for _ in range(ep_out.wMaxPacketSize)))
                try:
                    num_bytes_written += ep_out.write(payload_out)
                except usb.core.USBError:
                    break
            try:
                ep_buff_correct = request_abort_buff_check(dev, ep_out)
            except (usb.core.USBError, IndexError, TypeError) as err:
                raise_unconditionally(
                    lineno(), 'Unable to verify endpoint buffer content ({!r}).'.format(err))
            if not ep_buff_correct:
                raise_unconditionally(
                    lineno(), 'Endpoint buffer not released when aborting the '
                    'read operation on endpoint {0.bEndpointAddress:#04x}.'
                    .format(ep_out))
            if verbose:
                log('The size of data successfully sent to endpoint {0.bEndpointAddress:#04x}: {1} B.'
                    .format(ep_out, num_bytes_written))
            too_little = bool(num_bytes_written < (NUM_PACKETS_UNTIL_ABORT * ep_out.wMaxPacketSize))
            too_much = bool(num_bytes_written >= payload_size)
            if too_little or too_much:
                raise_unconditionally(
                    lineno(), 'Invalid size of data successfully sent to endpoint '
                    '{0.bEndpointAddress:#04x} before aborting the transfer. '
                    'Value {1} B out of range [{2}, {3}).'
                    .format(ep_out, num_bytes_written,
                            NUM_PACKETS_UNTIL_ABORT * ep_out.wMaxPacketSize, payload_size))


def ep_test_data_toggle(dev, log, verbose=False):
    """Test data toggle reset for bulk OUT/IN endpoint pairs.

    Given a USB device
    When an interface is set
    Then the data toggle bits for all endpoints are reset to DATA0
    When clear feature is called for an endpoint that *IS NOT* stalled
    Then the data toggle is reset to DATA0 for that endpoint
    When clear halt is called for an endpoint that *IS* stalled
    Then the data toggle is reset to DATA0 for that endpoint
    """
    cfg = dev.get_active_configuration()
    for intf in cfg:
        log('interface {}, alt {} -- '.format(intf.bInterfaceNumber, intf.bAlternateSetting), end='')
        if intf.bAlternateSetting == 0:
            log('skipping the default AlternateSetting')
            continue
        log('running tests')

        if verbose:
            log('Testing data toggle reset for bulk endpoint pair.')

        # 1.1 reset OUT and IN data toggle to DATA0
        intf.set_altsetting()
        bulk_out, bulk_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_BULK)

        # 1.2 send and receive a single data packet,
        # so both OUT and IN endpoints switch to DATA1
        loopback_ep_test(bulk_out, bulk_in, bulk_out.wMaxPacketSize)

        # 1.3 reset OUT and IN data toggle to DATA0
        # USB spec, section 9.1.1.5
        # "
        # Configuring a device or changing an alternate setting causes all of the status and
        # configuration values associated with endpoints in the affected interfaces to be set to their default values.
        # This includes setting the data toggle of any endpoint using data toggles to the value DATA0.
        # "
        intf.set_altsetting()
        bulk_out, bulk_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_BULK)

        # 1.4 verify that host and USB device are still in sync with respect to data toggle
        try:
            loopback_ep_test(bulk_out, bulk_in, bulk_out.wMaxPacketSize)
        except usb.USBError as err:
            if verbose:
                log(USB_ERROR_FMT.format(err, bulk_out, bulk_in, bulk_out.wMaxPacketSize))
            raise_unconditionally(lineno(), 'Data toggle not reset when setting interface.')

        # 2.1 reset OUT and IN data toggle to DATA0
        intf.set_altsetting()
        bulk_out, bulk_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_BULK)

        # 2.2 send and receive a single data packet,
        # so both OUT and IN endpoints switch to DATA1
        loopback_ep_test(bulk_out, bulk_in, bulk_out.wMaxPacketSize)

        # 2.3 reset OUT data toggle to DATA0
        # USB spec, section 9.4.5
        # "
        # For endpoints using data toggle, regardless of whether an endpoint has the Halt feature set, a
        # ClearFeature(ENDPOINT_HALT) request always results in the data toggle being reinitialized to DATA0.
        # "
        bulk_out.clear_halt()
        # The ClearFeature(ENDPOINT_HALT) terminates a pending read operation on the device end.
        # Use a custom vendor request to restart reading on the OUT endpoint.
        # This does not impact the state of the data toggle bit.
        request_endpoint_loops_restart(dev)

        # 2.4 verify that host and USB device are still in sync with respect to data toggle
        try:
            loopback_ep_test(bulk_out, bulk_in, bulk_out.wMaxPacketSize)
        except usb.USBError as err:
            if verbose:
                log(USB_ERROR_FMT.format(err, bulk_out, bulk_in, bulk_out.wMaxPacketSize))
            raise_unconditionally(lineno(), 'Data toggle not reset when calling ClearFeature(ENDPOINT_HALT) '
                                  'on an endpoint that has not been halted.')

        # 3.1 reset OUT and IN data toggle to DATA0
        intf.set_altsetting()
        bulk_out, bulk_in = find_ep_pair(intf, usb.ENDPOINT_TYPE_BULK)

        # 3.2 send and receive a single data packet,
        # so both OUT and IN endpoints switch to DATA1
        loopback_ep_test(bulk_out, bulk_in, bulk_out.wMaxPacketSize)

        # 3.3 reset IN data toggle to DATA0
        # USB spec, section 9.4.5
        # "
        # For endpoints using data toggle, regardless of whether an endpoint has the Halt feature set, a
        # ClearFeature(ENDPOINT_HALT) request always results in the data toggle being reinitialized to DATA0.
        # "
        usb.control.set_feature(dev, FEATURE_ENDPOINT_HALT, bulk_in)
        bulk_in.clear_halt()

        # 3.4 verify that host and USB device are still in sync with respect to data toggle
        try:
            loopback_ep_test(bulk_out, bulk_in, bulk_out.wMaxPacketSize)
        except usb.USBError as err:
            if verbose:
                log(USB_ERROR_FMT.format(err, bulk_out, bulk_in, bulk_out.wMaxPacketSize))
            raise_unconditionally(lineno(), 'Data toggle not reset when clearing endpoint halt.')


def device_reset_test(log):
    """
    Test USB implementation against repeated reset

    Given an initialized USB (HOST <---> DUT connection established)
    When USB device is reset repeatedly
    Then the USB is operational with no errors
    """

    dev = yield
    dev.reset();
    dev = yield
    # run other test to check if USB works fine after reset
    control_data_test(dev, [64, 256], log)
    dev.reset();
    dev = yield
    # run other test to check if USB works fine after reset
    control_data_test(dev, [64, 256], log)
    dev.reset();
    dev = yield
    # run other test to check if USB works fine after reset
    control_data_test(dev, [64, 256], log)
    yield


def device_soft_reconnection_test(log):
    """
    Test USB implementation against repeated reconnection

    Given an initialized USB (HOST <---> DUT connection established)
    When USB device is disconnected and then connected repeatedly
    Then the USB is operational with no errors
    """

    list = [64, 256]
    dev = yield
    # run other test to check if USB works fine before reconnection
    control_data_test(dev, list, log)
    dev = yield
    # run other test to check if USB works fine after reconnection
    control_data_test(dev, list, log)
    dev = yield
    # run other test to check if USB works fine after reconnection
    control_data_test(dev, list, log)
    dev = yield
    # run other test to check if USB works fine after reconnection
    control_data_test(dev, list, log)
    dev = yield
    # run other test to check if USB works fine after reconnection
    control_data_test(dev, list, log)
    yield


def device_suspend_resume_test(log):
    """
    Test USB implementation against repeated suspend and resume

    Given an initialized USB (HOST <---> DUT connection established)
    When USB device is suspended and then resumed repeatedly
    Then the USB is operational with no errors
    """

    dev = yield
    control_data_test(dev, [64, 256], log)
    # suspend code goes here
    # ...
    # resume code here
    # ...
    # run other test to check if USB works fine after resume
    control_data_test(dev, [64, 256], log)
    # suspend code here
    # ...
    # resume code here
    # ...
    # run other test to check if USB works fine after resume
    control_data_test(dev, [64, 256], log)
    # suspend code here
    # ...
    # resume code here
    # ...
    # run other test to check if USB works fine after resume
    control_data_test(dev, [64, 256], log)
    yield


def repeated_construction_destruction_test(log):
    """
    Test USB implementation against repeated initialization and deinitialization

    Given an initialized USB (HOST <---> DUT connection established)
    When USB device is deinitialized and then initialized repeatedly
    Then the USB is operational with no errors
    """

    list = [64, 256]
    dev = yield
    # run other test to check if USB works fine after repeated construction/destruction
    control_data_test(dev, list, log)
    dev = yield
    # run other test to check if USB works fine after repeated construction/destruction
    control_data_test(dev, list, log)
    dev = yield
    # run other test to check if USB works fine after repeated construction/destruction
    control_data_test(dev, list, log)
    yield


def release_interfaces(dev):
    """ Releases interfaces to allow configuration switch

        Fixes error while configuration change(on Windows machines):
        USBError: [Errno None] libusb0-dll:err [set_configuration] can't change configuration, an interface is still in use (claimed)
    """
    cfg = dev.get_active_configuration()
    for i in range(0, cfg.bNumInterfaces):
        usb.util.release_interface(dev, i)


def restore_default_configuration(dev):
    """ Set default configuration """

    cfg = dev[1]
    cfg.set()


def get_status(dev, recipient, index = 0):
    """ Get status of the recipient

    Args:
        dev - pyusb device
        recipient - CTRL_RECIPIENT_DEVICE/CTRL_RECIPIENT_INTERFACE/CTRL_RECIPIENT_ENDPOINT
        index - 0 if recipient is device, interface index if recipient is interface, endpoint index if recipient is endpoint

    Returns:
        status flag 32b int
    """
    request_type = build_request_type(CTRL_IN, CTRL_TYPE_STANDARD,
                                      recipient)
    request = REQUEST_GET_STATUS
    value = 0                                   # Always 0 for this request
    index = index                               # recipient index
    length = 2                                  # Always 2 for this request (size of return data)
    ret = dev.ctrl_transfer(request_type, request, value, index, length)
    ret = ret[0] | (ret[1] << 8)

    return ret


def get_descriptor(dev, type_index, lang_id, length):
    # Control IN GET_DESCRIPTOR - device
    request_type = build_request_type(CTRL_IN, CTRL_TYPE_STANDARD,
                                      CTRL_RECIPIENT_DEVICE)
    request = REQUEST_GET_DESCRIPTOR
    value = type_index                          # Descriptor Type (H) and Descriptor Index (L)
    index = lang_id                             # 0 or Language ID for this request
    length = length                             # Descriptor Length
    ret = dev.ctrl_transfer(request_type, request, value, index, length)

    return ret
