Ang Li | 73697b3 | 2015-12-03 00:41:53 +0000 | [diff] [blame] | 1 | # python3.4 |
| 2 | # Copyright (C) 2014 The Android Open Source Project |
| 3 | # |
| 4 | # Licensed under the Apache License, Version 2.0 (the "License"); you may not |
| 5 | # use this file except in compliance with the License. You may obtain a copy of |
| 6 | # the License at |
| 7 | # |
| 8 | # http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | # |
| 10 | # Unless required by applicable law or agreed to in writing, software |
| 11 | # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT |
| 12 | # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the |
| 13 | # License for the specific language governing permissions and limitations under |
| 14 | # the License. |
| 15 | |
| 16 | """ |
| 17 | Basic LE Stress tests. |
| 18 | """ |
| 19 | |
| 20 | import concurrent |
| 21 | import pprint |
| 22 | import time |
| 23 | |
| 24 | from queue import Empty |
| 25 | from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest |
| 26 | from acts.test_utils.bt.bt_test_utils import generate_ble_advertise_objects |
| 27 | from acts.test_utils.bt.bt_test_utils import generate_ble_scan_objects |
| 28 | from acts.test_utils.bt.bt_test_utils import get_advanced_droid_list |
| 29 | from acts.test_utils.bt.bt_test_utils import reset_bluetooth |
| 30 | from acts.test_utils.bt.bt_test_utils import scan_result |
| 31 | |
| 32 | class BleStressTest(BluetoothBaseTest): |
| 33 | tests = None |
| 34 | default_timeout = 10 |
| 35 | |
| 36 | def __init__(self, controllers): |
| 37 | BluetoothBaseTest.__init__(self, controllers) |
| 38 | self.droid_list = get_advanced_droid_list(self.droids, self.eds) |
| 39 | self.scn_droid, self.scn_ed = self.droids[0], self.eds[0] |
| 40 | self.adv_droid, self.adv_ed = self.droids[1], self.eds[1] |
| 41 | self.tests = ( |
| 42 | "test_loop_scanning_1000", |
| 43 | "test_restart_scan_callback_after_bt_toggle", |
Ang Li | 73697b3 | 2015-12-03 00:41:53 +0000 | [diff] [blame] | 44 | "test_start_le_scan_while_toggling_bt", |
| 45 | ) |
| 46 | if self.droid_list[0]['max_advertisements'] > 0: |
| 47 | self.tests = self.tests + ("test_loop_advertising_100", |
| 48 | "test_restart_advertise_callback_after_bt_toggle",) |
tturney | c60fc75 | 2015-12-03 10:30:22 -0800 | [diff] [blame^] | 49 | if self.droid_list[1]['max_advertisements'] >= 4: |
| 50 | self.tests = self.tests + ("test_loop_scanning_100_verify_no_hci_timeout",) |
Ang Li | 73697b3 | 2015-12-03 00:41:53 +0000 | [diff] [blame] | 51 | |
| 52 | def bleadvertise_verify_onsuccess_handler(self, event): |
| 53 | test_result = True |
| 54 | self.log.debug("Verifying onSuccess event") |
| 55 | self.log.debug(pprint.pformat(event)) |
| 56 | return test_result |
| 57 | |
| 58 | @BluetoothBaseTest.bt_test_wrap |
| 59 | def test_loop_scanning_1000(self): |
| 60 | """Stress start/stop scan instances. |
| 61 | |
| 62 | This test will start and stop scan instances as fast as possible. This |
| 63 | will guarantee that the scan instances are properly being cleaned up |
| 64 | when the scan is stopped. |
| 65 | |
| 66 | Steps: |
| 67 | 1. Start a scan instance. |
| 68 | 2. Stop the scan instance. |
| 69 | 3. Repeat steps 1-2 1000 times. |
| 70 | |
| 71 | Expected Result: |
| 72 | Neither starting or stopping scan instances causes any failures. |
| 73 | |
| 74 | Returns: |
| 75 | Pass if True |
| 76 | Fail if False |
| 77 | |
| 78 | TAGS: LE, Scanning, Stress |
| 79 | Priority: 1 |
| 80 | """ |
| 81 | scan_droid, scan_event_dispatcher = self.droid, self.ed |
| 82 | test_result = True |
| 83 | for _ in range(1000): |
| 84 | filter_list, scan_settings, scan_callback = generate_ble_scan_objects( |
| 85 | scan_droid) |
| 86 | self.scn_droid.bleStartBleScan( |
| 87 | filter_list, scan_settings, scan_callback) |
| 88 | self.scn_droid.bleStopBleScan(scan_callback) |
| 89 | return test_result |
| 90 | |
| 91 | @BluetoothBaseTest.bt_test_wrap |
| 92 | def test_loop_scanning_100_verify_no_hci_timeout(self): |
| 93 | """Stress start/stop scan instances variant. |
| 94 | |
| 95 | This test will start and stop scan instances with a one second timeout |
| 96 | in between each iteration. This testcase was added because the specific |
| 97 | timing combination caused hci timeouts. |
| 98 | |
| 99 | Steps: |
| 100 | 1. Start a scan instance. |
| 101 | 2. Stop the scan instance. |
| 102 | 3. Sleep for 1 second. |
| 103 | 4. Repeat steps 1-3 100 times. |
| 104 | |
| 105 | Expected Result: |
| 106 | Neither starting or stopping scan instances causes any failures. |
| 107 | |
| 108 | Returns: |
| 109 | Pass if True |
| 110 | Fail if False |
| 111 | |
| 112 | TAGS: LE, Scanning, Stress |
| 113 | Priority: 1 |
| 114 | """ |
tturney | c60fc75 | 2015-12-03 10:30:22 -0800 | [diff] [blame^] | 115 | for _ in range(self.droid_list[1]['max_advertisements']): |
| 116 | adv_callback, adv_data, adv_settings = generate_ble_advertise_objects( |
| 117 | self.adv_droid) |
| 118 | self.adv_droid.bleStartBleAdvertising( |
| 119 | adv_callback, adv_data, adv_settings) |
Ang Li | 73697b3 | 2015-12-03 00:41:53 +0000 | [diff] [blame] | 120 | for _ in range(100): |
| 121 | filter_list, scan_settings, scan_callback = generate_ble_scan_objects( |
| 122 | self.scn_droid) |
| 123 | self.scn_droid.bleStartBleScan( |
| 124 | filter_list, scan_settings, scan_callback) |
| 125 | self.log.info(self.scn_ed.pop_event( |
| 126 | scan_result.format(scan_callback))) |
| 127 | self.scn_droid.bleStopBleScan(scan_callback) |
| 128 | time.sleep(1) |
tturney | c60fc75 | 2015-12-03 10:30:22 -0800 | [diff] [blame^] | 129 | return True |
Ang Li | 73697b3 | 2015-12-03 00:41:53 +0000 | [diff] [blame] | 130 | |
| 131 | @BluetoothBaseTest.bt_test_wrap |
| 132 | def test_loop_advertising_100(self): |
| 133 | """Stress start/stop advertising instances. |
| 134 | |
| 135 | This test will start and stop advertising instances as fast as possible. |
| 136 | |
| 137 | Steps: |
| 138 | 1. Start a advertising instance. |
| 139 | 2. Find that an onSuccess callback is triggered. |
| 140 | 3. Stop the advertising instance. |
| 141 | 4. Repeat steps 1-3 100 times. |
| 142 | |
| 143 | Expected Result: |
| 144 | Neither starting or stopping advertising instances causes any failures. |
| 145 | |
| 146 | Returns: |
| 147 | Pass if True |
| 148 | Fail if False |
| 149 | |
| 150 | TAGS: LE, Advertising, Stress |
| 151 | Priority: 1 |
| 152 | """ |
| 153 | advertise_droid, advertise_event_dispatcher = self.droid, self.ed |
| 154 | test_result = True |
| 155 | for _ in range(100): |
| 156 | advertise_callback, advertise_data, advertise_settings = generate_ble_advertise_objects( |
| 157 | advertise_droid) |
| 158 | advertise_droid.bleStartBleAdvertising( |
| 159 | advertise_callback, advertise_data, advertise_settings) |
| 160 | expected_advertise_event_name = "".join( |
| 161 | ["BleAdvertise", str(advertise_callback), "onSuccess"]) |
| 162 | worker = advertise_event_dispatcher.handle_event( |
| 163 | self.bleadvertise_verify_onsuccess_handler, expected_advertise_event_name, ( |
| 164 | []), |
| 165 | self.default_timeout) |
| 166 | try: |
| 167 | self.log.debug(worker.result(self.default_timeout)) |
| 168 | except Empty as error: |
| 169 | self.log.debug( |
| 170 | " ".join(["Test failed with Empty error:", str(error)])) |
| 171 | test_result = False |
| 172 | except concurrent.futures._base.TimeoutError as error: |
| 173 | self.log.debug( |
| 174 | " ".join(["Test failed, filtering callback onSuccess never occurred:", |
| 175 | str(error)])) |
| 176 | test_result = False |
| 177 | advertise_droid.bleStopBleAdvertising(advertise_callback) |
| 178 | return test_result |
| 179 | |
| 180 | @BluetoothBaseTest.bt_test_wrap |
| 181 | def test_restart_advertise_callback_after_bt_toggle(self): |
| 182 | """Test to reuse an advertise callback. |
| 183 | |
| 184 | This will verify if advertising objects can be reused after a bluetooth |
| 185 | toggle. |
| 186 | |
| 187 | Steps: |
| 188 | 1. Start a advertising instance. |
| 189 | 2. Find that an onSuccess callback is triggered. |
| 190 | 3. Stop the advertising instance. |
| 191 | 4. Toggle bluetooth off and on. |
| 192 | 5. Start an advertising instance on the same objects used in step 1. |
| 193 | 6. Find that an onSuccess callback is triggered. |
| 194 | |
| 195 | Expected Result: |
| 196 | Advertisement should start successfully. |
| 197 | |
| 198 | Returns: |
| 199 | Pass if True |
| 200 | Fail if False |
| 201 | |
| 202 | TAGS: LE, Advertising, Stress |
| 203 | Priority: 1 |
| 204 | """ |
| 205 | test_result = True |
| 206 | advertise_droid, advertise_event_dispatcher = self.droid, self.ed |
| 207 | advertise_callback, advertise_data, advertise_settings = generate_ble_advertise_objects( |
| 208 | advertise_droid) |
| 209 | advertise_droid.bleStartBleAdvertising( |
| 210 | advertise_callback, advertise_data, advertise_settings) |
| 211 | expected_advertise_event_name = "".join( |
| 212 | ["BleAdvertise", str(advertise_callback), "onSuccess"]) |
| 213 | worker = advertise_event_dispatcher.handle_event( |
| 214 | self.bleadvertise_verify_onsuccess_handler, expected_advertise_event_name, ( |
| 215 | []), |
| 216 | self.default_timeout) |
| 217 | try: |
| 218 | self.log.debug(worker.result(self.default_timeout)) |
| 219 | except Empty as error: |
| 220 | self.log.debug( |
| 221 | " ".join(["Test failed with Empty error:", str(error)])) |
| 222 | test_result = False |
| 223 | except concurrent.futures._base.TimeoutError as error: |
| 224 | self.log.debug( |
| 225 | " ".join(["Test failed, filtering callback onSuccess never occurred:", |
| 226 | str(error)])) |
| 227 | test_result = reset_bluetooth([self.droids[0]], [self.eds[0]]) |
| 228 | if not test_result: |
| 229 | return test_result |
| 230 | time.sleep(5) |
| 231 | advertise_droid.bleStartBleAdvertising( |
| 232 | advertise_callback, advertise_data, advertise_settings) |
| 233 | worker = advertise_event_dispatcher.handle_event( |
| 234 | self.bleadvertise_verify_onsuccess_handler, expected_advertise_event_name, ( |
| 235 | []), |
| 236 | self.default_timeout) |
| 237 | try: |
| 238 | self.log.debug(worker.result(self.default_timeout)) |
| 239 | except Empty as error: |
| 240 | self.log.debug( |
| 241 | " ".join(["Test failed with Empty error:", str(error)])) |
| 242 | test_result = False |
| 243 | except concurrent.futures._base.TimeoutError as error: |
| 244 | self.log.debug( |
| 245 | " ".join(["Test failed, filtering callback onSuccess never occurred:", |
| 246 | str(error)])) |
| 247 | return test_result |
| 248 | |
| 249 | @BluetoothBaseTest.bt_test_wrap |
| 250 | def test_restart_scan_callback_after_bt_toggle(self): |
| 251 | """Test to reuse an scan callback. |
| 252 | |
| 253 | This will verify if scan objects can be reused after a bluetooth |
| 254 | toggle. |
| 255 | |
| 256 | Steps: |
| 257 | 1. Start a scanning instance. |
| 258 | 3. Stop the scanning instance. |
| 259 | 4. Toggle bluetooth off and on. |
| 260 | 5. Start an scanning instance on the same objects used in step 1. |
| 261 | |
| 262 | Expected Result: |
| 263 | Scanner should start successfully. |
| 264 | |
| 265 | Returns: |
| 266 | Pass if True |
| 267 | Fail if False |
| 268 | |
| 269 | TAGS: LE, Scanning, Stress |
| 270 | Priority: 1 |
| 271 | """ |
| 272 | test_result = True |
| 273 | scan_droid, scan_event_dispatcher = self.droid, self.ed |
| 274 | filter_list, scan_settings, scan_callback = generate_ble_scan_objects( |
| 275 | scan_droid) |
| 276 | self.scn_droid.bleStartBleScan( |
| 277 | filter_list, scan_settings, scan_callback) |
| 278 | reset_bluetooth([scan_droid], [scan_event_dispatcher]) |
| 279 | self.scn_droid.bleStartBleScan( |
| 280 | filter_list, scan_settings, scan_callback) |
| 281 | |
| 282 | return test_result |