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""" 17This test script exercises different testcases with a lot of ble beacon traffic. 18 19This test script was designed with this setup in mind: 20Shield box one: Android Device as DUT. 7x Sprout devices acting as 192 beacons 21""" 22 23import threading 24 25from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest 26from acts.test_utils.bt.BleEnum import AdvertiseSettingsAdvertiseMode 27from acts.test_utils.bt.BleEnum import ScanSettingsScanMode 28from acts.test_utils.bt.bt_test_utils import adv_succ 29from acts.test_utils.bt.bt_test_utils import batch_scan_result 30from acts.test_utils.bt.bt_test_utils import scan_result 31from acts.test_utils.bt.bt_test_utils import generate_ble_advertise_objects 32from acts.test_utils.bt.bt_test_utils import generate_ble_scan_objects 33from acts.test_utils.bt.bt_test_utils import log_energy_info 34from acts.test_utils.bt.bt_test_utils import reset_bluetooth 35from acts.test_utils.bt.bt_test_utils import setup_multiple_devices_for_bt_test 36from acts.test_utils.bt.bt_test_utils import take_btsnoop_logs 37 38 39class BeaconSwarmTest(BluetoothBaseTest): 40 default_timeout = 10 41 beacon_swarm_count = 0 42 advertising_device_name_list = [] 43 discovered_mac_address_list = [] 44 45 def __init__(self, controllers): 46 BluetoothBaseTest.__init__(self, controllers) 47 self.scn_ad = self.android_devices[0] 48 49 def setup_test(self): 50 self.log.debug(log_energy_info(self.android_devices, "Start")) 51 self.discovered_mac_address_list = [] 52 for a in self.android_devices: 53 a.ed.clear_all_events() 54 return True 55 56 def teardown_test(self): 57 self.log.debug(log_energy_info(self.android_devices, "End")) 58 reset_bluetooth([self.android_devices[0]]) 59 return True 60 61 def setup_class(self): 62 if not setup_multiple_devices_for_bt_test(self.android_devices): 63 return False 64 return self._start_special_advertisements() 65 66 def cleanup_class(self): 67 return reset_bluetooth(self.android_devices) 68 69 def on_fail(self, test_name, begin_time): 70 take_btsnoop_logs(self.android_devices, self, test_name) 71 reset_bluetooth([self.scn_ad]) 72 73 def _start_advertisements_thread(self, ad, beacon_count, restart=False): 74 d, e = ad.droid, ad.ed 75 if restart: 76 try: 77 reset_bluetooth([ad]) 78 except Exception: 79 self.log.debug("Failed resetting Bluetooth, continuing...") 80 return 81 try: 82 for _ in range(beacon_count): 83 d.bleSetAdvertiseDataIncludeDeviceName(True) 84 d.bleSetAdvertiseSettingsAdvertiseMode( 85 AdvertiseSettingsAdvertiseMode.ADVERTISE_MODE_LOW_LATENCY. 86 value) 87 advertise_callback, advertise_data, advertise_settings = ( 88 generate_ble_advertise_objects(d)) 89 d.bleStartBleAdvertising(advertise_callback, advertise_data, 90 advertise_settings) 91 try: 92 e.pop_event( 93 adv_succ.format(advertise_callback), 94 self.default_timeout) 95 self.beacon_swarm_count += 1 96 local_bt_name = d.bluetoothGetLocalName() 97 if local_bt_name not in self.advertising_device_name_list: 98 self.advertising_device_name_list.append( 99 d.bluetoothGetLocalName()) 100 except Exception as e: 101 self.log.info("Advertising failed due to " + str(e)) 102 self.log.info("Beacons active: {}".format( 103 self.beacon_swarm_count)) 104 except Exception: 105 self.log.debug( 106 "Something went wrong in starting advertisements, continuing.") 107 return 108 109 def _start_special_advertisements(self): 110 self.log.info("Setting up advertisements.") 111 beacon_serials = [] 112 beacon_count = 0 113 try: 114 beacon_serials = self.user_params['beacon_devices'] 115 beacon_count = self.user_params['beacon_count'] 116 except AttributeError: 117 self.log.info( 118 "No controllable devices connected to create beacons with." 119 " Continuing...") 120 threads = [] 121 for a in self.android_devices: 122 d, e = a.droid, a.ed 123 serial_no = d.getBuildSerial() 124 if serial_no not in beacon_serials: 125 continue 126 thread = threading.Thread(target=self._start_advertisements_thread, 127 args=(d, e, beacon_count)) 128 threads.append(thread) 129 thread.start() 130 for t in threads: 131 t.join() 132 if self.beacon_swarm_count < (beacon_count * len(beacon_serials)): 133 self.log.error("Not enough beacons advertising: {}".format( 134 self.beacon_swarm_count)) 135 return False 136 return True 137 138 def _restart_special_advertisements_thread(self): 139 beacon_serials = [] 140 beacon_count = 0 141 try: 142 beacon_serials = self.user_params['beacon_devices'] 143 beacon_count = self.user_params['beacon_count'] 144 except AttributeError: 145 self.log.info("No controllable devices connected to create beacons" 146 " with. Continuing...") 147 threads = [] 148 while True: 149 self.log.info("Restarting advertisements.") 150 for a in self.android_devices: 151 d, e = a.droid, a.ed 152 serial_no = d.getBuildSerial() 153 if serial_no not in beacon_serials: 154 continue 155 thread = threading.Thread( 156 target=self._start_advertisements_thread, 157 args=(d, e, beacon_count, True)) 158 threads.append(thread) 159 thread.start() 160 for t in threads: 161 t.join() 162 return True 163 164 def test_swarm_1000_on_scan_result(self): 165 """Test LE scanning in a mass beacon deployment. 166 167 Test finding 1000 LE scan results in a mass beacon deployment. 168 169 Steps: 170 1. Assume that mass beacon deployment is setup. 171 2. Set LE scanning mode to low latency. 172 3. Start LE scan. 173 4. Pop scan results off the event dispatcher 1000 times. 174 5. Stop LE scanning. 175 176 Expected Result: 177 1000 scan results should be found without any exceptions. 178 179 Returns: 180 Pass if True 181 Fail if False 182 183 TAGS: LE, Scanning, Beacon 184 Priority: 1 185 """ 186 self.scn_ad.droid.bleSetScanSettingsScanMode( 187 ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value) 188 filter_list, scan_settings, scan_callback = generate_ble_scan_objects( 189 self.scn_ad.droid) 190 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 191 scan_callback) 192 for _ in range(1000000): 193 event_info = self.scn_ad.ed.pop_event( 194 scan_result.format(scan_callback), self.default_timeout) 195 mac_address = event_info['data']['Result']['deviceInfo']['address'] 196 if mac_address not in self.discovered_mac_address_list: 197 self.discovered_mac_address_list.append(mac_address) 198 self.log.info("Discovered {} different devices.".format(len( 199 self.discovered_mac_address_list))) 200 self.log.debug("Discovered {} different devices.".format(len( 201 self.discovered_mac_address_list))) 202 self.scn_ad.droid.bleStopBleScan(scan_callback) 203 return True 204 205 def test_swarm_10000_on_batch_scan_result(self): 206 """Test LE batch scanning in a mass beacon deployment. 207 208 Test finding 10000 LE batch scan results in a mass beacon deployment. 209 210 Steps: 211 1. Assume that mass beacon deployment is setup. 212 2. Set LE scanning mode to low latency and report delay millis to 1 213 second. 214 3. Start LE scan. 215 4. Pop batch scan results off the event dispatcher 10000 times. 216 5. Stop LE scanning. 217 218 Expected Result: 219 1000 scan results should be found without any exceptions. 220 221 Returns: 222 Pass if True 223 Fail if False 224 225 TAGS: LE, Scanning, Beacon 226 Priority: 1 227 """ 228 self.scn_ad.droid.bleSetScanSettingsScanMode( 229 ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value) 230 self.scn_ad.droid.bleSetScanSettingsReportDelayMillis(1000) 231 filter_list, scan_settings, scan_callback = generate_ble_scan_objects( 232 self.scn_ad.droid) 233 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 234 scan_callback) 235 for _ in range(10000): 236 event_info = self.scn_ad.ed.pop_event( 237 batch_scan_result.format(scan_callback), self.default_timeout) 238 for result in event_info['data']['Results']: 239 mac_address = result['deviceInfo']['address'] 240 if mac_address not in self.discovered_mac_address_list: 241 self.discovered_mac_address_list.append(mac_address) 242 self.log.info("Discovered {} different devices.".format(len( 243 self.discovered_mac_address_list))) 244 self.scn_ad.droid.bleStopBleScan(scan_callback) 245 return True 246 247 def test_swarm_scan_result_filter_each_device_name(self): 248 """Test basic LE scan filtering in a mass beacon deployment. 249 250 Test finding LE scan results in a mass beacon deployment. This 251 test specifically tests scan filtering of different device names and 252 that each device name is found. 253 254 Steps: 255 1. Assume that mass beacon deployment is setup with device names 256 advertising. 257 2. Set LE scanning mode to low latency. 258 3. Filter device name from one of the known advertising device names 259 4. Start LE scan. 260 5. Pop scan results matching the scan filter. 261 6. Stop LE scanning. 262 7. Repeat steps 2-6 until all advertising device names are found. 263 264 Expected Result: 265 All advertising beacons are found by their device name. 266 267 Returns: 268 Pass if True 269 Fail if False 270 271 TAGS: LE, Scanning, Beacon, Filtering 272 Priority: 1 273 """ 274 for filter_name in self.advertising_device_name_list: 275 self.scn_ad.droid.bleSetScanSettingsScanMode( 276 ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value) 277 filter_list, scan_settings, scan_callback = ( 278 generate_ble_scan_objects(self.scn_ad.droid)) 279 try: 280 self.scn_ad.droid.bleSetScanFilterDeviceName(filter_name) 281 self.scn_ad.droid.bleBuildScanFilter(filter_list) 282 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 283 scan_callback) 284 self.log.debug(self.scn_ad.ed.pop_event( 285 scan_result.format(scan_callback), self.default_timeout)) 286 except Exception: 287 self.log.info("Couldn't find advertiser name {}.".format( 288 filter_name)) 289 return False 290 self.scn_ad.droid.bleStopBleScan(scan_callback) 291 return True 292 293 def test_swarm_rotate_addresses(self): 294 """Test basic LE scan filtering in a mass beacon deployment. 295 296 Test finding LE scan results in a mass beacon deployment. This test 297 rotates the mac address of the advertising devices at a consistent 298 interval in order to make the scanning device think there are 299 thousands of devices nearby. 300 301 Steps: 302 1. Assume that mass beacon deployment is setup with device names 303 advertising. 304 2. Set LE scanning mode to low latency on 28 scan instances. 305 3. Start LE scan on each of the scan instances. 306 5. Continuously Pop scan results matching the scan filter. 307 6. Rotate mac address of each advertising device sequentially. 308 7. 5-6 10,000 times. 309 8. Stop LE scanning 310 311 Expected Result: 312 The Bluetooth stack doesn't crash. 313 314 Returns: 315 Pass if True 316 Fail if False 317 318 TAGS: LE, Scanning, Beacon 319 Priority: 1 320 """ 321 scan_callback_list = [] 322 for _ in range(28): 323 self.scn_ad.droid.bleSetScanSettingsScanMode( 324 ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value) 325 filter_list, scan_settings, scan_callback = generate_ble_scan_objects( 326 self.scn_ad.droid) 327 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 328 scan_callback) 329 scan_callback_list.append(scan_callback) 330 thread = threading.Thread( 331 target=self._restart_special_advertisements_thread, 332 args=()) 333 thread.start() 334 n = 0 335 while n < 10000: 336 for cb in scan_callback_list: 337 event_info = self.scn_ad.ed.pop_event( 338 scan_result.format(cb), self.default_timeout) 339 mac_address = event_info['data']['Result']['deviceInfo'][ 340 'address'] 341 if mac_address not in self.discovered_mac_address_list: 342 self.discovered_mac_address_list.append(mac_address) 343 self.log.info("Discovered {} different devices.".format(len( 344 self.discovered_mac_address_list))) 345 n += 1 346 self.scn_ad.droid.bleStopBleScan(scan_callback) 347 return True 348