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

"""A Bluetooth adapter MTBF test of some Bluetooth use cases.

   To add a new use case we just need to inherit from the existing test class
   and then call the desired test methods in the batch method below. This allows
   the test case to be used as both part of a MTBF batch and a normal batch.
"""

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

import threading
import time

import common
from autotest_lib.client.common_lib import error
from autotest_lib.server.cros.bluetooth.bluetooth_adapter_audio_tests import (
        BluetoothAdapterAudioTests)
from autotest_lib.server.cros.bluetooth.bluetooth_adapter_better_together \
        import BluetoothAdapterBetterTogether
from autotest_lib.server.cros.bluetooth.bluetooth_adapter_hidreports_tests \
        import BluetoothAdapterHIDReportTests
from autotest_lib.server.cros.bluetooth.bluetooth_adapter_quick_tests import (
        BluetoothAdapterQuickTests)
from autotest_lib.server.cros.bluetooth.bluetooth_adapter_tests import (
        TABLET_MODELS)
from autotest_lib.client.cros.bluetooth.bluetooth_audio_test_data import A2DP_LONG
from six.moves import range

# Iterations to run the mouse report test, this equals about 10 mins
MOUSE_TEST_ITERATION = 15
# Iterations to run the keyboard report test, this equals about 10 mins
KEYBOARD_TEST_ITERATION = 60
A2DP_TEST_DURATION_SEC = 600
# Wait for some time before stating a new concurrent thread
SLEEP_BETWEEN_THREADS = 15

