# 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()