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 """ 17 OnLost onFound Stress Test. 18 """ 19 20 import threading 21 import time 22 23 from queue import Empty 24 from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest 25 from acts.test_utils.bt.BleEnum import AdvertiseSettingsAdvertiseMode 26 from acts.test_utils.bt.BleEnum import ScanSettingsCallbackType 27 from acts.test_utils.bt.BleEnum import ScanSettingsScanMode 28 from acts.test_utils.bt.BleEnum import ScanSettingsMatchMode 29 from acts.test_utils.bt.BleEnum import ScanSettingsMatchNum 30 from acts.test_utils.bt.bt_test_utils import cleanup_scanners_and_advertisers 31 from acts.test_utils.bt.bt_test_utils import get_advanced_droid_list 32 from acts.test_utils.bt.bt_gatt_utils import orchestrate_gatt_connection 33 from acts.test_utils.bt.bt_test_utils import reset_bluetooth 34 from acts.test_utils.bt.bt_gatt_utils import run_continuous_write_descriptor 35 from acts.test_utils.bt.bt_gatt_utils import setup_multiple_services 36 37 38 class BleOnLostOnFoundStressTest(BluetoothBaseTest): 39 default_timeout = 10 40 max_scan_instances = 28 41 report_delay = 2000 42 active_scan_callback_list = [] 43 active_adv_callback_list = [] 44 scan_result = "BleScan{}onScanResults" 45 batch_scan_result = "BleScan{}onBatchScanResult" 46 47 def __init__(self, controllers): 48 BluetoothBaseTest.__init__(self, controllers) 49 self.droid_list = get_advanced_droid_list(self.android_devices) 50 self.scn_ad = self.android_devices[0] 51 self.adv_ad = self.android_devices[1] 52 if self.droid_list[1]['max_advertisements'] == 0: 53 self.tests = () 54 return 55 56 def teardown_test(self): 57 cleanup_scanners_and_advertisers( 58 self.scn_ad, self.active_adv_callback_list, self.scn_ad, 59 self.active_adv_callback_list) 60 self.active_adv_callback_list = [] 61 self.active_scan_callback_list = [] 62 63 def on_exception(self, test_name, begin_time): 64 reset_bluetooth(self.android_devices) 65 66 def _start_generic_advertisement_include_device_name(self): 67 self.adv_ad.droid.bleSetAdvertiseDataIncludeDeviceName(True) 68 self.adv_ad.droid.bleSetAdvertiseSettingsAdvertiseMode( 69 AdvertiseSettingsAdvertiseMode.ADVERTISE_MODE_LOW_LATENCY.value) 70 advertise_data = self.adv_ad.droid.bleBuildAdvertiseData() 71 advertise_settings = self.adv_ad.droid.bleBuildAdvertiseSettings() 72 advertise_callback = self.adv_ad.droid.bleGenBleAdvertiseCallback() 73 self.adv_ad.droid.bleStartBleAdvertising( 74 advertise_callback, advertise_data, advertise_settings) 75 self.adv_ad.ed.pop_event( 76 "BleAdvertise{}onSuccess".format(advertise_callback), 77 self.default_timeout) 78 self.active_adv_callback_list.append(advertise_callback) 79 return advertise_callback 80 81 def _verify_no_events_found(self, event_name): 82 try: 83 self.scn_ad.ed.pop_event(event_name, self.default_timeout) 84 self.log.error("Found an event when none was expected.") 85 return False 86 except Empty: 87 self.log.info("No scan result found as expected.") 88 return True 89 90 def _poll_energy(self): 91 import random 92 while True: 93 self.log.debug( 94 self.scn_ad.droid.bluetoothGetControllerActivityEnergyInfo(1)) 95 time.sleep(2) 96 97 @BluetoothBaseTest.bt_test_wrap 98 def test_on_star_while_polling_energy_stats(self): 99 """ 100 Tests ... 101 Steps 102 1: ... 103 :return: boolean 104 """ 105 thread = threading.Thread(target=self._poll_energy) 106 thread.start() 107 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.bleSetScanSettingsScanMode( 112 ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value) 113 self.scn_ad.droid.bleSetScanSettingsCallbackType( 114 ScanSettingsCallbackType.CALLBACK_TYPE_FOUND_AND_LOST.value) 115 self.scn_ad.droid.bleSetScanSettingsMatchMode( 116 ScanSettingsMatchMode.AGGRESIVE.value) 117 self.scn_ad.droid.bleSetScanSettingsNumOfMatches( 118 ScanSettingsMatchNum.MATCH_NUM_ONE_ADVERTISEMENT.value) 119 scan_settings = self.scn_ad.droid.bleBuildScanSetting() 120 scan_callback = self.scn_ad.droid.bleGenScanCallback() 121 self.scn_ad.droid.bleBuildScanFilter(filter_list) 122 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 123 scan_callback) 124 self.active_scan_callback_list.append(scan_callback) 125 on_found_count = 0 126 on_lost_count = 0 127 from contextlib import suppress 128 for x in range(1000): 129 adv_callback = ( 130 self._start_generic_advertisement_include_device_name()) 131 with suppress(Exception): 132 event = self.scn_ad.ed.pop_event( 133 self.scan_result.format(scan_callback), 134 self.default_timeout * 3) 135 if event['data']['CallbackType'] == 2: 136 on_found_count += 1 137 elif event['data']['CallbackType'] == 4: 138 on_lost_count += 1 139 self.adv_ad.droid.bleStopBleAdvertising(adv_callback) 140 with suppress(Exception): 141 event2 = self.scn_ad.ed.pop_event( 142 self.scan_result.format(scan_callback), 143 self.default_timeout * 4) 144 if event2['data']['CallbackType'] == 2: 145 on_found_count += 1 146 elif event2['data']['CallbackType'] == 4: 147 on_lost_count += 1 148 thread.join() 149 return True 150 151 @BluetoothBaseTest.bt_test_wrap 152 def test_more_stress_test(self): 153 gatt_server_callback, gatt_server = setup_multiple_services( 154 self.adv_ad) 155 bluetooth_gatt, gatt_callback, adv_callback = ( 156 orchestrate_gatt_connection(self.scn_ad, self.adv_ad)) 157 self.active_scan_callback_list.append(adv_callback) 158 if self.scn_ad.droid.gattClientDiscoverServices(bluetooth_gatt): 159 event = self.scn_ad.ed.pop_event( 160 "GattConnect{}onServicesDiscovered".format(bluetooth_gatt), 161 self.default_timeout) 162 discovered_services_index = event['data']['ServicesIndex'] 163 else: 164 self.log.info("Failed to discover services.") 165 return False 166 services_count = self.scn_ad.droid.gattClientGetDiscoveredServicesCount( 167 discovered_services_index) 168 thread = threading.Thread( 169 target=run_continuous_write_descriptor, 170 args=(self.scn_ad.droid, self.scn_ad.ed, self.adv_ad.droid, 171 self.adv_ad.ed, gatt_server, gatt_server_callback, 172 bluetooth_gatt, services_count, discovered_services_index)) 173 thread.start() 174 thread2 = threading.Thread(target=self._poll_energy) 175 thread2.start() 176 177 filter_list = self.scn_ad.droid.bleGenFilterList() 178 self.scn_ad.droid.bleSetScanFilterDeviceName( 179 self.adv_ad.droid.bluetoothGetLocalName()) 180 self.scn_ad.droid.bleSetScanSettingsScanMode( 181 ScanSettingsScanMode.SCAN_MODE_LOW_LATENCY.value) 182 self.scn_ad.droid.bleSetScanSettingsCallbackType( 183 ScanSettingsCallbackType.CALLBACK_TYPE_FOUND_AND_LOST.value) 184 self.scn_ad.droid.bleSetScanSettingsMatchMode( 185 ScanSettingsMatchMode.AGGRESIVE.value) 186 self.scn_ad.droid.bleSetScanSettingsNumOfMatches( 187 ScanSettingsMatchNum.MATCH_NUM_ONE_ADVERTISEMENT.value) 188 scan_settings = self.scn_ad.droid.bleBuildScanSetting() 189 scan_callback = self.scn_ad.droid.bleGenScanCallback() 190 self.scn_ad.droid.bleBuildScanFilter(filter_list) 191 self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings, 192 scan_callback) 193 self.active_scan_callback_list.append(scan_callback) 194 on_found_count = 0 195 on_lost_count = 0 196 time.sleep(60) 197 from contextlib import suppress 198 for x in range(1000): 199 adv_callback = self._start_generic_advertisement_include_device_name( 200 ) 201 with suppress(Exception): 202 event = self.scn_ad.ed.pop_event( 203 self.scan_result.format(scan_callback), 204 self.default_timeout * 3) 205 if event['data']['CallbackType'] == 2: 206 on_found_count += 1 207 elif event['data']['CallbackType'] == 4: 208 on_lost_count += 1 209 self.adv_ad.droid.bleStopBleAdvertising(adv_callback) 210 with suppress(Exception): 211 event2 = self.scn_ad.ed.pop_event( 212 self.scan_result.format(scan_callback), 213 self.default_timeout * 4) 214 if event2['data']['CallbackType'] == 2: 215 on_found_count += 1 216 elif event2['data']['CallbackType'] == 4: 217 on_lost_count += 1 218 thread.join() 219 thread2.join() 220 return True 221
