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