1#!/usr/bin/env python3 2# 3# Copyright (C) 2016 The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); you may not 6# use this file except in compliance with the License. You may obtain a copy of 7# the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 14# License for the specific language governing permissions and limitations under 15# the License. 16""" 17Basic LE Stress tests. 18""" 19 20import concurrent 21import pprint 22import time 23 24from queue import Empty 25from acts.test_decorators import test_tracker_info 26from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest 27from acts.test_utils.bt.bt_test_utils import BtTestUtilsError 28from acts.test_utils.bt.bt_test_utils import clear_bonded_devices 29from acts.test_utils.bt.bt_test_utils import generate_ble_advertise_objects 30from acts.test_utils.bt.bt_test_utils import generate_ble_scan_objects 31from acts.test_utils.bt.bt_test_utils import get_advanced_droid_list 32from acts.test_utils.bt.bt_test_utils import get_mac_address_of_generic_advertisement 33from acts.test_utils.bt.bt_test_utils import reset_bluetooth 34from acts.test_utils.bt.bt_constants import scan_result 35 36 37class BleStressTest(BluetoothBaseTest): 38 default_timeout = 10 39 PAIRING_TIMEOUT = 20 40 droid_list = [] 41 42 def setup_class(self): 43 super().setup_class() 44 self.droid_list = get_advanced_droid_list(self.android_devices) 45 self.scn_ad = self.android_devices[0] 46 self.adv_ad = self.android_devices[1] 47 48 def teardown_test(self): 49 super().teardown_test() 50 self.log_stats() 51 52 def bleadvertise_verify_onsuccess_handler(self, event): 53 test_result = True 54 self.log.debug("Verifying onSuccess event") 55 self.log.debug(pprint.pformat(event)) 56 return test_result 57 58 def _verify_successful_bond(self, target_address): 59 end_time = time.time() + self.PAIRING_TIMEOUT 60 self.log.info("Verifying devices are bonded") 61 while time.time() < end_time: 62 bonded_devices = self.scn_ad.droid.bluetoothGetBondedDevices() 63 if target_address in {d['address'] for d in bonded_devices}: 64 self.log.info("Successfully bonded to device") 65 return True 66 return False 67 68 @BluetoothBaseTest.bt_test_wrap 69 @test_tracker_info(uuid='22f98085-6ed8-4ad8-b62d-b8d1eae20b89') 70 def test_loop_scanning_1000(self): 71 """Stress start/stop scan instances. 72 73 This test will start and stop scan instances as fast as possible. This 74 will guarantee that the scan instances are properly being cleaned up 75 when the scan is stopped. 76 77 Steps: 78 1. Start a scan instance. 79 2. Stop the scan instance. 80 3. Repeat steps 1-2 1000 times. 81 82 Expected Result: 83 Neither starting or stopping scan instances causes any failures. 84 85 Returns: 86 Pass if True 87 Fail if False 88 89 TAGS: LE, Scanning, Stress 90 Priority: 1 91 """ 92 test_result = True 93 for _ in range(1000): 94 filter_list, scan_settings, scan_callback = generate_ble_scan_objects( 95 self.scn_ad.droid) 96 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 97 scan_callback) 98 self.scn_ad.droid.bleStopBleScan(scan_callback) 99 return test_result 100 101 @BluetoothBaseTest.bt_test_wrap 102 @test_tracker_info(uuid='6caa84c2-50ac-46f2-a5e5-f942fd2cd6f6') 103 def test_loop_scanning_100_verify_no_hci_timeout(self): 104 """Stress start/stop scan instances variant. 105 106 This test will start and stop scan instances with a one second timeout 107 in between each iteration. This testcase was added because the specific 108 timing combination caused hci timeouts. 109 110 Steps: 111 1. Start a scan instance. 112 2. Stop the scan instance. 113 3. Sleep for 1 second. 114 4. Repeat steps 1-3 100 times. 115 116 Expected Result: 117 Neither starting or stopping scan instances causes any failures. 118 119 Returns: 120 Pass if True 121 Fail if False 122 123 TAGS: LE, Scanning, Stress 124 Priority: 1 125 """ 126 for _ in range(self.droid_list[1]['max_advertisements']): 127 adv_callback, adv_data, adv_settings = generate_ble_advertise_objects( 128 self.adv_ad.droid) 129 self.adv_ad.droid.bleStartBleAdvertising(adv_callback, adv_data, 130 adv_settings) 131 for _ in range(100): 132 filter_list, scan_settings, scan_callback = generate_ble_scan_objects( 133 self.scn_ad.droid) 134 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 135 scan_callback) 136 self.log.info( 137 self.scn_ad.ed.pop_event(scan_result.format(scan_callback))) 138 self.scn_ad.droid.bleStopBleScan(scan_callback) 139 time.sleep(1) 140 return True 141 142 @BluetoothBaseTest.bt_test_wrap 143 @test_tracker_info(uuid='5e9e4c8d-b72e-4767-81e5-f907c1834430') 144 def test_loop_advertising_100(self): 145 """Stress start/stop advertising instances. 146 147 This test will start and stop advertising instances as fast as possible. 148 149 Steps: 150 1. Start a advertising instance. 151 2. Find that an onSuccess callback is triggered. 152 3. Stop the advertising instance. 153 4. Repeat steps 1-3 100 times. 154 155 Expected Result: 156 Neither starting or stopping advertising instances causes any failures. 157 158 Returns: 159 Pass if True 160 Fail if False 161 162 TAGS: LE, Advertising, Stress 163 Priority: 1 164 """ 165 test_result = True 166 for _ in range(100): 167 advertise_callback, advertise_data, advertise_settings = generate_ble_advertise_objects( 168 self.adv_ad.droid) 169 self.adv_ad.droid.bleStartBleAdvertising( 170 advertise_callback, advertise_data, advertise_settings) 171 expected_advertise_event_name = "".join( 172 ["BleAdvertise", 173 str(advertise_callback), "onSuccess"]) 174 worker = self.adv_ad.ed.handle_event( 175 self.bleadvertise_verify_onsuccess_handler, 176 expected_advertise_event_name, ([]), self.default_timeout) 177 try: 178 self.log.debug(worker.result(self.default_timeout)) 179 except Empty as error: 180 self.log.debug(" ".join( 181 ["Test failed with Empty error:", 182 str(error)])) 183 test_result = False 184 except concurrent.futures._base.TimeoutError as error: 185 self.log.debug(" ".join([ 186 "Test failed, filtering callback onSuccess never occurred:", 187 str(error) 188 ])) 189 test_result = False 190 self.adv_ad.droid.bleStopBleAdvertising(advertise_callback) 191 return test_result 192 193 @BluetoothBaseTest.bt_test_wrap 194 @test_tracker_info(uuid='11a2f51b-7178-4c32-bb5c-7eddd100a50f') 195 def test_restart_advertise_callback_after_bt_toggle(self): 196 """Test to reuse an advertise callback. 197 198 This will verify if advertising objects can be reused after a bluetooth 199 toggle. 200 201 Steps: 202 1. Start a advertising instance. 203 2. Find that an onSuccess callback is triggered. 204 3. Stop the advertising instance. 205 4. Toggle bluetooth off and on. 206 5. Start an advertising instance on the same objects used in step 1. 207 6. Find that an onSuccess callback is triggered. 208 209 Expected Result: 210 Advertisement should start successfully. 211 212 Returns: 213 Pass if True 214 Fail if False 215 216 TAGS: LE, Advertising, Stress 217 Priority: 1 218 """ 219 test_result = True 220 advertise_callback, advertise_data, advertise_settings = generate_ble_advertise_objects( 221 self.adv_ad.droid) 222 self.adv_ad.droid.bleStartBleAdvertising( 223 advertise_callback, advertise_data, advertise_settings) 224 expected_advertise_event_name = "".join( 225 ["BleAdvertise", 226 str(advertise_callback), "onSuccess"]) 227 worker = self.adv_ad.ed.handle_event( 228 self.bleadvertise_verify_onsuccess_handler, 229 expected_advertise_event_name, ([]), self.default_timeout) 230 try: 231 self.log.debug(worker.result(self.default_timeout)) 232 except Empty as error: 233 self.log.debug(" ".join( 234 ["Test failed with Empty error:", 235 str(error)])) 236 test_result = False 237 except concurrent.futures._base.TimeoutError as error: 238 self.log.debug(" ".join([ 239 "Test failed, filtering callback onSuccess never occurred:", 240 str(error) 241 ])) 242 test_result = reset_bluetooth([self.scn_ad]) 243 if not test_result: 244 return test_result 245 time.sleep(5) 246 self.adv_ad.droid.bleStartBleAdvertising( 247 advertise_callback, advertise_data, advertise_settings) 248 worker = self.adv_ad.ed.handle_event( 249 self.bleadvertise_verify_onsuccess_handler, 250 expected_advertise_event_name, ([]), self.default_timeout) 251 try: 252 self.log.debug(worker.result(self.default_timeout)) 253 except Empty as error: 254 self.log.debug(" ".join( 255 ["Test failed with Empty error:", 256 str(error)])) 257 test_result = False 258 except concurrent.futures._base.TimeoutError as error: 259 self.log.debug(" ".join([ 260 "Test failed, filtering callback onSuccess never occurred:", 261 str(error) 262 ])) 263 return test_result 264 265 @BluetoothBaseTest.bt_test_wrap 266 @test_tracker_info(uuid='88f3c068-41be-41df-920c-c0ecaae1a619') 267 def test_restart_scan_callback_after_bt_toggle(self): 268 """Test to reuse an scan callback. 269 270 This will verify if scan objects can be reused after a bluetooth 271 toggle. 272 273 Steps: 274 1. Start a scanning instance. 275 3. Stop the scanning instance. 276 4. Toggle bluetooth off and on. 277 5. Start an scanning instance on the same objects used in step 1. 278 279 Expected Result: 280 Scanner should start successfully. 281 282 Returns: 283 Pass if True 284 Fail if False 285 286 TAGS: LE, Scanning, Stress 287 Priority: 1 288 """ 289 test_result = True 290 filter_list, scan_settings, scan_callback = generate_ble_scan_objects( 291 self.scn_ad.droid) 292 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 293 scan_callback) 294 reset_bluetooth([self.scn_ad]) 295 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 296 scan_callback) 297 298 return test_result 299 300 @BluetoothBaseTest.bt_test_wrap 301 @test_tracker_info(uuid='ce8adfc0-384f-4751-9438-13a76cada8da') 302 def test_le_pairing(self): 303 """Test LE pairing transport stress 304 305 This will test LE pairing between two android devices. 306 307 Steps: 308 1. Start an LE advertisement on secondary device. 309 2. Find address from primary device. 310 3. Discover and bond to LE address. 311 4. Stop LE advertisement on secondary device. 312 5. Repeat steps 1-4 100 times 313 314 Expected Result: 315 LE pairing should pass 100 times. 316 317 Returns: 318 Pass if True 319 Fail if False 320 321 TAGS: LE, Scanning, Stress, Pairing 322 Priority: 1 323 """ 324 iterations = 100 325 for i in range(iterations): 326 try: 327 target_address, adv_callback, scan_callback = get_mac_address_of_generic_advertisement( 328 self.scn_ad, self.adv_ad) 329 except BtTestUtilsError as err: 330 self.log.error(err) 331 return False 332 self.log.info("Begin interation {}/{}".format(i + 1, iterations)) 333 self.scn_ad.droid.bluetoothStartPairingHelper() 334 self.adv_ad.droid.bluetoothStartPairingHelper() 335 start_time = self.start_timer() 336 self.scn_ad.droid.bluetoothDiscoverAndBond(target_address) 337 if not self._verify_successful_bond(target_address): 338 self.log.error("Failed to bond devices.") 339 return False 340 self.log.info("Total time (ms): {}".format(self.end_timer())) 341 if not self._verify_successful_bond(self.adv_ad.droid.bluetoothGetLocalAddress()): 342 self.log.error("Failed to bond BREDR devices.") 343 return False 344 if not self.scn_ad.droid.bluetoothUnbond(target_address): 345 self.log.error("Failed to unbond device from scanner.") 346 return False 347 time.sleep(2) 348 if not self.adv_ad.droid.bluetoothUnbond(self.scn_ad.droid.bluetoothGetLocalAddress()): 349 self.log.error("Failed to unbond device from advertiser.") 350 return False 351 self.adv_ad.droid.bleStopBleAdvertising(adv_callback) 352 self.scn_ad.droid.bleStopBleScan(scan_callback) 353 # Magic sleep to let unbonding finish 354 time.sleep(2) 355 return True 356