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"""
17Test script to exercises Ble Scans can run in concurrency.
18This test was designed to be run in a shield box.
19"""
20
21import concurrent
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_constants import ble_advertise_settings_modes
28from acts.test_utils.bt.bt_constants import ble_scan_settings_callback_types
29from acts.test_utils.bt.bt_constants import ble_scan_settings_modes
30from acts.test_utils.bt.bt_constants import adv_succ
31from acts.test_utils.bt.bt_test_utils import generate_ble_advertise_objects
32from acts.test_utils.bt.bt_test_utils import reset_bluetooth
33from acts.test_utils.bt.bt_constants import scan_failed
34from acts.test_utils.bt.bt_constants import scan_result
35from acts.test_utils.bt.bt_test_utils import take_btsnoop_logs
36
37
38class ConcurrentBleScanningTest(BluetoothBaseTest):
39    default_timeout = 20
40    max_concurrent_scans = 27
41
42    def setup_class(self):
43        super().setup_class()
44        self.scn_ad = self.android_devices[0]
45        self.adv_ad = self.android_devices[1]
46
47    def on_fail(self, test_name, begin_time):
48        self.log.debug("Test {} failed. Gathering bugreport and btsnoop logs."
49                       .format(test_name))
50        take_btsnoop_logs(self.android_devices, self, test_name)
51        reset_bluetooth(self.android_devices)
52
53    def setup_test(self):
54        return reset_bluetooth(self.android_devices)
55
56    @BluetoothBaseTest.bt_test_wrap
57    @test_tracker_info(uuid='e7f68b9b-fb3f-48e9-a272-e41c2a32b4bd')
58    def test_max_concurrent_ble_scans(self):
59        """Test max LE scans.
60
61        Test that a single device can have max scans concurrently scanning.
62
63        Steps:
64        1. Initialize scanner
65        2. Initialize advertiser
66        3. Start advertising on the device from step 2
67        4. Create max ble scan callbacks
68        5. Start ble scan on each callback
69        6. Verify that each callback triggers
70        7. Stop all scans and advertisements
71
72        Expected Result:
73        All scanning instances should start without errors and the advertisement
74        should be found on each scan instance.
75
76        Returns:
77          Pass if True
78          Fail if False
79
80        TAGS: LE, Scanning, Concurrency
81        Priority: 0
82        """
83        test_result = True
84        self.adv_ad.droid.bleSetAdvertiseDataIncludeDeviceName(True)
85        self.scn_ad.droid.bleSetScanSettingsCallbackType(
86            ble_scan_settings_callback_types['all_matches'])
87        self.scn_ad.droid.bleSetScanSettingsScanMode(ble_scan_settings_modes[
88            'low_latency'])
89        self.adv_ad.droid.bleSetAdvertiseSettingsAdvertiseMode(
90            ble_advertise_settings_modes['low_latency'])
91        advertise_callback, advertise_data, advertise_settings = (
92            generate_ble_advertise_objects(self.adv_ad.droid))
93        self.adv_ad.droid.bleSetAdvertiseSettingsIsConnectable(False)
94        self.adv_ad.droid.bleStartBleAdvertising(
95            advertise_callback, advertise_data, advertise_settings)
96        try:
97            self.adv_ad.ed.pop_event(
98                adv_succ.format(advertise_callback), self.default_timeout)
99        except Empty as error:
100            self.log.exception("Test failed with Empty error: {}".format(
101                error))
102            test_result = False
103        except concurrent.futures._base.TimeoutError as error:
104            self.log.exception(
105                "Test failed callback onSuccess never occurred: "
106                "{}".format(error))
107            test_result = False
108        if not test_result:
109            return test_result
110        filter_list = self.scn_ad.droid.bleGenFilterList()
111        self.scn_ad.droid.bleSetScanFilterDeviceName(
112            self.adv_ad.droid.bluetoothGetLocalName())
113        self.scn_ad.droid.bleBuildScanFilter(filter_list)
114        scan_settings = self.scn_ad.droid.bleBuildScanSetting()
115        scan_callback_list = []
116        for i in range(self.max_concurrent_scans):
117            self.log.debug("Concurrent Ble Scan iteration {}".format(i + 1))
118            scan_callback = self.scn_ad.droid.bleGenScanCallback()
119            scan_callback_list.append(scan_callback)
120            self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
121                                              scan_callback)
122            try:
123                self.scn_ad.ed.pop_event(
124                    scan_result.format(scan_callback), self.default_timeout)
125                self.log.info("Found scan event successfully. Iteration {} "
126                              "successful.".format(i))
127            except Exception:
128                self.log.info("Failed to find a scan result for callback {}"
129                              .format(scan_callback))
130                test_result = False
131                break
132        for callback in scan_callback_list:
133            self.scn_ad.droid.bleStopBleScan(callback)
134        self.adv_ad.droid.bleStopBleAdvertising(advertise_callback)
135        if not test_result:
136            return test_result
137        self.log.info("Waiting for scan callbacks to stop completely.")
138        # Wait for all scan callbacks to stop. There is no confirmation
139        # otherwise.
140        time.sleep(10)
141        return test_result
142
143    @BluetoothBaseTest.bt_test_wrap
144    @test_tracker_info(uuid='58b0c45e-1cbc-420a-9e89-901518ffe3d1')
145    def test_max_concurrent_ble_scans_then_discover_advertisement(self):
146        """Test max LE scans variant.
147
148        Test that a single device can have max scans concurrently scanning.
149
150        Steps:
151        1. Initialize scanner
152        2. Initialize advertiser
153        3. Create max ble scan callbacks
154        4. Start ble scan on each callback
155        5. Start advertising on the device from step 2
156        6. Verify that each callback triggers
157        7. Stop all scans and advertisements
158
159        Expected Result:
160        All scanning instances should start without errors and the advertisement
161        should be found on each scan instance.
162
163        Returns:
164          Pass if True
165          Fail if False
166
167        TAGS: LE, Scanning, Concurrency
168        Priority: 1
169        """
170        self.adv_ad.droid.bleSetAdvertiseDataIncludeDeviceName(True)
171        self.scn_ad.droid.bleSetScanSettingsCallbackType(
172            ble_scan_settings_callback_types['all_matches'])
173        self.scn_ad.droid.bleSetScanSettingsScanMode(ble_scan_settings_modes[
174            'low_latency'])
175        self.adv_ad.droid.bleSetAdvertiseSettingsAdvertiseMode(
176            ble_advertise_settings_modes['low_latency'])
177        advertise_callback, advertise_data, advertise_settings = (
178            generate_ble_advertise_objects(self.adv_ad.droid))
179        filter_list = self.scn_ad.droid.bleGenFilterList()
180        self.scn_ad.droid.bleSetScanFilterDeviceName(
181            self.adv_ad.droid.bluetoothGetLocalName())
182        self.scn_ad.droid.bleBuildScanFilter(filter_list)
183        scan_settings = self.scn_ad.droid.bleBuildScanSetting()
184        scan_callback_list = []
185        for i in range(self.max_concurrent_scans):
186            self.log.debug("Concurrent Ble Scan iteration {}".format(i + 1))
187            scan_callback = self.scn_ad.droid.bleGenScanCallback()
188            scan_callback_list.append(scan_callback)
189            self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
190                                              scan_callback)
191        self.adv_ad.droid.bleStartBleAdvertising(
192            advertise_callback, advertise_data, advertise_settings)
193        try:
194            self.adv_ad.ed.pop_event(
195                adv_succ.format(advertise_callback), self.default_timeout)
196        except Empty as error:
197            self.log.exception("Test failed with Empty error: {}".format(
198                error))
199            return False
200        except concurrent.futures._base.TimeoutError as error:
201            self.log.exception("Test failed, filtering callback onSuccess "
202                               "never occurred: {}".format(error))
203            return False
204        i = 0
205        for callback in scan_callback_list:
206            try:
207                self.scn_ad.ed.pop_event(
208                    scan_result.format(callback), self.default_timeout)
209                self.log.info(
210                    "Found scan event successfully. Iteration {} successful."
211                    .format(i))
212            except Exception:
213                self.log.info("Failed to find a scan result for callback {}"
214                              .format(callback))
215                return False
216            i += 1
217        for callback in scan_callback_list:
218            self.scn_ad.droid.bleStopBleScan(callback)
219        self.adv_ad.droid.bleStopBleAdvertising(advertise_callback)
220        return True
221
222    @BluetoothBaseTest.bt_test_wrap
223    @test_tracker_info(uuid='7a45e45c-faf3-4e89-abb7-a52f63e53208')
224    def test_max_concurrent_ble_scans_plus_one(self):
225        """Test mac LE scans variant.
226
227        Test that a single device can have max scans concurrently scanning.
228
229        Steps:
230        1. Initialize scanner
231        3. Create max ble scan callbacks plus one
232        5. Start ble scan on each callback
233        6. Verify that the n+1th scan fails.
234        7. Stop all scans
235
236        Expected Result:
237        The n+1th scan should fail to start.
238
239        Returns:
240          Pass if True
241          Fail if False
242
243        TAGS: LE, Scanning, Concurrency
244        Priority: 1
245        """
246        test_result = True
247        filter_list = self.scn_ad.droid.bleGenFilterList()
248        scan_settings = self.scn_ad.droid.bleBuildScanSetting()
249        scan_callback_list = []
250        for i in range(self.max_concurrent_scans):
251            self.log.debug("Concurrent Ble Scan iteration {}".format(i + 1))
252            scan_callback = self.scn_ad.droid.bleGenScanCallback()
253            self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
254                                              scan_callback)
255            scan_callback_list.append(scan_callback)
256        scan_callback = self.scn_ad.droid.bleGenScanCallback()
257        self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
258                                          scan_callback)
259        try:
260            self.scn_ad.ed.pop_event(
261                scan_failed.format(scan_callback), self.default_timeout)
262            self.log.info(
263                "Found scan event successfully. Iteration {} successful."
264                .format(self.max_concurrent_scans + 1))
265        except Exception:
266            self.log.error("Failed to find a onScanFailed event for callback {}"
267                           .format(scan_callback))
268            test_result = False
269        for callback in scan_callback_list:
270            self.scn_ad.droid.bleStopBleScan(callback)
271        return test_result
272
273    @BluetoothBaseTest.bt_test_wrap
274    @test_tracker_info(uuid='5a91f612-69e5-490f-b9d0-50d58a3db736')
275    def test_max_concurrent_ble_scans_verify_scans_stop_independently(self):
276        """Test max LE scans variant.
277
278        Test that a single device can have max scans concurrently scanning.
279
280        Steps:
281        1. Initialize scanner
282        2. Initialize advertiser
283        3. Create max ble scan callbacks
284        4. Start ble scan on each callback
285        5. Start advertising on the device from step 2
286        6. Verify that the first callback triggers
287        7. Stop the scan and repeat steps 6 and 7 until all scans stopped
288
289        Expected Result:
290        All scanning instances should start without errors and the advertisement
291        should be found on each scan instance. All scanning instances should
292        stop successfully.
293
294        Returns:
295          Pass if True
296          Fail if False
297
298        TAGS: LE, Scanning, Concurrency
299        Priority: 1
300        """
301        self.adv_ad.droid.bleSetAdvertiseDataIncludeDeviceName(True)
302        self.scn_ad.droid.bleSetScanSettingsCallbackType(
303            ble_scan_settings_callback_types['all_matches'])
304        self.scn_ad.droid.bleSetScanSettingsScanMode(ble_scan_settings_modes[
305            'low_latency'])
306        self.adv_ad.droid.bleSetAdvertiseSettingsAdvertiseMode(
307            ble_advertise_settings_modes['low_latency'])
308        advertise_callback, advertise_data, advertise_settings = (
309            generate_ble_advertise_objects(self.adv_ad.droid))
310        filter_list = self.scn_ad.droid.bleGenFilterList()
311        self.scn_ad.droid.bleSetScanFilterDeviceName(
312            self.adv_ad.droid.bluetoothGetLocalName())
313        self.scn_ad.droid.bleBuildScanFilter(filter_list)
314        scan_settings = self.scn_ad.droid.bleBuildScanSetting()
315        scan_callback_list = []
316        for i in range(self.max_concurrent_scans):
317            self.log.debug("Concurrent Ble Scan iteration {}".format(i + 1))
318            scan_callback = self.scn_ad.droid.bleGenScanCallback()
319            scan_callback_list.append(scan_callback)
320            self.scn_ad.droid.bleStartBleScan(filter_list, scan_settings,
321                                              scan_callback)
322        self.adv_ad.droid.bleStartBleAdvertising(
323            advertise_callback, advertise_data, advertise_settings)
324        try:
325            self.adv_ad.ed.pop_event(
326                adv_succ.format(advertise_callback), self.default_timeout)
327        except Empty as error:
328            self.log.exception("Test failed with Empty error: {}".format(
329                error))
330            return False
331        except concurrent.futures._base.TimeoutError as error:
332            self.log.exception(
333                "Test failed, filtering callback onSuccess never"
334                " occurred: {}".format(error))
335            return False
336        i = 0
337        for callback in scan_callback_list:
338            expected_scan_event_name = scan_result.format(scan_callback)
339            try:
340                self.scn_ad.ed.pop_event(expected_scan_event_name,
341                                         self.default_timeout)
342                self.log.info(
343                    "Found scan event successfully. Iteration {} successful.".
344                    format(i))
345                i += 1
346            except Exception:
347                self.log.info("Failed to find a scan result for callback {}".
348                              format(scan_callback))
349                return False
350            self.scn_ad.droid.bleStopBleScan(callback)
351        self.adv_ad.droid.bleStopBleAdvertising(advertise_callback)
352        return True
353