• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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"""
17OnLost onFound Stress Test.
18"""
19
20import threading
21import time
22
23from queue import Empty
24from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
25from acts.test_utils.bt.BleEnum import AdvertiseSettingsAdvertiseMode
26from acts.test_utils.bt.BleEnum import ScanSettingsCallbackType
27from acts.test_utils.bt.BleEnum import ScanSettingsScanMode
28from acts.test_utils.bt.BleEnum import ScanSettingsMatchMode
29from acts.test_utils.bt.BleEnum import ScanSettingsMatchNum
30from acts.test_utils.bt.bt_test_utils import cleanup_scanners_and_advertisers
31from acts.test_utils.bt.bt_gatt_utils import orchestrate_gatt_connection
32from acts.test_utils.bt.bt_test_utils import reset_bluetooth
33from acts.test_utils.bt.bt_gatt_utils import run_continuous_write_descriptor
34from acts.test_utils.bt.bt_gatt_utils import setup_multiple_services
35
36
37class 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