# Lint as: python2, python3
# Copyright 2019 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Server side bluetooth adapter stress tests involving power consumption."""

import logging
import multiprocessing
import time

from autotest_lib.client.common_lib import error
from autotest_lib.server.cros.bluetooth.bluetooth_adapter_tests import (
        test_case_log)
from autotest_lib.server.cros.bluetooth.bluetooth_adapter_quick_tests import (
        BluetoothAdapterQuickTests)


class bluetooth_AdapterPowerMeasure(BluetoothAdapterQuickTests):
    """Server side bluetooth adapter power consumption test."""

    test_wrapper = BluetoothAdapterQuickTests.quick_test_test_decorator
    batch_wrapper = BluetoothAdapterQuickTests.quick_test_batch_decorator


    def _check_legitimate_board(self):
        """Check if this is a legitimate board to run the test.

        Only a limited set of boards are supported for now, primarily
        the kukui family of barods.

        @raises: TestNAError if the board is not legitimate.

        """
        board = self.host.get_board().split(':')[1]
        if board not in ('kukui'):
            raise error.TestNAError('%s not legitimate to run the test.' %
                                    board)


    def _initialize_servod(self, device):
        """Perform initialize for servod task.

        @param device: the peer device

        @raises: TestError if not able to enable or start servo.
        """
        self.count_fail_to_sleep = 0
        self.count_fail_to_resume = 0
        self.count_system_resume_prematurely = 0
        self.count_success = 0

        # When the autotest restarts ui, chrome would issue some Bluetooth
        # commands which may prevent the system from suspending properly.
        # Hence, let's stop ui for now.
        self.host.run_short('stop ui')

        board = self.host.get_board().split(':')[1]
        logging.info('board: %s', board)

        # TODO(b/152737849): figure out a way to support more boards.
        self._check_legitimate_board()

        # device is a pure XMLRPC server running as chameleond
        # on the bluetooth peer. We need to enable Servod.
        if not device.EnableServod(board):
            raise error.TestError('Failed to enable Servod.')

        # Start the Servod process on the bluetooth peer.
        if not device.servod.Start():
            raise error.TestError('Failed to start Servod on bluetooth peer.')


    def _cleanup_servod(self, device):
        """Perform cleanup for servod.

        @param device: the peer device
        """
        if not device.servod.Stop():
            logging.error('Failed to stop Servod on bluetooth peer.')

        self.host.run_short('start ui')

        logging.info('count_fail_to_sleep: %d', self.count_fail_to_sleep)
        logging.info('count_fail_to_resume: %d', self.count_fail_to_resume)
        logging.info('count_system_resume_prematurely: %d',
                     self.count_system_resume_prematurely)
        logging.info('count_success: %d', self.count_success)


    # ---------------------------------------------------------------
    # Definitions of test cases
    # ---------------------------------------------------------------

    @test_case_log
    def test_case_suspend_power_measurement(self, host, device, max_power_mw,
                                            suspend_time_secs,
                                            resume_network_timeout_secs=60):
        """Test Case: measure the Bluetooth chip power consumption on suspend"""

        def print_debug_count():
            """Print the debug message about count values."""
            logging.debug('count_fail_to_sleep: %d', self.count_fail_to_sleep)
            logging.debug('count_fail_to_resume: %d', self.count_fail_to_resume)
            logging.debug('count_system_resume_prematurely: %d',
                          self.count_system_resume_prematurely)
            logging.debug('count_success: %d', self.count_success)

        def action_suspend():
            """Calls the host method suspend."""
            host.suspend(suspend_time=suspend_time_secs,
                         allow_early_resume=True)

        boot_id = host.get_boot_id()
        proc = multiprocessing.Process(target=action_suspend)
        proc.daemon = True
        start_time = time.time()
        proc.start()

        # Block waiting until the system has suspended.
        try:
            host.test_wait_for_sleep(suspend_time_secs)
        except Exception as e:
            logging.error('host.test_wait_for_sleep failed: %s', e)
            self.count_fail_to_sleep += 1
            print_debug_count()
            # Skip this time since the system failed to suspend.
            proc.join()
            return

        # Test the Bluetooth chip power consumption.
        if self.test_power_consumption(device, max_power_mw):
            self.count_success += 1

        # Block waiting until the system has resumed.
        try:
            host.test_wait_for_resume(
                    boot_id, suspend_time_secs + resume_network_timeout_secs)
        except Exception as e:
            logging.error('host.test_wait_for_resume failed: %s', e)
            self.count_fail_to_resume += 1

        # If the system resumes prematurely, do not conduct the test in
        # this iteration.
        actual_suspend_time_secs = time.time() - start_time
        if actual_suspend_time_secs < suspend_time_secs:
            logging.error('actual suspension time %f is less than expected %f',
                          actual_suspend_time_secs, suspend_time_secs)
            self.count_system_resume_prematurely += 1

        print_debug_count()
        proc.join()

        if self.count_success == 0:
            raise error.TestError('System failed to suspend/resume.')


    # ---------------------------------------------------------------
    # Definitions of test wrapper tests and batch wrapper tests.
    # ---------------------------------------------------------------


    @test_wrapper('Power measurement test', devices={'BLUETOOTH_BASE':1})
    def pw_measurement_suspension_test(self):
        """power measurement test during system suspension."""
        device = self.devices['BLUETOOTH_BASE'][0]
        self._initialize_servod(device)
        self.test_power_on_adapter()
        self.test_bluetoothd_running()
        self.test_case_suspend_power_measurement(self.host, device,
                                                 self.max_power_mw,
                                                 self.suspend_time_secs)
        self._cleanup_servod(device)


    @batch_wrapper('Bluetooth Power Measurement Health Tests')
    def pw_health_batch_run(self, num_iterations=1, test_name=None):
        """Run bluetooth power measurement health test batch or a specific test.

        @param num_iterations: how many iterations to run
        @param test_name: specific test to run otherwise None to run the
                whole batch
        """
        self.pw_measurement_suspension_test()


    def run_once(self,
                 host,
                 num_iterations=1,
                 args_dict=None,
                 test_name=None,
                 max_power_mw=3,
                 suspend_time_secs=30,
                 flag='Quick Health'):
        """Running Bluetooth adapter power consumption autotest during system
        suspension.

        @param host: the DUT host.
        @param num_iterations: number of times to perform the tests.
        @param test_name: the test to run, or None for all tests
        @param max_power_mw: max power allowed in milli-watt
        @param suspend_time_secs: the system suspension duration in seconds

        """
        self.host = host
        self.max_power_mw = max_power_mw
        self.suspend_time_secs = suspend_time_secs

        self.quick_test_init(host,
                             use_btpeer=True,
                             flag=flag,
                             args_dict=args_dict)
        self.pw_health_batch_run(num_iterations, test_name)
        self.quick_test_cleanup()