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