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""" 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_decorators import test_tracker_info 26from acts_contrib.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest 27from acts_contrib.test_utils.bt.bt_constants import ble_advertise_settings_modes 28from acts_contrib.test_utils.bt.bt_constants import ble_scan_settings_callback_types 29from acts_contrib.test_utils.bt.bt_constants import ble_scan_settings_modes 30from acts_contrib.test_utils.bt.bt_constants import adv_succ 31from acts_contrib.test_utils.bt.bt_test_utils import generate_ble_advertise_objects 32from acts_contrib.test_utils.bt.bt_test_utils import reset_bluetooth 33from acts_contrib.test_utils.bt.bt_constants import scan_failed 34from acts_contrib.test_utils.bt.bt_constants import scan_result 35from acts_contrib.test_utils.bt.bt_test_utils import take_btsnoop_logs 36 37 38class ConcurrentBleScanningTest(BluetoothBaseTest): 39 default_timeout = 20 40 max_concurrent_scans = 27 41 42 def setup_class(self): 43 super().setup_class() 44 self.scn_ad = self.android_devices[0] 45 self.adv_ad = self.android_devices[1] 46 47 def on_fail(self, test_name, begin_time): 48 self.log.debug("Test {} failed. Gathering bugreport and btsnoop logs." 49 .format(test_name)) 50 take_btsnoop_logs(self.android_devices, self, test_name) 51 reset_bluetooth(self.android_devices) 52 53 def setup_test(self): 54 return reset_bluetooth(self.android_devices) 55 56 @BluetoothBaseTest.bt_test_wrap 57 @test_tracker_info(uuid='e7f68b9b-fb3f-48e9-a272-e41c2a32b4bd') 58 def test_max_concurrent_ble_scans(self): 59 """Test max LE scans. 60 61 Test that a single device can have max scans concurrently scanning. 62 63 Steps: 64 1. Initialize scanner 65 2. Initialize advertiser 66 3. Start advertising on the device from step 2 67 4. Create max ble scan callbacks 68 5. Start ble scan on each callback 69 6. Verify that each callback triggers 70 7. Stop all scans and advertisements 71 72 Expected Result: 73 All scanning instances should start without errors and the advertisement 74 should be found on each scan instance. 75 76 Returns: 77 Pass if True 78 Fail if False 79 80 TAGS: LE, Scanning, Concurrency 81 Priority: 0 82 """ 83 test_result = True 84 self.adv_ad.droid.bleSetAdvertiseDataIncludeDeviceName(True) 85 self.scn_ad.droid.bleSetScanSettingsCallbackType( 86 ble_scan_settings_callback_types['all_matches']) 87 self.scn_ad.droid.bleSetScanSettingsScanMode(ble_scan_settings_modes[ 88 'low_latency']) 89 self.adv_ad.droid.bleSetAdvertiseSettingsAdvertiseMode( 90 ble_advertise_settings_modes['low_latency']) 91 advertise_callback, advertise_data, advertise_settings = ( 92 generate_ble_advertise_objects(self.adv_ad.droid)) 93 self.adv_ad.droid.bleSetAdvertiseSettingsIsConnectable(False) 94 self.adv_ad.droid.bleStartBleAdvertising( 95 advertise_callback, advertise_data, advertise_settings) 96 try: 97 self.adv_ad.ed.pop_event( 98 adv_succ.format(advertise_callback), self.default_timeout) 99 except Empty as error: 100 self.log.exception("Test failed with Empty error: {}".format( 101 error)) 102 test_result = False 103 except concurrent.futures._base.TimeoutError as error: 104 self.log.exception( 105 "Test failed callback onSuccess never occurred: " 106 "{}".format(error)) 107 test_result = False 108 if not test_result: 109 return test_result 110 filter_list = self.scn_ad.droid.bleGenFilterList() 111 self.scn_ad.droid.bleSetScanFilterDeviceName( 112 self.adv_ad.droid.bluetoothGetLocalName()) 113 self.scn_ad.droid.bleBuildScanFilter(filter_list) 114 scan_settings = self.scn_ad.droid.bleBuildScanSetting() 115 scan_callback_list = [] 116 for i in range(self.max_concurrent_scans): 117 self.log.debug("Concurrent Ble Scan iteration {}".format(i + 1)) 118 scan_callback = self.scn_ad.droid.bleGenScanCallback() 119 scan_callback_list.append(scan_callback) 120 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 121 scan_callback) 122 try: 123 self.scn_ad.ed.pop_event( 124 scan_result.format(scan_callback), self.default_timeout) 125 self.log.info("Found scan event successfully. Iteration {} " 126 "successful.".format(i)) 127 except Exception: 128 self.log.info("Failed to find a scan result for callback {}" 129 .format(scan_callback)) 130 test_result = False 131 break 132 for callback in scan_callback_list: 133 self.scn_ad.droid.bleStopBleScan(callback) 134 self.adv_ad.droid.bleStopBleAdvertising(advertise_callback) 135 if not test_result: 136 return test_result 137 self.log.info("Waiting for scan callbacks to stop completely.") 138 # Wait for all scan callbacks to stop. There is no confirmation 139 # otherwise. 140 time.sleep(10) 141 return test_result 142 143 @BluetoothBaseTest.bt_test_wrap 144 @test_tracker_info(uuid='58b0c45e-1cbc-420a-9e89-901518ffe3d1') 145 def test_max_concurrent_ble_scans_then_discover_advertisement(self): 146 """Test max LE scans variant. 147 148 Test that a single device can have max scans concurrently scanning. 149 150 Steps: 151 1. Initialize scanner 152 2. Initialize advertiser 153 3. Create max ble scan callbacks 154 4. Start ble scan on each callback 155 5. Start advertising on the device from step 2 156 6. Verify that each callback triggers 157 7. Stop all scans and advertisements 158 159 Expected Result: 160 All scanning instances should start without errors and the advertisement 161 should be found on each scan instance. 162 163 Returns: 164 Pass if True 165 Fail if False 166 167 TAGS: LE, Scanning, Concurrency 168 Priority: 1 169 """ 170 self.adv_ad.droid.bleSetAdvertiseDataIncludeDeviceName(True) 171 self.scn_ad.droid.bleSetScanSettingsCallbackType( 172 ble_scan_settings_callback_types['all_matches']) 173 self.scn_ad.droid.bleSetScanSettingsScanMode(ble_scan_settings_modes[ 174 'low_latency']) 175 self.adv_ad.droid.bleSetAdvertiseSettingsAdvertiseMode( 176 ble_advertise_settings_modes['low_latency']) 177 advertise_callback, advertise_data, advertise_settings = ( 178 generate_ble_advertise_objects(self.adv_ad.droid)) 179 filter_list = self.scn_ad.droid.bleGenFilterList() 180 self.scn_ad.droid.bleSetScanFilterDeviceName( 181 self.adv_ad.droid.bluetoothGetLocalName()) 182 self.scn_ad.droid.bleBuildScanFilter(filter_list) 183 scan_settings = self.scn_ad.droid.bleBuildScanSetting() 184 scan_callback_list = [] 185 for i in range(self.max_concurrent_scans): 186 self.log.debug("Concurrent Ble Scan iteration {}".format(i + 1)) 187 scan_callback = self.scn_ad.droid.bleGenScanCallback() 188 scan_callback_list.append(scan_callback) 189 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 190 scan_callback) 191 self.adv_ad.droid.bleStartBleAdvertising( 192 advertise_callback, advertise_data, advertise_settings) 193 try: 194 self.adv_ad.ed.pop_event( 195 adv_succ.format(advertise_callback), self.default_timeout) 196 except Empty as error: 197 self.log.exception("Test failed with Empty error: {}".format( 198 error)) 199 return False 200 except concurrent.futures._base.TimeoutError as error: 201 self.log.exception("Test failed, filtering callback onSuccess " 202 "never occurred: {}".format(error)) 203 return False 204 i = 0 205 for callback in scan_callback_list: 206 try: 207 self.scn_ad.ed.pop_event( 208 scan_result.format(callback), self.default_timeout) 209 self.log.info( 210 "Found scan event successfully. Iteration {} successful." 211 .format(i)) 212 except Exception: 213 self.log.info("Failed to find a scan result for callback {}" 214 .format(callback)) 215 return False 216 i += 1 217 for callback in scan_callback_list: 218 self.scn_ad.droid.bleStopBleScan(callback) 219 self.adv_ad.droid.bleStopBleAdvertising(advertise_callback) 220 return True 221 222 @BluetoothBaseTest.bt_test_wrap 223 @test_tracker_info(uuid='7a45e45c-faf3-4e89-abb7-a52f63e53208') 224 def test_max_concurrent_ble_scans_plus_one(self): 225 """Test mac LE scans variant. 226 227 Test that a single device can have max scans concurrently scanning. 228 229 Steps: 230 1. Initialize scanner 231 3. Create max ble scan callbacks plus one 232 5. Start ble scan on each callback 233 6. Verify that the n+1th scan fails. 234 7. Stop all scans 235 236 Expected Result: 237 The n+1th scan should fail to start. 238 239 Returns: 240 Pass if True 241 Fail if False 242 243 TAGS: LE, Scanning, Concurrency 244 Priority: 1 245 """ 246 test_result = True 247 filter_list = self.scn_ad.droid.bleGenFilterList() 248 scan_settings = self.scn_ad.droid.bleBuildScanSetting() 249 scan_callback_list = [] 250 for i in range(self.max_concurrent_scans): 251 self.log.debug("Concurrent Ble Scan iteration {}".format(i + 1)) 252 scan_callback = self.scn_ad.droid.bleGenScanCallback() 253 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 254 scan_callback) 255 scan_callback_list.append(scan_callback) 256 scan_callback = self.scn_ad.droid.bleGenScanCallback() 257 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 258 scan_callback) 259 try: 260 self.scn_ad.ed.pop_event( 261 scan_failed.format(scan_callback), self.default_timeout) 262 self.log.info( 263 "Found scan event successfully. Iteration {} successful." 264 .format(self.max_concurrent_scans + 1)) 265 except Exception: 266 self.log.error("Failed to find a onScanFailed event for callback {}" 267 .format(scan_callback)) 268 test_result = False 269 for callback in scan_callback_list: 270 self.scn_ad.droid.bleStopBleScan(callback) 271 return test_result 272 273 @BluetoothBaseTest.bt_test_wrap 274 @test_tracker_info(uuid='5a91f612-69e5-490f-b9d0-50d58a3db736') 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 ble_scan_settings_callback_types['all_matches']) 304 self.scn_ad.droid.bleSetScanSettingsScanMode(ble_scan_settings_modes[ 305 'low_latency']) 306 self.adv_ad.droid.bleSetAdvertiseSettingsAdvertiseMode( 307 ble_advertise_settings_modes['low_latency']) 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.". 344 format(i)) 345 i += 1 346 except Exception: 347 self.log.info("Failed to find a scan result for callback {}". 348 format(scan_callback)) 349 return False 350 self.scn_ad.droid.bleStopBleScan(callback) 351 self.adv_ad.droid.bleStopBleAdvertising(advertise_callback) 352 return True 353