1# Lint as: python2, python3 2# Copyright 2020 The Chromium OS Authors. All rights reserved. 3# Use of this source code is governed by a BSD-style license that can be 4# found in the LICENSE file. 5 6"""A Bluetooth adapter MTBF test of some Bluetooth use cases. 7 8 To add a new use case we just need to inherit from the existing test class 9 and then call the desired test methods in the batch method below. This allows 10 the test case to be used as both part of a MTBF batch and a normal batch. 11""" 12 13from __future__ import absolute_import 14from __future__ import division 15from __future__ import print_function 16 17import threading 18import time 19 20import common 21from autotest_lib.client.common_lib import error 22from autotest_lib.server.cros.bluetooth.bluetooth_adapter_audio_tests import ( 23 BluetoothAdapterAudioTests) 24from autotest_lib.server.cros.bluetooth.bluetooth_adapter_better_together \ 25 import BluetoothAdapterBetterTogether 26from autotest_lib.server.cros.bluetooth.bluetooth_adapter_hidreports_tests \ 27 import BluetoothAdapterHIDReportTests 28from autotest_lib.server.cros.bluetooth.bluetooth_adapter_quick_tests import ( 29 BluetoothAdapterQuickTests) 30from autotest_lib.server.cros.bluetooth.bluetooth_adapter_tests import ( 31 TABLET_MODELS) 32from autotest_lib.client.cros.bluetooth.bluetooth_audio_test_data import A2DP_LONG 33from six.moves import range 34 35# Iterations to run the mouse report test, this equals about 10 mins 36MOUSE_TEST_ITERATION = 15 37# Iterations to run the keyboard report test, this equals about 10 mins 38KEYBOARD_TEST_ITERATION = 60 39A2DP_TEST_DURATION_SEC = 600 40# Wait for some time before stating a new concurrent thread 41SLEEP_BETWEEN_THREADS = 15 42 43class bluetooth_AdapterMTBF(BluetoothAdapterBetterTogether, 44 BluetoothAdapterHIDReportTests, 45 BluetoothAdapterAudioTests): 46 """A Batch of Bluetooth adapter tests for MTBF. This test is written 47 as a batch of tests in order to reduce test time, since auto-test ramp up 48 time is costly. The batch is using BluetoothAdapterQuickTests wrapper 49 methods to start and end a test and a batch of tests. 50 51 This class can be called to run the entire test batch or to run a 52 specific test only 53 """ 54 55 MTBF_TIMEOUT_MINS = 300 56 batch_wrapper = BluetoothAdapterQuickTests.quick_test_batch_decorator 57 mtbf_wrapper = BluetoothAdapterQuickTests.quick_test_mtbf_decorator 58 test_wrapper = BluetoothAdapterQuickTests.quick_test_test_decorator 59 60 @test_wrapper('MTBF Typical Use Cases', 61 devices={'BLE_MOUSE': 1, 62 'BLE_PHONE': 1, 63 'BLUETOOTH_AUDIO': 1, 64 'KEYBOARD': 1}) 65 def typical_use_cases_test(self): 66 """Do some initialization work then start the typical MTBF test loop""" 67 68 # TODO(b/165606673) - Remove blooglet once the bug is fixed 69 self.skip_wake_test = self.host.get_model_from_cros_config() in \ 70 TABLET_MODELS + ['blooglet'] 71 mouse = self.devices['BLE_MOUSE'][0] 72 phone = self.devices['BLE_PHONE'][0] 73 audio = self.devices['BLUETOOTH_AUDIO'][0] 74 keyboard = self.devices['KEYBOARD'][0] 75 76 self.test_device_pairing(mouse) 77 self.test_device_pairing(keyboard) 78 79 self.run_typical_use_cases(mouse, phone, audio, keyboard) 80 81 82 @mtbf_wrapper(timeout_mins=MTBF_TIMEOUT_MINS, test_name='typical_use_cases') 83 def run_typical_use_cases(self, mouse, phone, audio, keyboard): 84 """Run typical MTBF test scenarios: 85 1. Run the better together test 86 2. Run the concurrent mouse and A2DP tests for 10 minutes 87 3. Suspend/Resume 88 4. Run the concurrent mouse amd keyboard tests for 10 minutes 89 5. Suspend and wake up the DUT by mouse 90 6. Run the concurrent mouse, keyboard and A2DP tests for 10 minutes 91 """ 92 # Run the better together test on the phone 93 self.test_better_together(phone) 94 95 # Restore the discovery filter since better together test changed it 96 self.test_set_discovery_filter({'Transport':'auto'}) 97 98 self.test_mouse_and_audio(mouse, audio) 99 self.test_suspend_resume(mouse, keyboard) 100 self.test_mouse_and_keyboard(mouse, keyboard) 101 # Mouse wakeup test 102 self.test_suspend_and_mouse_wakeup(mouse, keyboard) 103 self.test_hid_and_audio(mouse, keyboard, audio) 104 105 106 def test_mouse_and_audio(self, mouse, audio): 107 """Run the mouse and audio tests concurrently for 10 mins""" 108 audio_thread = threading.Thread( 109 target=self.test_audio, args=(audio, A2DP_TEST_DURATION_SEC)) 110 mouse_thread = threading.Thread( 111 target=self.test_mouse, args=(mouse, MOUSE_TEST_ITERATION)) 112 113 audio_thread.start() 114 time.sleep(SLEEP_BETWEEN_THREADS) 115 mouse_thread.start() 116 time.sleep(SLEEP_BETWEEN_THREADS) 117 audio_thread.join() 118 mouse_thread.join() 119 if self.fails: 120 raise error.TestFail(self.fails) 121 122 123 def test_mouse_and_keyboard(self, mouse, keyboard): 124 """Run the mouse and keyboard tests concurrently for 10 mins""" 125 mouse_thread = threading.Thread( 126 target=self.test_mouse, args=(mouse, MOUSE_TEST_ITERATION)) 127 keyboard_thread = \ 128 threading.Thread(target=self.test_keyboard, 129 args=(keyboard, KEYBOARD_TEST_ITERATION)) 130 time.sleep(SLEEP_BETWEEN_THREADS) 131 mouse_thread.start() 132 time.sleep(SLEEP_BETWEEN_THREADS) 133 keyboard_thread.start() 134 mouse_thread.join() 135 keyboard_thread.join() 136 if self.fails: 137 raise error.TestFail(self.fails) 138 139 140 def test_hid_and_audio(self, mouse, keyboard, audio): 141 """Run the audio, mouse and keyboard tests concurrently for 10 mins""" 142 audio_thread = threading.Thread( 143 target=self.test_audio, args=(audio, A2DP_TEST_DURATION_SEC)) 144 mouse_thread = threading.Thread( 145 target=self.test_mouse, args=(mouse, MOUSE_TEST_ITERATION)) 146 keyboard_thread = \ 147 threading.Thread(target=self.test_keyboard, 148 args=(keyboard, KEYBOARD_TEST_ITERATION)) 149 audio_thread.start() 150 time.sleep(SLEEP_BETWEEN_THREADS) 151 mouse_thread.start() 152 time.sleep(SLEEP_BETWEEN_THREADS) 153 keyboard_thread.start() 154 audio_thread.join() 155 mouse_thread.join() 156 keyboard_thread.join() 157 if self.fails: 158 raise error.TestFail(self.fails) 159 160 161 def test_mouse(self, mouse, iteration): 162 """Run mouse report test for certain iterations""" 163 for i in range(iteration): 164 self.run_mouse_tests(device=mouse) 165 166 167 def test_keyboard(self, keyboard, iteration): 168 """Run keyboard report test for certain iterations""" 169 for i in range(iteration): 170 self.run_keyboard_tests(device=keyboard) 171 172 173 def test_audio(self, device, duration): 174 """Test A2DP 175 176 This test plays A2DP audio on the DUT and record on the peer device, 177 then verify the legitimacy of the frames recorded. 178 179 """ 180 self.bluetooth_facade.remove_device_object(device.address) 181 device.RemoveDevice(self.bluetooth_facade.address) 182 183 self.initialize_bluetooth_audio(device, A2DP_LONG) 184 self.test_device_set_discoverable(device, True) 185 self.test_discover_device(device.address) 186 self.test_pairing(device.address, device.pin, trusted=True) 187 device.SetTrustedByRemoteAddress(self.bluetooth_facade.address) 188 self.test_connection_by_adapter(device.address) 189 self.test_a2dp_sinewaves(device, A2DP_LONG, duration) 190 self.test_disconnection_by_adapter(device.address) 191 self.cleanup_bluetooth_audio(device, A2DP_LONG) 192 self.test_remove_device_object(device.address) 193 194 195 def test_device_pairing(self, device): 196 """Test device pairing""" 197 198 # Remove the pairing first 199 self.bluetooth_facade.remove_device_object(device.address) 200 device.RemoveDevice(self.bluetooth_facade.address) 201 202 self.test_device_set_discoverable(device, True) 203 self.test_discover_device(device.address) 204 time.sleep(self.TEST_SLEEP_SECS) 205 self.test_pairing(device.address, device.pin, trusted=True) 206 time.sleep(self.TEST_SLEEP_SECS) 207 self.test_connection_by_adapter(device.address) 208 209 210 def test_suspend_resume(self, mouse, keyboard): 211 """Test the device can connect after suspending and resuming""" 212 boot_id = self.host.get_boot_id() 213 suspend = self.suspend_async(suspend_time=15) 214 start_time = self.bluetooth_facade.get_device_time() 215 216 self.test_device_set_discoverable(mouse, False) 217 218 self.test_suspend_and_wait_for_sleep( 219 suspend, sleep_timeout=15) 220 self.test_wait_for_resume(boot_id, 221 suspend, 222 resume_timeout=15, 223 test_start_time=start_time) 224 225 # LE can't reconnect without advertising/discoverable 226 self.test_device_set_discoverable(mouse, True) 227 self.test_device_is_connected(mouse.address) 228 self.test_hid_device_created(mouse.address) 229 230 self.test_connection_by_device(keyboard) 231 self.test_hid_device_created(keyboard.address) 232 233 234 def test_suspend_and_mouse_wakeup(self, mouse, keyboard): 235 """Test the device can be waken up by the mouse""" 236 if self.skip_wake_test: 237 return 238 boot_id = self.host.get_boot_id() 239 suspend = self.suspend_async( 240 suspend_time=60, expect_bt_wake=True) 241 start_time = self.bluetooth_facade.get_device_time() 242 243 self.test_adapter_wake_enabled() 244 self.test_suspend_and_wait_for_sleep( 245 suspend, sleep_timeout=5) 246 247 # Trigger peer wakeup 248 peer_wake = self.device_connect_async('BLE_MOUSE', mouse, 249 self.bluetooth_facade.address) 250 peer_wake.start() 251 252 # Expect a quick resume. If a timeout occurs, test fails. 253 self.test_wait_for_resume(boot_id, 254 suspend, 255 resume_timeout=20, 256 test_start_time=start_time, 257 fail_on_timeout=True) 258 259 # Finish peer wake process 260 peer_wake.join() 261 262 # Make sure we're actually connected 263 self.test_device_is_connected(mouse.address) 264 self.test_hid_device_created(mouse.address) 265 266 self.test_connection_by_device(keyboard) 267 self.test_hid_device_created(keyboard.address) 268 269 270 @test_wrapper('MTBF Better Together Stress', devices={'BLE_PHONE': 1}) 271 def better_together_stress_test(self): 272 """Run better together stress test""" 273 274 phone = self.devices['BLE_PHONE'][0] 275 phone.RemoveDevice(self.bluetooth_facade.address) 276 self.run_better_together_stress(address=phone.address) 277 278 279 def test_better_together(self, phone): 280 """Test better together""" 281 # Clean up the environment 282 self.bluetooth_facade.disconnect_device(phone.address) 283 self.bluetooth_facade.remove_device_object(phone.address) 284 phone.RemoveDevice(self.bluetooth_facade.address) 285 self.test_smart_unlock(address=phone.address) 286 287 288 @mtbf_wrapper( 289 timeout_mins=MTBF_TIMEOUT_MINS, test_name='better_together_stress') 290 def run_better_together_stress(self, address): 291 """Run better together stress test""" 292 293 self.test_smart_unlock(address) 294 295 296 @batch_wrapper('Adapter MTBF') 297 def mtbf_batch_run(self, num_iterations=1, test_name=None): 298 """Run the Bluetooth MTBF test batch or a specific 299 given test. The wrapper of this method is implemented in 300 batch_decorator. Using the decorator a test batch method can 301 implement the only its core tests invocations and let the 302 decorator handle the wrapper, which is taking care for whether to 303 run a specific test or the batch as a whole, and running the batch 304 in iterations 305 306 @param num_iterations: how many iterations to run 307 @param test_name: specific test to run otherwise None to run the 308 whole batch 309 """ 310 # TODO: finalize the test cases that need to be run as MTBF 311 self.typical_use_cases_test() 312 313 314 def run_once(self, 315 host, 316 num_iterations=1, 317 test_name=None, 318 args_dict=None): 319 """Run the batch of Bluetooth MTBF tests 320 321 @param host: the DUT, usually a chromebook 322 @param num_iterations: the number of rounds to execute the test 323 @test_name: the test to run, or None for all tests 324 """ 325 326 # Initialize and run the test batch or the requested specific test 327 self.set_fail_fast(args_dict, True) 328 self.quick_test_init(host, use_btpeer=True, args_dict=args_dict) 329 330 self.mtbf_batch_run(num_iterations, test_name) 331 332 self.quick_test_cleanup() 333