# Lint as: python2, python3
# Copyright 2019 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.

"""
This class provides wrapper functions for Bluetooth quick health test
batches or packages
"""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import functools
import logging
import tempfile
import threading
import time

import common
from autotest_lib.client.common_lib import error
from autotest_lib.server import site_utils
from autotest_lib.server.cros.bluetooth import bluetooth_adapter_tests
from autotest_lib.server.cros.multimedia import remote_facade_factory
from autotest_lib.client.bin import utils
from six.moves import range

PROFILE_CONNECT_WAIT = 15
SUSPEND_SEC = 15
EXPECT_NO_WAKE_SUSPEND_SEC = 30
EXPECT_PEER_WAKE_SUSPEND_SEC = 60
EXPECT_PEER_WAKE_RESUME_BY = 30


class BluetoothAdapterQuickTests(bluetooth_adapter_tests.BluetoothAdapterTests):
    """This class provide wrapper function for Bluetooth quick health test
    batches or packages.
    The Bluetooth quick test infrastructure provides a way to quickly run a set
    of tests. As for today, auto-test ramp up time per test is about 90-120
    seconds, where a typical Bluetooth test may take ~30-60 seconds to run.

    The quick test infra, implemented in this class, saves this huge overhead
    by running only the minimal reset and cleanup operations required between
    each set of tests (takes a few seconds).

    This class provides wrapper functions to start and end a test, a batch or a
    package. A batch is defined as a set of tests, preferably with a common
    subject. A package is a set of batches.
    This class takes care of tests, batches, and packages test results, and
    prints out summaries to results. The class also resets and cleans up
    required states between tests, batches and packages.

    A batch can also run as a separate auto-test. There is a place holder to
    add a way to run a specific test of a batch autonomously.

    A batch can be implemented by inheriting from this class, and using its
    wrapper functions. A package can be implemented by inheriting from a set of
    batches.

    Adding a test to one of the batches is as easy as adding a method to the
    class of the batch.
    """

    # Some delay is needed between tests. TODO(yshavit): investigate and remove
    TEST_SLEEP_SECS = 3
    GCS_MTBF_BUCKET = 'gs://chromeos-mtbf-bt-results/'


    def restart_peers(self):
        """Restart and clear peer devices"""
        # Restart the link to device
        logging.info('Restarting peer devices...')

        # Grab current device list for initialization
        connected_devices = self.devices
        self.cleanup_bt_test(test_state='MID')

        for device_type, device_list in connected_devices.items():
            for device in device_list:
                if device is not None:
                    logging.info('Restarting %s', device_type)
                    self.get_device(device_type, on_start=False)


    def start_peers(self, devices):
        """Start peer devices"""
        # Start the link to devices
        if self.use_btpeer:
            logging.info('Starting peer devices...')
            self.get_device_rasp(devices)

            # Grab all the devices to verify RSSI
            devices = []
            for device_type, device_list in self.devices.items():
                # Skip bluetooth_tester since it won't be discoverable
                if 'TESTER' in device_type:
                    continue

                for device in device_list:
                    devices.append(device)
                    self.start_agent(device)

            if self.rssi_check:
                # Make sure device RSSI is sufficient
                self.verify_device_rssi(devices)
            else:
                logging.info('Skip RSSI check.')

    def _print_delimiter(self):
        logging.info('=======================================================')

    @staticmethod
    def _get_bool_arg(arg, args_dict, default_value):
        """Get the target bool argument from args_dict.

        @param arg: the target argument to query
        @param args_dict: the argument dictionary
        @param default_value: the default value of the argument
                if the arg is not in args_dict or
                if arg is neither 'true' nor 'false'

        @returns: the bool value of the target argument
        """
        if args_dict is not None and arg in args_dict:
            arg_value = args_dict[arg].lower()
            if arg_value == 'true':
                return True
            elif arg_value == 'false':
                return False
        return default_value

    @staticmethod
    def _get_clean_kernel_log_arguments(args_dict=None):
        """Parse the clean_kernel_log argument"""
        key = 'clean_kernel_log'
        if args_dict is not None and key in args_dict:
            return args_dict[key].upper()
        return 'DEBUG'

    def quick_test_init(self,
                        host,
                        use_btpeer=True,
                        flag='Quick Health',
                        args_dict=None,
                        start_browser=False,
                        floss=False):
        """Inits the test batch"""
        self.host = host
        self.start_browser = start_browser
        self.use_btpeer = use_btpeer
        self.floss = floss

        logging.debug('args_dict %s', args_dict)
        update_btpeers = self._get_bool_arg('update_btpeers', args_dict, True)
        self.rssi_check = self._get_bool_arg('rssi_check', args_dict, True)
        clean_log = self._get_clean_kernel_log_arguments(args_dict)
        btpeer_args = []
        if args_dict is not None:
            btpeer_args = self.host.get_btpeer_arguments(args_dict)
        #factory can not be declared as local variable, otherwise
        #factory._proxy.__del__ will be invoked, which shutdown the xmlrpc
        # server, which log out the user.

        self.factory = remote_facade_factory.RemoteFacadeFactory(
                host, no_chrome=not self.start_browser, disable_arc=True)
        try:
            self.bluetooth_facade = self.factory.create_bluetooth_facade(
                    self.floss)
        except Exception as e:
            logging.error('Exception %s while creating bluetooth_facade',
                          str(e))
            raise error.TestFail('Unable to create bluetooth_facade')

        if clean_log is not 'FALSE':
            # Clean Bluetooth kernel logs in the DUT to prevent
            # /var/log/messages occupying too much space
            self.clean_bluetooth_kernel_log(clean_log)

        if self.use_btpeer:
            self.input_facade = self.factory.create_input_facade()

            self.host.initialize_btpeer(btpeer_args=btpeer_args)
            logging.info('%s Bluetooth peers found',
                         len(self.host.btpeer_list))
            logging.info('labels: %s', self.host.get_labels())

            if len(self.host.btpeer_list) == 0:
                raise error.TestFail('Unable to find a Bluetooth peer')

            # Check the chameleond version on the peer and update if necessary
            if update_btpeers:
                if not self.update_btpeer():
                    logging.error('Updating btpeers failed. Ignored')
            else:
                logging.info('No attempting peer update.')

            # Query connected devices on our btpeer at init time
            self.available_devices = self.list_devices_available()

            for btpeer in self.host.btpeer_list:
                btpeer.register_raspPi_log(self.outputdir)

            self.btpeer_group = dict()
            # Create copy of btpeer_group
            self.btpeer_group_copy = dict()
            self.group_btpeers_type()


        # Clear the active devices for this test
        self.active_test_devices = {}

        self.enable_disable_debug_log(enable=True)

        # Disable cellular services, as they can sometimes interfere with
        # suspend/resume, i.e. b/161920740
        self.enable_disable_cellular(enable=False)

        self.enable_disable_ui(enable=False)

        # Delete files created in previous run
        self.host.run('[ ! -d {0} ] || rm -rf {0} || true'.format(
                                                    self.BTMON_DIR_LOG_PATH))
        self.host.run('[ ! -d {0} ] || rm -rf {0} || true'.format(
                                                    self.USBMON_DIR_LOG_PATH))

        self.start_new_btmon()
        self.start_new_usbmon()

        self.flag = flag
        self.test_iter = None

        self.bat_tests_results = []
        self.bat_pass_count = 0
        self.bat_fail_count = 0
        self.bat_testna_count = 0
        self.bat_warn_count = 0
        self.bat_name = None
        self.bat_iter = None

        self.pkg_tests_results = []
        self.pkg_pass_count = 0
        self.pkg_fail_count = 0
        self.pkg_testna_count = 0
        self.pkg_warn_count = 0
        self.pkg_name = None
        self.pkg_iter = None
        self.pkg_is_running = False
        self.mtbf_end = False
        self.mtbf_end_lock = threading.Lock()


    @staticmethod
    def quick_test_test_decorator(test_name,
                                  devices={},
                                  flags=['All'],
                                  model_testNA=[],
                                  model_testWarn=[],
                                  skip_models=[],
                                  skip_chipsets=[],
                                  skip_common_errors=False,
                                  shared_devices_count=0):
        """A decorator providing a wrapper to a quick test.
           Using the decorator a test method can implement only the core
           test and let the decorator handle the quick test wrapper methods
           (test_start and test_end).

           @param test_name: the name of the test to log.
           @param devices:   list of device names which are going to be used
                             in the following test.
           @param flags: list of string to describe who should run the
                         test. The string could be one of the following:
                         ['AVL', 'Quick Health', 'All'].
           @param model_testNA: If the current platform is in this list,
                                failures are emitted as TestNAError.
           @param model_testWarn: If the current platform is in this list,
                                  failures are emitted as TestWarn.
           @param skip_models: Raises TestNA on these models and doesn't attempt
                               to run the tests.
           @param skip_chipsets: Raises TestNA on these chipset and doesn't
                                 attempt to run the tests.
           @param skip_common_errors: If the test encounters a common error
                                      (such as USB disconnect or daemon crash),
                                      mark the test as TESTNA instead.
                                      USE THIS SPARINGLY, it may mask bugs. This
                                      is available for tests that require state
                                      to be properly retained throughout the
                                      whole test (i.e. advertising) and any
                                      outside failure will cause the test to
                                      fail.
        """

        def decorator(test_method):
            """A decorator wrapper of the decorated test_method.
               @param test_method: the test method being decorated.
               @returns the wrapper of the test method.
            """

            def _check_runnable(self):
                """Check if the test could be run"""

                # Check that the test is runnable in current setting
                if not(self.flag in flags or 'All' in flags):
                    logging.info('SKIPPING TEST %s', test_name)
                    logging.info('flag %s not in %s', self.flag, flags)
                    self._print_delimiter()
                    return False
                return True


            def _is_enough_peers_present(self):
                """Check if enough peer devices are available."""

                # Check that btpeer has all required devices before running
                for device_type, number in devices.items():
                    if self.available_devices.get(device_type, 0) < number:
                        logging.info('SKIPPING TEST %s', test_name)
                        logging.info('%s not available', device_type)
                        self._print_delimiter()
                        return False

                # Check if there are enough peers
                total_num_devices = sum(devices.values()) + shared_devices_count
                if total_num_devices > len(self.host.btpeer_list):
                    logging.info('SKIPPING TEST %s', test_name)
                    logging.info(
                            'Number of devices required %s is greater'
                            'than number of peers available %d',
                            total_num_devices, len(self.host.btpeer_list))
                    self._print_delimiter()
                    return False
                return True

            @functools.wraps(test_method)
            def wrapper(self):
                """A wrapper of the decorated method."""
                # Set test name before exiting so batches correctly identify
                # failing tests
                self.test_name = test_name

                # Every test_method should pass by default.
                # This should be placed in the beginning of the wrapper before
                # executing any self.test_xxx tests. This statement should not
                # be placed inside quick_test_test_start(). If there are any
                # exceptions raised in the try block below, the
                # quick_test_test_end() which contains self.test_xxx tests
                # would be executed without the _expected_result attribute
                # getting populated.
                self._expected_result = True

                if not _check_runnable(self):
                    return

                try:
                    if not _is_enough_peers_present(self):
                        logging.info('Not enough peer available')
                        raise error.TestNAError('Not enough peer available')

                    model = self.get_base_platform_name()
                    if model in skip_models:
                        logging.info('SKIPPING TEST %s', test_name)
                        raise error.TestNAError(
                                'Test not supported on this model')

                    chipset = self.get_chipset_name()
                    logging.debug('Bluetooth module name is %s', chipset)
                    if chipset in skip_chipsets:
                        logging.info('SKIPPING TEST %s on chipset %s',
                                     test_name, chipset)
                        raise error.TestNAError(
                                'Test not supported on this chipset')

                    self.quick_test_test_start(test_name, devices,
                                               shared_devices_count)

                    test_method(self)
                except error.TestError as e:
                    if not bool(self.fails):
                        self.fails.append('[--- error {} ({})]'.format(
                                test_method.__name__, str(e)))
                except error.TestFail as e:
                    if not bool(self.fails):
                        self.fails.append('[--- failed {} ({})]'.format(
                                test_method.__name__, str(e)))
                except error.TestNAError as e:
                    if not bool(self.fails):
                        self.fails.append('[--- SKIPPED {} ({})]'.format(
                                test_method.__name__, str(e)))
                except Exception as e:
                    self.fails.append('[--- unknown error {} ({})]'.format(
                            test_method.__name__, str(e)))

                self.quick_test_test_end(model_testNA=model_testNA,
                                         model_testWarn=model_testWarn,
                                         skip_common_errors=skip_common_errors)

            return wrapper

        return decorator


    def quick_test_test_start(
            self, test_name=None, devices={}, shared_devices_count=0):
        """Start a quick test. The method clears and restarts adapter on DUT
           as well as peer devices. In addition the methods prints test start
           traces.
        """
        # Bluetoothd could have crashed behind the scenes; check to see if
        # everything is still ok and recover if needed.
        self.test_is_facade_valid()
        self.test_is_adapter_valid()

        # Reset the adapter
        self.test_reset_on_adapter()

        # Reset the policy allowlist so that all UUIDs are allowed.
        self.test_reset_allowlist()

        # Initialize bluetooth_adapter_tests class (also clears self.fails)
        self.initialize()
        # Start and peer HID devices
        self.start_peers(devices)
        self.shared_peers = self.host.btpeer_list[-shared_devices_count:]

        if test_name is not None:
            time.sleep(self.TEST_SLEEP_SECS)
            self._print_delimiter()
            logging.info('Starting test: %s', test_name)
            self.log_message('Starting test: %s'% test_name)

    def quick_test_test_end(self,
                            model_testNA=[],
                            model_testWarn=[],
                            skip_common_errors=False):
        """Log and track the test results"""
        result_msgs = []
        model = self.get_base_platform_name()

        if self.test_iter is not None:
            result_msgs += ['Test Iter: ' + str(self.test_iter)]

        if self.bat_iter is not None:
            result_msgs += ['Batch Iter: ' + str(self.bat_iter)]

        if self.pkg_is_running is True:
            result_msgs += ['Package iter: ' + str(self.pkg_iter)]

        if self.bat_name is not None:
            result_msgs += ['Batch Name: ' + self.bat_name]

        if self.test_name is not None:
            result_msgs += ['Test Name: ' + self.test_name]

        result_msg = ", ".join(result_msgs)

        if not bool(self.fails):
            result_msg = 'PASSED | ' + result_msg
            self.bat_pass_count += 1
            self.pkg_pass_count += 1
        # The test should be marked as TESTNA if any of the test expressions
        # were SKIPPED (they threw their own TESTNA error) or the model is in
        # the list of NA models (so any failure is considered NA instead)
        elif model in model_testNA or any(['SKIPPED' in x
                                           for x in self.fails]):
            result_msg = 'TESTNA | ' + result_msg
            self.bat_testna_count += 1
            self.pkg_testna_count += 1
        elif model in model_testWarn:
            result_msg = 'WARN   | ' + result_msg
            self.bat_warn_count += 1
            self.pkg_warn_count += 1
        # Some tests may fail due to known common failure reasons (like usb
        # disconnect during suspend, bluetoothd crashes, etc). Skip those tests
        # with TESTNA when that happens.
        #
        # This should be used sparingly because it may hide legitimate errors.
        elif bool(self.had_known_common_failure) and skip_common_errors:
            result_msg = 'TESTNA | ' + result_msg
            self.bat_testna_count += 1
            self.pkg_testna_count += 1
        else:
            result_msg = 'FAIL   | ' + result_msg
            self.bat_fail_count += 1
            self.pkg_fail_count += 1

        logging.info(result_msg)
        self.log_message(result_msg)
        self._print_delimiter()
        self.bat_tests_results.append(result_msg)
        self.pkg_tests_results.append(result_msg)

        if self.test_name is not None:
            logging.info('Cleanning up and restarting towards next test...')


        # Bluetoothd could have crashed behind the scenes; check if everything
        # is ok and recover if needed. This is done as part of clean-up as well
        # to make sure we can fully remove pairing info between tests
        self.test_is_facade_valid()

        self.bluetooth_facade.stop_discovery()

        # Catch possible exceptions in test_reset_allowlist().
        # Refer to b/184947150 for more context.
        try:
            # Reset the policy allowlist so that all UUIDs are allowed.
            self.test_reset_allowlist()
        except:
            msg = ('Failed to reset the policy allowlist.\n'
                   '### Note: reset the allowlist manually if needed. ###\n\n'
                   'dbus-send --system --print-reply --dest=org.bluez '
                   '/org/bluez/hci0 org.bluez.AdminPolicy1.SetServiceAllowList '
                   'array:string:"" \n')
            logging.error(msg)

        # Store a copy of active devices for raspi reset in the final step
        self.active_test_devices = self.devices

        # Disconnect devices used in the test, and remove the pairing.
        for device_list in self.devices.values():
            for device in device_list:
                if device is not None:
                    self.stop_agent(device)
                    logging.info('Clear device %s from DUT', device.name)
                    self.bluetooth_facade.disconnect_device(device.address)
                    device_is_paired = self.bluetooth_facade.device_is_paired(
                            device.address)
                    if device_is_paired:
                        self.bluetooth_facade.remove_device_object(
                                device.address)

                    # Also remove pairing on Peer
                    logging.info('Clearing DUT from %s', device.name)
                    try:
                        device.RemoveDevice(self.bluetooth_facade.address)
                    except Exception as e:
                        # If peer doesn't expose RemoveDevice, ignore failure
                        if not (e.__class__.__name__ == 'Fault' and
                                'is not supported' in str(e)):
                            logging.info('Couldn\'t Remove: {}'.format(e))
                            raise


        # Repopulate btpeer_group for next tests
        # Clear previous tets's leftover entries. Don't delete the
        # btpeer_group dictionary though, it'll be used as it is.
        if self.use_btpeer:
            for device_type in self.btpeer_group:
                if len(self.btpeer_group[device_type]) > 0:
                    del self.btpeer_group[device_type][:]

            # Repopulate
            self.group_btpeers_type()

        # Close the connection between peers
        self.cleanup_bt_test(test_state='NEW')

    @staticmethod
    def quick_test_batch_decorator(batch_name):
        """A decorator providing a wrapper to a batch.
           Using the decorator a test batch method can implement only its
           core tests invocations and let the decorator handle the wrapper,
           which is taking care for whether to run a specific test or the
           batch as a whole and and running the batch in iterations

           @param batch_name: the name of the batch to log
        """

        def decorator(batch_method):
            """A decorator wrapper of the decorated test_method.
               @param test_method: the test method being decorated.
               @returns the wrapper of the test method.
            """

            @functools.wraps(batch_method)
            def wrapper(self, num_iterations=1, test_name=None):
                """A wrapper of the decorated method.
                  @param num_iterations: how many interations to run
                  @param test_name: specifc test to run otherwise None to run
                                    the whole batch
                """
                self.identify_platform_failure_reasons()
                if test_name is not None:
                    single_test_method = getattr(self,  test_name)
                    for iter in range(1,num_iterations+1):
                        self.test_iter = iter
                        single_test_method()

                    if self.fails:
                        # If failure is marked as TESTNA, prioritize that over
                        # a failure. Same with WARN.
                        if self.bat_testna_count > 0:
                            raise error.TestNAError(self.fails)
                        elif self.bat_warn_count > 0:
                            raise error.TestWarn(self.fails)
                        else:
                            raise error.TestFail(self.fails)
                else:
                    for iter in range(1,num_iterations+1):
                        self.quick_test_batch_start(batch_name, iter)
                        batch_method(self, num_iterations, test_name)
                        self.quick_test_batch_end()
            return wrapper

        return decorator


    def quick_test_batch_start(self, bat_name, iteration=1):
        """Start a test batch. The method clears and set batch variables"""
        self.bat_tests_results = []
        self.bat_pass_count = 0
        self.bat_fail_count = 0
        self.bat_testna_count = 0
        self.bat_warn_count = 0
        self.bat_name = bat_name
        self.bat_iter = iteration


    def quick_test_batch_end(self):
        """Print results summary of a test batch """
        logging.info(
                '%s Test Batch Summary: total pass %d, total fail %d, '
                'warn %d, NA %d', self.bat_name, self.bat_pass_count,
                self.bat_fail_count, self.bat_warn_count,
                self.bat_testna_count)
        for result in self.bat_tests_results:
            logging.info(result)
        self._print_delimiter();
        if self.bat_fail_count > 0:
            logging.error('===> Test Batch Failed! More than one failure')
            self._print_delimiter();
            if self.pkg_is_running is False:
                raise error.TestFail(self.bat_tests_results)
        elif self.bat_testna_count > 0:
            logging.error('===> Test Batch Passed! Some TestNA results')
            self._print_delimiter();
            if self.pkg_is_running is False:
                raise error.TestNAError(self.bat_tests_results)
        elif self.bat_warn_count > 0:
            logging.error('===> Test Batch Passed! Some WARN results')
            self._print_delimiter();
            if self.pkg_is_running is False:
                raise error.TestWarn(self.bat_tests_results)
        else:
            logging.info('===> Test Batch Passed! zero failures')
            self._print_delimiter();


    def quick_test_package_start(self, pkg_name):
        """Start a test package. The method clears and set batch variables"""
        self.pkg_tests_results = []
        self.pkg_pass_count = 0
        self.pkg_fail_count = 0
        self.pkg_name = pkg_name
        self.pkg_is_running = True


    def quick_test_print_summary(self):
        """Print results summary of a test package"""
        logging.info(
                '%s Test Package Summary: total pass %d, total fail %d, '
                'Warn %d, NA %d', self.pkg_name, self.pkg_pass_count,
                self.pkg_fail_count, self.pkg_warn_count,
                self.pkg_testna_count)
        for result in self.pkg_tests_results:
            logging.info(result)
        self._print_delimiter();


    def quick_test_package_update_iteration(self, iteration):
        """Update state and print log per package iteration.
           Must be called to have a proper package test result tracking.
        """
        self.pkg_iter = iteration
        if self.pkg_name is None:
            logging.error('Error: no quick package is running')
            raise error.TestFail('Error: no quick package is running')
        logging.info('Starting %s Test Package iteration %d',
                     self.pkg_name, iteration)


    def quick_test_package_end(self):
        """Print final result of a test package"""
        if self.pkg_fail_count > 0:
            logging.error('===> Test Package Failed! More than one failure')
            self._print_delimiter();
            raise error.TestFail(self.bat_tests_results)
        elif self.pkg_testna_count > 0:
            logging.error('===> Test Package Passed! Some TestNA results')
            self._print_delimiter();
            raise error.TestNAError(self.bat_tests_results)
        elif self.pkg_warn_count > 0:
            logging.error('===> Test Package Passed! Some WARN results')
            self._print_delimiter();
            raise error.TestWarn(self.bat_tests_results)
        else:
            logging.info('===> Test Package Passed! zero failures')
            self._print_delimiter();
        self.pkg_is_running = False


    def quick_test_cleanup(self):
        """ Cleanup any state test server and all device"""

        # Clear any raspi devices at very end of test
        for device_list in self.active_test_devices.values():
            for device in device_list:
                if device is not None:
                    self.clear_raspi_device(device)
                    self.device_set_powered(device, False)

        # Reset the adapter
        self.test_reset_on_adapter()
        # Initialize bluetooth_adapter_tests class (also clears self.fails)
        self.initialize()


    @staticmethod
    def quick_test_mtbf_decorator(timeout_mins, test_name):
        """A decorator enabling a test to be run as a MTBF test, it will run
           the underlying test in a infinite loop until it fails or timeout is
           reached, in both cases the time elapsed time will be reported.

           @param timeout_mins: the max execution time of the test, once the
                                time is up the test will report success and exit
           @param test_name: the MTBF test name to be output to the dashboard
        """

        def decorator(batch_method):
            """A decorator wrapper of the decorated batch_method.
               @param batch_method: the batch method being decorated.
               @returns the wrapper of the batch method.
            """

            @functools.wraps(batch_method)
            def wrapper(self, *args, **kwargs):
                """A wrapper of the decorated method"""
                self.mtbf_end = False
                mtbf_timer = threading.Timer(
                    timeout_mins * 60, self.mtbf_timeout)
                mtbf_timer.start()
                start_time = time.time()
                board = self.host.get_board().split(':')[1]
                model = self.host.get_model_from_cros_config()
                build = self.host.get_release_version()
                milestone = 'M' + self.host.get_chromeos_release_milestone()
                in_lab = site_utils.host_in_lab(self.host.hostname)
                while True:
                    with self.mtbf_end_lock:
                        # The test ran the full duration without failure
                        if self.mtbf_end:
                            self.report_mtbf_result(
                                True, start_time, test_name, model, build,
                                milestone, board, in_lab)
                            break
                    try:
                        batch_method(self, *args, **kwargs)
                    except Exception as e:
                        logging.info("Caught a failure: %r", e)
                        self.report_mtbf_result(
                            False, start_time, test_name, model, build,
                            milestone, board, in_lab)
                        # Don't report the test run as failed for MTBF
                        self.fails = []
                        break

                mtbf_timer.cancel()

            return wrapper

        return decorator


    def mtbf_timeout(self):
        """Handle time out event of a MTBF test"""
        with self.mtbf_end_lock:
            self.mtbf_end = True


    def report_mtbf_result(self, success, start_time, test_name, model, build,
        milestone, board, in_lab):
        """Report MTBF result by uploading it to GCS"""
        duration_secs = int(time.time() - start_time)
        start_time = int(start_time)
        gm_time_struct = time.localtime(start_time)
        output_file_name = self.GCS_MTBF_BUCKET + \
                           time.strftime('%Y-%m-%d/', gm_time_struct) + \
                           time.strftime('%H-%M-%S.csv', gm_time_struct)

        mtbf_result = '{0},{1},{2},{3},{4},{5},{6},{7}'.format(
            model, build, milestone, start_time * 1000000, duration_secs,
            success, test_name, board)
        with tempfile.NamedTemporaryFile() as tmp_file:
            tmp_file.write(mtbf_result.encode('utf-8'))
            tmp_file.flush()
            cmd = 'gsutil cp {0} {1}'.format(tmp_file.name, output_file_name)
            logging.info('Result to upload %s %s', mtbf_result, cmd)
            # Only upload the result when running in the lab.
            if in_lab:
                logging.info('Uploading result')
                utils.run(cmd)


    # ---------------------------------------------------------------
    # Wake from suspend tests
    # ---------------------------------------------------------------

    def run_peer_wakeup_device(self,
                               device_type,
                               device,
                               device_test=None,
                               iterations=1,
                               should_wake=True,
                               should_pair=True):
        """ Uses paired peer device to wake the device from suspend.

        @param device_type: the device type (used to determine if it's LE)
        @param device: the meta device with the paired device
        @param device_test: What to test to run after waking and connecting the
                            adapter/host
        @param iterations: Number of suspend + peer wake loops to run
        @param should_wake: Whether wakeup should occur on this test. With HID
                            peers, this should be True. With non-HID peers, this
                            should be false.
        @param should_pair: Pair and connect the device first before running
                            the wakeup test.
        """
        boot_id = self.host.get_boot_id()

        if should_wake:
            sleep_time = EXPECT_PEER_WAKE_SUSPEND_SEC
            resume_time = EXPECT_PEER_WAKE_RESUME_BY
            resume_slack = 5  # Allow 5s slack for resume timeout
            measure_resume = True
        else:
            sleep_time = EXPECT_NO_WAKE_SUSPEND_SEC
            resume_time = EXPECT_NO_WAKE_SUSPEND_SEC
            # Negative resume slack lets us wake a bit earlier than expected
            # If suspend takes a while to enter, this may be necessary to get
            # the timings right.
            resume_slack = -5
            measure_resume = False

        try:
            if should_pair:
                # Clear wake before testing
                self.test_adapter_set_wake_disabled()
                self.assert_discover_and_pair(device)
                self.assert_on_fail(
                        self.test_device_set_discoverable(device, False))

                # Confirm connection completed
                self.assert_on_fail(
                        self.test_device_is_connected(device.address))

            # Profile connection may not have completed yet and this will
            # race with a subsequent disconnection (due to suspend). Use the
            # device test to force profile connect or wait if no test was
            # given.
            if device_test is not None:
                self.assert_on_fail(device_test(device))
            else:
                time.sleep(PROFILE_CONNECT_WAIT)

            for it in range(iterations):
                logging.info(
                        'Running iteration {}/{} of suspend peer wake'.format(
                                it + 1, iterations))

                # Start a new suspend instance
                suspend = self.suspend_async(suspend_time=sleep_time,
                                             expect_bt_wake=should_wake)
                start_time = self.bluetooth_facade.get_device_utc_time()

                if should_wake:
                    self.test_device_wake_allowed(device.address)
                    # Also wait until powerd marks adapter as wake enabled
                    self.test_adapter_wake_enabled()
                else:
                    self.test_device_wake_not_allowed(device.address)

                # Trigger suspend, asynchronously wake and wait for resume
                self.test_suspend_and_wait_for_sleep(suspend,
                                                     sleep_timeout=SUSPEND_SEC)

                # Trigger peer wakeup
                adapter_address = self.bluetooth_facade.address
                peer_wake = self.device_connect_async(device_type,
                                                      device,
                                                      adapter_address,
                                                      delay_wake=5,
                                                      should_wake=should_wake)
                peer_wake.start()

                # Expect a quick resume. If a timeout occurs, test fails. Since
                # we delay sending the wake signal, we should accommodate that
                # in our expected timeout.
                self.test_wait_for_resume(boot_id,
                                          suspend,
                                          resume_timeout=resume_time,
                                          test_start_time=start_time,
                                          resume_slack=resume_slack,
                                          fail_on_timeout=should_wake,
                                          fail_early_wake=not should_wake,
                                          collect_resume_time=measure_resume)

                # Finish peer wake process
                peer_wake.join()

                # Only check peer device connection state if we expected to wake
                # from it. Otherwise, we may or may not be connected based on
                # the specific profile's reconnection policy.
                if should_wake:
                    # Make sure we're actually connected
                    self.test_device_is_connected(device.address)

                    # Verify the profile is working
                    if device_test is not None:
                        device_test(device)

        finally:
            if should_pair:
                self.test_remove_pairing(device.address)