1#/usr/bin/env python3.4 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""" 17Test script to exercises Ble Scans can run in concurrency. 18This test was designed to be run in a shield box. 19""" 20 21import concurrent 22import time 23 24from queue import Empty 25from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest 26from acts.test_utils.bt.BleEnum import AdvertiseSettingsAdvertiseMode 27from acts.test_utils.bt.BleEnum import ScanSettingsCallbackType 28from acts.test_utils.bt.BleEnum import ScanSettingsScanMode 29from acts.test_utils.bt.bt_test_utils import adv_succ 30from acts.test_utils.bt.bt_test_utils import generate_ble_advertise_objects 31from acts.test_utils.bt.bt_test_utils import reset_bluetooth 32from acts.test_utils.bt.bt_test_utils import scan_failed 33from acts.test_utils.bt.bt_test_utils import scan_result 34from acts.test_utils.bt.bt_test_utils import take_btsnoop_logs 35 36 37class ConcurrentBleScanningTest(BluetoothBaseTest): 38 default_timeout = 20 39 max_concurrent_scans = 28 40 41 def __init__(self, controllers): 42 BluetoothBaseTest.__init__(self, controllers) 43 self.scn_ad = self.android_devices[0] 44 self.adv_ad = self.android_devices[1] 45 46 def on_fail(self, test_name, begin_time): 47 self.log.debug("Test {} failed. Gathering bugreport and btsnoop logs." 48 .format(test_name)) 49 take_btsnoop_logs(self.android_devices, self, test_name) 50 reset_bluetooth(self.android_devices) 51 52 def setup_test(self): 53 return reset_bluetooth(self.android_devices) 54 55 @BluetoothBaseTest.bt_test_wrap 56 def test_max_concurrent_ble_scans(self): 57 """Test max LE scans. 58 59 Test that a single device can have max scans concurrently scanning. 60 61 Steps: 62 1. Initialize scanner 63 2. Initialize advertiser 64 3. Start advertising on the device from step 2 65 4. Create max ble scan callbacks 66 5. Start ble scan on each callback 67 6. Verify that each callback triggers 68 7. Stop all scans and advertisements 69 70 Expected Result: 71 All scanning instances should start without errors and the advertisement 72 should be found on each scan instance. 73 74 Returns: 75 Pass if True 76 Fail if False 77 78 TAGS: LE, Scanning, Concurrency 79 Priority: 0 80 """ 81 test_result = True 82 self.adv_ad.droid.bleSetAdvertiseDataIncludeDeviceName(True) 83 self.scn_ad.droid.bleSetScanSettingsCallbackType( 84 ScanSettingsCallbackType.CALLBACK_TYPE_ALL_MATCHES.value) 85 self.scn_ad.droid.bleSetScanSettingsScanMode( 86 ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value) 87 self.adv_ad.droid.bleSetAdvertiseSettingsAdvertiseMode( 88 AdvertiseSettingsAdvertiseMode.ADVERTISE_MODE_LOW_LATENCY.value) 89 advertise_callback, advertise_data, advertise_settings = ( 90 generate_ble_advertise_objects(self.adv_ad.droid)) 91 self.adv_ad.droid.bleSetAdvertiseSettingsIsConnectable(False) 92 self.adv_ad.droid.bleStartBleAdvertising( 93 advertise_callback, advertise_data, advertise_settings) 94 try: 95 self.adv_ad.ed.pop_event( 96 adv_succ.format(advertise_callback), self.default_timeout) 97 except Empty as error: 98 self.log.exception("Test failed with Empty error: {}".format( 99 error)) 100 test_result = False 101 except concurrent.futures._base.TimeoutError as error: 102 self.log.exception( 103 "Test failed callback onSuccess never occurred: " 104 "{}".format(error)) 105 test_result = False 106 if not test_result: 107 return test_result 108 filter_list = self.scn_ad.droid.bleGenFilterList() 109 self.scn_ad.droid.bleSetScanFilterDeviceName( 110 self.adv_ad.droid.bluetoothGetLocalName()) 111 self.scn_ad.droid.bleBuildScanFilter(filter_list) 112 scan_settings = self.scn_ad.droid.bleBuildScanSetting() 113 scan_callback_list = [] 114 for i in range(self.max_concurrent_scans): 115 self.log.debug("Concurrent Ble Scan iteration {}".format(i + 1)) 116 scan_callback = self.scn_ad.droid.bleGenScanCallback() 117 scan_callback_list.append(scan_callback) 118 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 119 scan_callback) 120 try: 121 self.scn_ad.ed.pop_event( 122 scan_result.format(scan_callback), self.default_timeout) 123 self.log.info("Found scan event successfully. Iteration {} " 124 "successful.".format(i)) 125 except Exception: 126 self.log.info("Failed to find a scan result for callback {}" 127 .format(scan_callback)) 128 test_result = False 129 break 130 for callback in scan_callback_list: 131 self.scn_ad.droid.bleStopBleScan(callback) 132 self.adv_ad.droid.bleStopBleAdvertising(advertise_callback) 133 if not test_result: 134 return test_result 135 self.log.info("Waiting for scan callbacks to stop completely.") 136 # Wait for all scan callbacks to stop. There is no confirmation 137 # otherwise. 138 time.sleep(10) 139 return test_result 140 141 @BluetoothBaseTest.bt_test_wrap 142 def test_max_concurrent_ble_scans_then_discover_advertisement(self): 143 """Test max LE scans variant. 144 145 Test that a single device can have max scans concurrently scanning. 146 147 Steps: 148 1. Initialize scanner 149 2. Initialize advertiser 150 3. Create max ble scan callbacks 151 4. Start ble scan on each callback 152 5. Start advertising on the device from step 2 153 6. Verify that each callback triggers 154 7. Stop all scans and advertisements 155 156 Expected Result: 157 All scanning instances should start without errors and the advertisement 158 should be found on each scan instance. 159 160 Returns: 161 Pass if True 162 Fail if False 163 164 TAGS: LE, Scanning, Concurrency 165 Priority: 1 166 """ 167 self.adv_ad.droid.bleSetAdvertiseDataIncludeDeviceName(True) 168 self.scn_ad.droid.bleSetScanSettingsCallbackType( 169 ScanSettingsCallbackType.CALLBACK_TYPE_ALL_MATCHES.value) 170 self.scn_ad.droid.bleSetScanSettingsScanMode( 171 ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value) 172 self.adv_ad.droid.bleSetAdvertiseSettingsAdvertiseMode( 173 AdvertiseSettingsAdvertiseMode.ADVERTISE_MODE_LOW_LATENCY.value) 174 advertise_callback, advertise_data, advertise_settings = ( 175 generate_ble_advertise_objects(self.adv_ad.droid)) 176 filter_list = self.scn_ad.droid.bleGenFilterList() 177 self.scn_ad.droid.bleSetScanFilterDeviceName( 178 self.adv_ad.droid.bluetoothGetLocalName()) 179 self.scn_ad.droid.bleBuildScanFilter(filter_list) 180 scan_settings = self.scn_ad.droid.bleBuildScanSetting() 181 scan_callback_list = [] 182 for i in range(self.max_concurrent_scans): 183 self.log.debug("Concurrent Ble Scan iteration {}".format(i + 1)) 184 scan_callback = self.scn_ad.droid.bleGenScanCallback() 185 scan_callback_list.append(scan_callback) 186 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 187 scan_callback) 188 self.adv_ad.droid.bleStartBleAdvertising( 189 advertise_callback, advertise_data, advertise_settings) 190 try: 191 self.adv_ad.ed.pop_event( 192 adv_succ.format(advertise_callback), self.default_timeout) 193 except Empty as error: 194 self.log.exception("Test failed with Empty error: {}".format( 195 error)) 196 return False 197 except concurrent.futures._base.TimeoutError as error: 198 self.log.exception("Test failed, filtering callback onSuccess " 199 "never occurred: {}".format(error)) 200 return False 201 i = 0 202 for callback in scan_callback_list: 203 try: 204 self.scn_ad.ed.pop_event( 205 scan_result.format(scan_callback), self.default_timeout) 206 self.log.info( 207 "Found scan event successfully. Iteration {} successful." 208 .format(i)) 209 except Exception: 210 self.log.info("Failed to find a scan result for callback {}" 211 .format(scan_callback)) 212 return False 213 i += 1 214 for callback in scan_callback_list: 215 self.scn_ad.droid.bleStopBleScan(callback) 216 self.adv_ad.droid.bleStopBleAdvertising(advertise_callback) 217 return True 218 219 @BluetoothBaseTest.bt_test_wrap 220 def test_max_concurrent_ble_scans_plus_one(self): 221 """Test mac LE scans variant. 222 223 Test that a single device can have max scans concurrently scanning. 224 225 Steps: 226 1. Initialize scanner 227 3. Create max ble scan callbacks plus one 228 5. Start ble scan on each callback 229 6. Verify that the n+1th scan fails. 230 7. Stop all scans 231 232 Expected Result: 233 The n+1th scan should fail to start. 234 235 Returns: 236 Pass if True 237 Fail if False 238 239 TAGS: LE, Scanning, Concurrency 240 Priority: 1 241 """ 242 test_result = True 243 self.scn_ad.droid.bleSetScanSettingsCallbackType( 244 ScanSettingsCallbackType.CALLBACK_TYPE_ALL_MATCHES.value) 245 self.scn_ad.droid.bleSetScanSettingsScanMode( 246 ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value) 247 filter_list = self.scn_ad.droid.bleGenFilterList() 248 self.scn_ad.droid.bleBuildScanFilter(filter_list) 249 scan_settings = self.scn_ad.droid.bleBuildScanSetting() 250 scan_callback_list = [] 251 for i in range(self.max_concurrent_scans): 252 self.log.debug("Concurrent Ble Scan iteration {}".format(i + 1)) 253 scan_callback = self.scn_ad.droid.bleGenScanCallback() 254 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 255 scan_callback) 256 scan_callback_list.append(scan_callback) 257 scan_callback = self.scn_ad.droid.bleGenScanCallback() 258 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 259 scan_callback) 260 try: 261 self.scn_ad.ed.pop_event( 262 scan_failed.format(scan_callback), self.default_timeout) 263 self.log.info( 264 "Found scan event successfully. Iteration {} successful." 265 .format(i)) 266 except Exception: 267 self.log.info("Failed to find a onScanFailed event for callback {}" 268 .format(scan_callback)) 269 test_result = False 270 for callback in scan_callback_list: 271 self.scn_ad.droid.bleStopBleScan(callback) 272 return test_result 273 274 @BluetoothBaseTest.bt_test_wrap 275 def test_max_concurrent_ble_scans_verify_scans_stop_independently(self): 276 """Test max LE scans variant. 277 278 Test that a single device can have max scans concurrently scanning. 279 280 Steps: 281 1. Initialize scanner 282 2. Initialize advertiser 283 3. Create max ble scan callbacks 284 4. Start ble scan on each callback 285 5. Start advertising on the device from step 2 286 6. Verify that the first callback triggers 287 7. Stop the scan and repeat steps 6 and 7 until all scans stopped 288 289 Expected Result: 290 All scanning instances should start without errors and the advertisement 291 should be found on each scan instance. All scanning instances should 292 stop successfully. 293 294 Returns: 295 Pass if True 296 Fail if False 297 298 TAGS: LE, Scanning, Concurrency 299 Priority: 1 300 """ 301 self.adv_ad.droid.bleSetAdvertiseDataIncludeDeviceName(True) 302 self.scn_ad.droid.bleSetScanSettingsCallbackType( 303 ScanSettingsCallbackType.CALLBACK_TYPE_ALL_MATCHES.value) 304 self.scn_ad.droid.bleSetScanSettingsScanMode( 305 ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value) 306 self.adv_ad.droid.bleSetAdvertiseSettingsAdvertiseMode( 307 AdvertiseSettingsAdvertiseMode.ADVERTISE_MODE_LOW_LATENCY.value) 308 advertise_callback, advertise_data, advertise_settings = ( 309 generate_ble_advertise_objects(self.adv_ad.droid)) 310 filter_list = self.scn_ad.droid.bleGenFilterList() 311 self.scn_ad.droid.bleSetScanFilterDeviceName( 312 self.adv_ad.droid.bluetoothGetLocalName()) 313 self.scn_ad.droid.bleBuildScanFilter(filter_list) 314 scan_settings = self.scn_ad.droid.bleBuildScanSetting() 315 scan_callback_list = [] 316 for i in range(self.max_concurrent_scans): 317 self.log.debug("Concurrent Ble Scan iteration {}".format(i + 1)) 318 scan_callback = self.scn_ad.droid.bleGenScanCallback() 319 scan_callback_list.append(scan_callback) 320 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 321 scan_callback) 322 self.adv_ad.droid.bleStartBleAdvertising( 323 advertise_callback, advertise_data, advertise_settings) 324 try: 325 self.adv_ad.ed.pop_event( 326 adv_succ.format(advertise_callback), self.default_timeout) 327 except Empty as error: 328 self.log.exception("Test failed with Empty error: {}".format( 329 error)) 330 return False 331 except concurrent.futures._base.TimeoutError as error: 332 self.log.exception( 333 "Test failed, filtering callback onSuccess never" 334 " occurred: {}".format(error)) 335 return False 336 i = 0 337 for callback in scan_callback_list: 338 expected_scan_event_name = scan_result.format(scan_callback) 339 try: 340 self.scn_ad.ed.pop_event(expected_scan_event_name, 341 self.default_timeout) 342 self.log.info( 343 "Found scan event successfully. Iteration {} successful.".format( 344 i)) 345 i += 1 346 except Exception: 347 self.log.info( 348 "Failed to find a scan result for callback {}".format( 349 scan_callback)) 350 return False 351 self.scn_ad.droid.bleStopBleScan(callback) 352 self.adv_ad.droid.bleStopBleAdvertising(advertise_callback) 353 return True 354