class bluetooth_AdapterMTBF(BluetoothAdapterBetterTogether,
                            BluetoothAdapterHIDReportTests,
                            BluetoothAdapterAudioTests):
    """A Batch of Bluetooth adapter tests for MTBF. This test is written
       as a batch of tests in order to reduce test time, since auto-test ramp up
       time is costly. The batch is using BluetoothAdapterQuickTests wrapper
       methods to start and end a test and a batch of tests.

       This class can be called to run the entire test batch or to run a
       specific test only
    """

    MTBF_TIMEOUT_MINS = 300
    batch_wrapper = BluetoothAdapterQuickTests.quick_test_batch_decorator
    mtbf_wrapper = BluetoothAdapterQuickTests.quick_test_mtbf_decorator
    test_wrapper = BluetoothAdapterQuickTests.quick_test_test_decorator

    @test_wrapper('MTBF Typical Use Cases',
                  devices={'BLE_MOUSE': 1,
                           'BLE_PHONE': 1,
                           'BLUETOOTH_AUDIO': 1,
                           'KEYBOARD': 1})
    def typical_use_cases_test(self):
        """Do some initialization work then start the typical MTBF test loop"""

        # TODO(b/165606673) - Remove blooglet once the bug is fixed
        self.skip_wake_test = self.host.get_model_from_cros_config() in \
                              TABLET_MODELS + ['blooglet']
        mouse = self.devices['BLE_MOUSE'][0]
        phone = self.devices['BLE_PHONE'][0]
        audio = self.devices['BLUETOOTH_AUDIO'][0]
        keyboard = self.devices['KEYBOARD'][0]

        self.test_device_pairing(mouse)
        self.test_device_pairing(keyboard)

        self.run_typical_use_cases(mouse, phone, audio, keyboard)


    @mtbf_wrapper(timeout_mins=MTBF_TIMEOUT_MINS, test_name='typical_use_cases')
    def run_typical_use_cases(self, mouse, phone, audio, keyboard):
        """Run typical MTBF test scenarios:
           1. Run the better together test
           2. Run the concurrent mouse and A2DP tests for 10 minutes
           3. Suspend/Resume
           4. Run the concurrent mouse amd keyboard tests for 10 minutes
           5. Suspend and wake up the DUT by mouse
           6. Run the concurrent mouse, keyboard and A2DP tests for 10 minutes
        """
        # Run the better together test on the phone
        self.test_better_together(phone)

        # Restore the discovery filter since better together test changed it
        self.test_set_discovery_filter({'Transport':'auto'})

        self.test_mouse_and_audio(mouse, audio)
        self.test_suspend_resume(mouse, keyboard)
        self.test_mouse_and_keyboard(mouse, keyboard)
        # Mouse wakeup test
        self.test_suspend_and_mouse_wakeup(mouse, keyboard)
        self.test_hid_and_audio(mouse, keyboard, audio)


    def test_mouse_and_audio(self, mouse, audio):
        """Run the mouse and audio tests concurrently for 10 mins"""
        audio_thread = threading.Thread(
            target=self.test_audio, args=(audio, A2DP_TEST_DURATION_SEC))
        mouse_thread = threading.Thread(
            target=self.test_mouse, args=(mouse, MOUSE_TEST_ITERATION))

        audio_thread.start()
        time.sleep(SLEEP_BETWEEN_THREADS)
        mouse_thread.start()
        time.sleep(SLEEP_BETWEEN_THREADS)
        audio_thread.join()
        mouse_thread.join()
        if self.fails:
            raise error.TestFail(self.fails)


    def test_mouse_and_keyboard(self, mouse, keyboard):
        """Run the mouse and keyboard tests concurrently for 10 mins"""
        mouse_thread = threading.Thread(
            target=self.test_mouse, args=(mouse, MOUSE_TEST_ITERATION))
        keyboard_thread = \
            threading.Thread(target=self.test_keyboard,
                             args=(keyboard, KEYBOARD_TEST_ITERATION))
        time.sleep(SLEEP_BETWEEN_THREADS)
        mouse_thread.start()
        time.sleep(SLEEP_BETWEEN_THREADS)
        keyboard_thread.start()
        mouse_thread.join()
        keyboard_thread.join()
        if self.fails:
            raise error.TestFail(self.fails)


    def test_hid_and_audio(self, mouse, keyboard, audio):
        """Run the audio, mouse and keyboard tests concurrently for 10 mins"""
        audio_thread = threading.Thread(
            target=self.test_audio, args=(audio, A2DP_TEST_DURATION_SEC))
        mouse_thread = threading.Thread(
            target=self.test_mouse, args=(mouse, MOUSE_TEST_ITERATION))
        keyboard_thread = \
            threading.Thread(target=self.test_keyboard,
                             args=(keyboard, KEYBOARD_TEST_ITERATION))
        audio_thread.start()
        time.sleep(SLEEP_BETWEEN_THREADS)
        mouse_thread.start()
        time.sleep(SLEEP_BETWEEN_THREADS)
        keyboard_thread.start()
        audio_thread.join()
        mouse_thread.join()
        keyboard_thread.join()
        if self.fails:
            raise error.TestFail(self.fails)


    def test_mouse(self, mouse, iteration):
        """Run mouse report test for certain iterations"""
        for i in range(iteration):
            self.run_mouse_tests(device=mouse)


    def test_keyboard(self, keyboard, iteration):
        """Run keyboard report test for certain iterations"""
        for i in range(iteration):
            self.run_keyboard_tests(device=keyboard)


    def test_audio(self, device, duration):
        """Test A2DP

           This test plays A2DP audio on the DUT and record on the peer device,
           then verify the legitimacy of the frames recorded.

        """
        self.bluetooth_facade.remove_device_object(device.address)
        device.RemoveDevice(self.bluetooth_facade.address)

        self.initialize_bluetooth_audio(device, A2DP_LONG)
        self.test_device_set_discoverable(device, True)
        self.test_discover_device(device.address)
        self.test_pairing(device.address, device.pin, trusted=True)
        device.SetTrustedByRemoteAddress(self.bluetooth_facade.address)
        self.test_connection_by_adapter(device.address)
        self.test_a2dp_sinewaves(device, A2DP_LONG, duration)
        self.test_disconnection_by_adapter(device.address)
        self.cleanup_bluetooth_audio(device, A2DP_LONG)
        self.test_remove_device_object(device.address)


    def test_device_pairing(self, device):
        """Test device pairing"""

        # Remove the pairing first
        self.bluetooth_facade.remove_device_object(device.address)
        device.RemoveDevice(self.bluetooth_facade.address)

        self.test_device_set_discoverable(device, True)
        self.test_discover_device(device.address)
        time.sleep(self.TEST_SLEEP_SECS)
        self.test_pairing(device.address, device.pin, trusted=True)
        time.sleep(self.TEST_SLEEP_SECS)
        self.test_connection_by_adapter(device.address)


    def test_suspend_resume(self, mouse, keyboard):
        """Test the device can connect after suspending and resuming"""
        boot_id = self.host.get_boot_id()
        suspend = self.suspend_async(suspend_time=15)
        start_time = self.bluetooth_facade.get_device_utc_time()

        self.test_device_set_discoverable(mouse, False)

        self.test_suspend_and_wait_for_sleep(
            suspend, sleep_timeout=15)
        self.test_wait_for_resume(boot_id,
                                  suspend,
                                  resume_timeout=15,
                                  test_start_time=start_time)

        # LE can't reconnect without advertising/discoverable
        self.test_device_set_discoverable(mouse, True)
        self.test_device_is_connected(mouse.address)
        self.test_hid_device_created(mouse.address)

        self.test_connection_by_device(keyboard)
        self.test_hid_device_created(keyboard.address)


    def test_suspend_and_mouse_wakeup(self, mouse, keyboard):
        """Test the device can be waken up by the mouse"""
        if self.skip_wake_test:
            return

        self.run_peer_wakeup_device('MOUSE', mouse, should_pair=False)
        # Make sure we're actually connected
        self.test_device_is_connected(mouse.address)
        self.test_hid_device_created(mouse.address)

        self.test_connection_by_device(keyboard)
        self.test_hid_device_created(keyboard.address)


    @test_wrapper('MTBF Better Together Stress', devices={'BLE_PHONE': 1})
    def better_together_stress_test(self):
        """Run better together stress test"""

        phone = self.devices['BLE_PHONE'][0]
        phone.RemoveDevice(self.bluetooth_facade.address)
        self.run_better_together_stress(address=phone.address)


    def test_better_together(self, phone):
        """Test better together"""
        # Clean up the environment
        self.bluetooth_facade.disconnect_device(phone.address)
        self.bluetooth_facade.remove_device_object(phone.address)
        phone.RemoveDevice(self.bluetooth_facade.address)
        self.test_smart_unlock(address=phone.address)


    @mtbf_wrapper(
        timeout_mins=MTBF_TIMEOUT_MINS, test_name='better_together_stress')
    def run_better_together_stress(self, address):
        """Run better together stress test"""

        self.test_smart_unlock(address)


    @batch_wrapper('Adapter MTBF')
    def mtbf_batch_run(self, num_iterations=1, test_name=None):
        """Run the Bluetooth MTBF test batch or a specific
           given test. The wrapper of this method is implemented in
           batch_decorator. Using the decorator a test batch method can
           implement the 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 running the batch
           in iterations

           @param num_iterations: how many iterations to run
           @param test_name: specific test to run otherwise None to run the
                             whole batch
        """
        # TODO: finalize the test cases that need to be run as MTBF
        self.typical_use_cases_test()


    def run_once(self,
                 host,
                 num_iterations=1,
                 test_name=None,
                 args_dict=None):
        """Run the batch of Bluetooth MTBF tests

        @param host: the DUT, usually a chromebook
        @param num_iterations: the number of rounds to execute the test
        @test_name: the test to run, or None for all tests
        """

        # Initialize and run the test batch or the requested specific test
        self.set_fail_fast(args_dict, True)
        self.quick_test_init(host, use_btpeer=True, args_dict=args_dict)

        self.mtbf_batch_run(num_iterations, test_name)

        self.quick_test_cleanup()