1#!/usr/bin/env python3
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"""
17Basic LE Stress tests.
18"""
19
20import concurrent
21import pprint
22import time
23
24from queue import Empty
25from acts.test_decorators import test_tracker_info
26from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
27from acts.test_utils.bt.bt_test_utils import BtTestUtilsError
28from acts.test_utils.bt.bt_test_utils import clear_bonded_devices
29from acts.test_utils.bt.bt_test_utils import generate_ble_advertise_objects
30from acts.test_utils.bt.bt_test_utils import generate_ble_scan_objects
31from acts.test_utils.bt.bt_test_utils import get_advanced_droid_list
32from acts.test_utils.bt.bt_test_utils import get_mac_address_of_generic_advertisement
33from acts.test_utils.bt.bt_test_utils import reset_bluetooth
34from acts.test_utils.bt.bt_constants import scan_result
35
36
37class BleStressTest(BluetoothBaseTest):
38    default_timeout = 10
39    PAIRING_TIMEOUT = 20
40
41    def setup_class(self):
42        super().setup_class()
43        self.droid_list = get_advanced_droid_list(self.android_devices)
44        self.scn_ad = self.android_devices[0]
45        self.adv_ad = self.android_devices[1]
46
47    def teardown_test(self):
48        super(BluetoothBaseTest, self).teardown_test()
49        self.log_stats()
50
51    def bleadvertise_verify_onsuccess_handler(self, event):
52        test_result = True
53        self.log.debug("Verifying onSuccess event")
54        self.log.debug(pprint.pformat(event))
55        return test_result
56
57    def _verify_successful_bond(self, target_address):
58        end_time = time.time() + self.PAIRING_TIMEOUT
59        self.log.info("Verifying devices are bonded")
60        while time.time() < end_time:
61            bonded_devices = self.scn_ad.droid.bluetoothGetBondedDevices()
62            if target_address in {d['address'] for d in bonded_devices}:
63                self.log.info("Successfully bonded to device")
64                return True
65        return False
66
67    @BluetoothBaseTest.bt_test_wrap
68    @test_tracker_info(uuid='22f98085-6ed8-4ad8-b62d-b8d1eae20b89')
69    def test_loop_scanning_1000(self):
70        """Stress start/stop scan instances.
71
72        This test will start and stop scan instances as fast as possible. This
73        will guarantee that the scan instances are properly being cleaned up
74        when the scan is stopped.
75
76        Steps:
77        1. Start a scan instance.
78        2. Stop the scan instance.
79        3. Repeat steps 1-2 1000 times.
80
81        Expected Result:
82        Neither starting or stopping scan instances causes any failures.
83
84        Returns:
85          Pass if True
86          Fail if False
87
88        TAGS: LE, Scanning, Stress
89        Priority: 1
90        """
91        test_result = True
92        for _ in range(1000):
93            filter_list, scan_settings, scan_callback = generate_ble_scan_objects(
94                self.scn_ad.droid)
95            self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
96                                              scan_callback)
97            self.scn_ad.droid.bleStopBleScan(scan_callback)
98        return test_result
99
100    @BluetoothBaseTest.bt_test_wrap
101    @test_tracker_info(uuid='6caa84c2-50ac-46f2-a5e5-f942fd2cd6f6')
102    def test_loop_scanning_100_verify_no_hci_timeout(self):
103        """Stress start/stop scan instances variant.
104
105        This test will start and stop scan instances with a one second timeout
106        in between each iteration. This testcase was added because the specific
107        timing combination caused hci timeouts.
108
109        Steps:
110        1. Start a scan instance.
111        2. Stop the scan instance.
112        3. Sleep for 1 second.
113        4. Repeat steps 1-3 100 times.
114
115        Expected Result:
116        Neither starting or stopping scan instances causes any failures.
117
118        Returns:
119          Pass if True
120          Fail if False
121
122        TAGS: LE, Scanning, Stress
123        Priority: 1
124        """
125        for _ in range(self.droid_list[1]['max_advertisements']):
126            adv_callback, adv_data, adv_settings = generate_ble_advertise_objects(
127                self.adv_ad.droid)
128            self.adv_ad.droid.bleStartBleAdvertising(adv_callback, adv_data,
129                                                     adv_settings)
130        for _ in range(100):
131            filter_list, scan_settings, scan_callback = generate_ble_scan_objects(
132                self.scn_ad.droid)
133            self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
134                                              scan_callback)
135            self.log.info(
136                self.scn_ad.ed.pop_event(scan_result.format(scan_callback)))
137            self.scn_ad.droid.bleStopBleScan(scan_callback)
138            time.sleep(1)
139        return True
140
141    @BluetoothBaseTest.bt_test_wrap
142    @test_tracker_info(uuid='5e9e4c8d-b72e-4767-81e5-f907c1834430')
143    def test_loop_advertising_100(self):
144        """Stress start/stop advertising instances.
145
146        This test will start and stop advertising instances as fast as possible.
147
148        Steps:
149        1. Start a advertising instance.
150        2. Find that an onSuccess callback is triggered.
151        3. Stop the advertising instance.
152        4. Repeat steps 1-3 100 times.
153
154        Expected Result:
155        Neither starting or stopping advertising instances causes any failures.
156
157        Returns:
158          Pass if True
159          Fail if False
160
161        TAGS: LE, Advertising, Stress
162        Priority: 1
163        """
164        test_result = True
165        for _ in range(100):
166            advertise_callback, advertise_data, advertise_settings = generate_ble_advertise_objects(
167                self.adv_ad.droid)
168            self.adv_ad.droid.bleStartBleAdvertising(
169                advertise_callback, advertise_data, advertise_settings)
170            expected_advertise_event_name = "".join(
171                ["BleAdvertise",
172                 str(advertise_callback), "onSuccess"])
173            worker = self.adv_ad.ed.handle_event(
174                self.bleadvertise_verify_onsuccess_handler,
175                expected_advertise_event_name, ([]), self.default_timeout)
176            try:
177                self.log.debug(worker.result(self.default_timeout))
178            except Empty as error:
179                self.log.debug(" ".join(
180                    ["Test failed with Empty error:",
181                     str(error)]))
182                test_result = False
183            except concurrent.futures._base.TimeoutError as error:
184                self.log.debug(" ".join([
185                    "Test failed, filtering callback onSuccess never occurred:",
186                    str(error)
187                ]))
188                test_result = False
189            self.adv_ad.droid.bleStopBleAdvertising(advertise_callback)
190        return test_result
191
192    @BluetoothBaseTest.bt_test_wrap
193    @test_tracker_info(uuid='11a2f51b-7178-4c32-bb5c-7eddd100a50f')
194    def test_restart_advertise_callback_after_bt_toggle(self):
195        """Test to reuse an advertise callback.
196
197        This will verify if advertising objects can be reused after a bluetooth
198        toggle.
199
200        Steps:
201        1. Start a advertising instance.
202        2. Find that an onSuccess callback is triggered.
203        3. Stop the advertising instance.
204        4. Toggle bluetooth off and on.
205        5. Start an advertising instance on the same objects used in step 1.
206        6. Find that an onSuccess callback is triggered.
207
208        Expected Result:
209        Advertisement should start successfully.
210
211        Returns:
212          Pass if True
213          Fail if False
214
215        TAGS: LE, Advertising, Stress
216        Priority: 1
217        """
218        test_result = True
219        advertise_callback, advertise_data, advertise_settings = generate_ble_advertise_objects(
220            self.adv_ad.droid)
221        self.adv_ad.droid.bleStartBleAdvertising(
222            advertise_callback, advertise_data, advertise_settings)
223        expected_advertise_event_name = "".join(
224            ["BleAdvertise",
225             str(advertise_callback), "onSuccess"])
226        worker = self.adv_ad.ed.handle_event(
227            self.bleadvertise_verify_onsuccess_handler,
228            expected_advertise_event_name, ([]), self.default_timeout)
229        try:
230            self.log.debug(worker.result(self.default_timeout))
231        except Empty as error:
232            self.log.debug(" ".join(
233                ["Test failed with Empty error:",
234                 str(error)]))
235            test_result = False
236        except concurrent.futures._base.TimeoutError as error:
237            self.log.debug(" ".join([
238                "Test failed, filtering callback onSuccess never occurred:",
239                str(error)
240            ]))
241        test_result = reset_bluetooth([self.scn_ad])
242        if not test_result:
243            return test_result
244        time.sleep(5)
245        self.adv_ad.droid.bleStartBleAdvertising(
246            advertise_callback, advertise_data, advertise_settings)
247        worker = self.adv_ad.ed.handle_event(
248            self.bleadvertise_verify_onsuccess_handler,
249            expected_advertise_event_name, ([]), self.default_timeout)
250        try:
251            self.log.debug(worker.result(self.default_timeout))
252        except Empty as error:
253            self.log.debug(" ".join(
254                ["Test failed with Empty error:",
255                 str(error)]))
256            test_result = False
257        except concurrent.futures._base.TimeoutError as error:
258            self.log.debug(" ".join([
259                "Test failed, filtering callback onSuccess never occurred:",
260                str(error)
261            ]))
262        return test_result
263
264    @BluetoothBaseTest.bt_test_wrap
265    @test_tracker_info(uuid='88f3c068-41be-41df-920c-c0ecaae1a619')
266    def test_restart_scan_callback_after_bt_toggle(self):
267        """Test to reuse an scan callback.
268
269        This will verify if scan objects can be reused after a bluetooth
270        toggle.
271
272        Steps:
273        1. Start a scanning instance.
274        3. Stop the scanning instance.
275        4. Toggle bluetooth off and on.
276        5. Start an scanning instance on the same objects used in step 1.
277
278        Expected Result:
279        Scanner should start successfully.
280
281        Returns:
282          Pass if True
283          Fail if False
284
285        TAGS: LE, Scanning, Stress
286        Priority: 1
287        """
288        test_result = True
289        filter_list, scan_settings, scan_callback = generate_ble_scan_objects(
290            self.scn_ad.droid)
291        self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
292                                          scan_callback)
293        reset_bluetooth([self.scn_ad])
294        self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
295                                          scan_callback)
296
297        return test_result
298
299    @BluetoothBaseTest.bt_test_wrap
300    @test_tracker_info(uuid='ce8adfc0-384f-4751-9438-13a76cada8da')
301    def test_le_pairing(self):
302        """Test LE pairing transport stress
303
304        This will test LE pairing between two android devices.
305
306        Steps:
307        1. Start an LE advertisement on secondary device.
308        2. Find address from primary device.
309        3. Discover and bond to LE address.
310        4. Stop LE advertisement on secondary device.
311        5. Repeat steps 1-4 100 times
312
313        Expected Result:
314        LE pairing should pass 100 times.
315
316        Returns:
317          Pass if True
318          Fail if False
319
320        TAGS: LE, Scanning, Stress, Pairing
321        Priority: 1
322        """
323        iterations = 100
324        for i in range(iterations):
325            try:
326                target_address, adv_callback, scan_callback = get_mac_address_of_generic_advertisement(
327                    self.scn_ad, self.adv_ad)
328            except BtTestUtilsError as err:
329                self.log.error(err)
330                return False
331            self.log.info("Begin interation {}/{}".format(i + 1, iterations))
332            self.scn_ad.droid.bluetoothStartPairingHelper()
333            self.adv_ad.droid.bluetoothStartPairingHelper()
334            start_time = self.start_timer()
335            self.scn_ad.droid.bluetoothDiscoverAndBond(target_address)
336            if not self._verify_successful_bond(target_address):
337                self.log.error("Failed to bond devices.")
338                return False
339            self.log.info("Total time (ms): {}".format(self.end_timer()))
340            if not self._verify_successful_bond(self.adv_ad.droid.bluetoothGetLocalAddress()):
341                self.log.error("Failed to bond BREDR devices.")
342                return False
343            if not self.scn_ad.droid.bluetoothUnbond(target_address):
344                self.log.error("Failed to unbond device from scanner.")
345                return False
346            time.sleep(2)
347            if not self.adv_ad.droid.bluetoothUnbond(self.scn_ad.droid.bluetoothGetLocalAddress()):
348                self.log.error("Failed to unbond device from advertiser.")
349                return False
350            self.adv_ad.droid.bleStopBleAdvertising(adv_callback)
351            self.scn_ad.droid.bleStopBleScan(scan_callback)
352            # Magic sleep to let unbonding finish
353            time.sleep(2)
354        return True
355