#!/usr/bin/env python3 # # Copyright (C) 2016 The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); you may not # use this file except in compliance with the License. You may obtain a copy of # the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations under # the License. """ This test script exercises different testcases with a lot of ble beacon traffic. This test script was designed with this setup in mind: Shield box one: Android Device as DUT. 7x Sprout devices acting as 192 beacons """ import threading from acts_contrib.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest from acts_contrib.test_utils.bt.BleEnum import AdvertiseSettingsAdvertiseMode from acts_contrib.test_utils.bt.BleEnum import ScanSettingsScanMode from acts_contrib.test_utils.bt.bt_test_utils import adv_succ from acts_contrib.test_utils.bt.bt_test_utils import batch_scan_result from acts_contrib.test_utils.bt.bt_test_utils import scan_result from acts_contrib.test_utils.bt.bt_test_utils import generate_ble_advertise_objects from acts_contrib.test_utils.bt.bt_test_utils import generate_ble_scan_objects from acts_contrib.test_utils.bt.bt_test_utils import reset_bluetooth from acts_contrib.test_utils.bt.bt_test_utils import setup_multiple_devices_for_bt_test from acts_contrib.test_utils.bt.bt_test_utils import take_btsnoop_logs class BeaconSwarmTest(BluetoothBaseTest): default_timeout = 10 beacon_swarm_count = 0 advertising_device_name_list = [] discovered_mac_address_list = [] def setup_test(self): self.discovered_mac_address_list = [] for a in self.android_devices: a.ed.clear_all_events() return True def teardown_test(self): reset_bluetooth([self.android_devices[0]]) return True def setup_class(self): self.scn_ad = self.android_devices[0] if not setup_multiple_devices_for_bt_test(self.android_devices): return False return self._start_special_advertisements() def cleanup_class(self): return reset_bluetooth(self.android_devices) def on_fail(self, test_name, begin_time): take_btsnoop_logs(self.android_devices, self, test_name) reset_bluetooth([self.scn_ad]) def _start_advertisements_thread(self, ad, beacon_count, restart=False): d, e = ad.droid, ad.ed if restart: try: reset_bluetooth([ad]) except Exception: self.log.debug("Failed resetting Bluetooth, continuing...") return try: for _ in range(beacon_count): d.bleSetAdvertiseDataIncludeDeviceName(True) d.bleSetAdvertiseSettingsAdvertiseMode( AdvertiseSettingsAdvertiseMode.ADVERTISE_MODE_LOW_LATENCY. value) advertise_callback, advertise_data, advertise_settings = ( generate_ble_advertise_objects(d)) d.bleStartBleAdvertising(advertise_callback, advertise_data, advertise_settings) try: e.pop_event( adv_succ.format(advertise_callback), self.default_timeout) self.beacon_swarm_count += 1 local_bt_name = d.bluetoothGetLocalName() if local_bt_name not in self.advertising_device_name_list: self.advertising_device_name_list.append( d.bluetoothGetLocalName()) except Exception as e: self.log.info("Advertising failed due to " + str(e)) self.log.info("Beacons active: {}".format( self.beacon_swarm_count)) except Exception: self.log.debug( "Something went wrong in starting advertisements, continuing.") return def _start_special_advertisements(self): self.log.info("Setting up advertisements.") beacon_serials = [] beacon_count = 0 try: beacon_serials = self.user_params['beacon_devices'] beacon_count = self.user_params['beacon_count'] except AttributeError: self.log.info( "No controllable devices connected to create beacons with." " Continuing...") threads = [] for a in self.android_devices: d, e = a.droid, a.ed serial_no = a.serial if serial_no not in beacon_serials: continue thread = threading.Thread(target=self._start_advertisements_thread, args=(d, e, beacon_count)) threads.append(thread) thread.start() for t in threads: t.join() if self.beacon_swarm_count < (beacon_count * len(beacon_serials)): self.log.error("Not enough beacons advertising: {}".format( self.beacon_swarm_count)) return False return True def _restart_special_advertisements_thread(self): beacon_serials = [] beacon_count = 0 try: beacon_serials = self.user_params['beacon_devices'] beacon_count = self.user_params['beacon_count'] except AttributeError: self.log.info("No controllable devices connected to create beacons" " with. Continuing...") threads = [] while True: self.log.info("Restarting advertisements.") for a in self.android_devices: d, e = a.droid, a.ed serial_no = a.serial if serial_no not in beacon_serials: continue thread = threading.Thread( target=self._start_advertisements_thread, args=(d, e, beacon_count, True)) threads.append(thread) thread.start() for t in threads: t.join() return True def test_swarm_1000_on_scan_result(self): """Test LE scanning in a mass beacon deployment. Test finding 1000 LE scan results in a mass beacon deployment. Steps: 1. Assume that mass beacon deployment is setup. 2. Set LE scanning mode to low latency. 3. Start LE scan. 4. Pop scan results off the event dispatcher 1000 times. 5. Stop LE scanning. Expected Result: 1000 scan results should be found without any exceptions. Returns: Pass if True Fail if False TAGS: LE, Scanning, Beacon Priority: 1 """ self.scn_ad.droid.bleSetScanSettingsScanMode( ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value) filter_list, scan_settings, scan_callback = generate_ble_scan_objects( self.scn_ad.droid) self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, scan_callback) for _ in range(1000000): event_info = self.scn_ad.ed.pop_event( scan_result.format(scan_callback), self.default_timeout) mac_address = event_info['data']['Result']['deviceInfo']['address'] if mac_address not in self.discovered_mac_address_list: self.discovered_mac_address_list.append(mac_address) self.log.info("Discovered {} different devices.".format(len( self.discovered_mac_address_list))) self.log.debug("Discovered {} different devices.".format(len( self.discovered_mac_address_list))) self.scn_ad.droid.bleStopBleScan(scan_callback) return True def test_swarm_10000_on_batch_scan_result(self): """Test LE batch scanning in a mass beacon deployment. Test finding 10000 LE batch scan results in a mass beacon deployment. Steps: 1. Assume that mass beacon deployment is setup. 2. Set LE scanning mode to low latency and report delay millis to 1 second. 3. Start LE scan. 4. Pop batch scan results off the event dispatcher 10000 times. 5. Stop LE scanning. Expected Result: 1000 scan results should be found without any exceptions. Returns: Pass if True Fail if False TAGS: LE, Scanning, Beacon Priority: 1 """ self.scn_ad.droid.bleSetScanSettingsScanMode( ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value) self.scn_ad.droid.bleSetScanSettingsReportDelayMillis(1000) filter_list, scan_settings, scan_callback = generate_ble_scan_objects( self.scn_ad.droid) self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, scan_callback) for _ in range(10000): event_info = self.scn_ad.ed.pop_event( batch_scan_result.format(scan_callback), self.default_timeout) for result in event_info['data']['Results']: mac_address = result['deviceInfo']['address'] if mac_address not in self.discovered_mac_address_list: self.discovered_mac_address_list.append(mac_address) self.log.info("Discovered {} different devices.".format(len( self.discovered_mac_address_list))) self.scn_ad.droid.bleStopBleScan(scan_callback) return True def test_swarm_scan_result_filter_each_device_name(self): """Test basic LE scan filtering in a mass beacon deployment. Test finding LE scan results in a mass beacon deployment. This test specifically tests scan filtering of different device names and that each device name is found. Steps: 1. Assume that mass beacon deployment is setup with device names advertising. 2. Set LE scanning mode to low latency. 3. Filter device name from one of the known advertising device names 4. Start LE scan. 5. Pop scan results matching the scan filter. 6. Stop LE scanning. 7. Repeat steps 2-6 until all advertising device names are found. Expected Result: All advertising beacons are found by their device name. Returns: Pass if True Fail if False TAGS: LE, Scanning, Beacon, Filtering Priority: 1 """ for filter_name in self.advertising_device_name_list: self.scn_ad.droid.bleSetScanSettingsScanMode( ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value) filter_list, scan_settings, scan_callback = ( generate_ble_scan_objects(self.scn_ad.droid)) try: self.scn_ad.droid.bleSetScanFilterDeviceName(filter_name) self.scn_ad.droid.bleBuildScanFilter(filter_list) self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, scan_callback) self.log.debug(self.scn_ad.ed.pop_event( scan_result.format(scan_callback), self.default_timeout)) except Exception: self.log.info("Couldn't find advertiser name {}.".format( filter_name)) return False self.scn_ad.droid.bleStopBleScan(scan_callback) return True def test_swarm_rotate_addresses(self): """Test basic LE scan filtering in a mass beacon deployment. Test finding LE scan results in a mass beacon deployment. This test rotates the mac address of the advertising devices at a consistent interval in order to make the scanning device think there are thousands of devices nearby. Steps: 1. Assume that mass beacon deployment is setup with device names advertising. 2. Set LE scanning mode to low latency on 28 scan instances. 3. Start LE scan on each of the scan instances. 5. Continuously Pop scan results matching the scan filter. 6. Rotate mac address of each advertising device sequentially. 7. 5-6 10,000 times. 8. Stop LE scanning Expected Result: The Bluetooth stack doesn't crash. Returns: Pass if True Fail if False TAGS: LE, Scanning, Beacon Priority: 1 """ scan_callback_list = [] for _ in range(28): self.scn_ad.droid.bleSetScanSettingsScanMode( ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value) filter_list, scan_settings, scan_callback = generate_ble_scan_objects( self.scn_ad.droid) self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, scan_callback) scan_callback_list.append(scan_callback) thread = threading.Thread( target=self._restart_special_advertisements_thread, args=()) thread.start() n = 0 while n < 10000: for cb in scan_callback_list: event_info = self.scn_ad.ed.pop_event( scan_result.format(cb), self.default_timeout) mac_address = event_info['data']['Result']['deviceInfo'][ 'address'] if mac_address not in self.discovered_mac_address_list: self.discovered_mac_address_list.append(mac_address) self.log.info("Discovered {} different devices.".format(len( self.discovered_mac_address_list))) n += 1 self.scn_ad.droid.bleStopBleScan(scan_callback) return True