1#!/usr/bin/env python3
2#
3#   Copyright 2016 - Google
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of 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,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16"""
17    Base Class for Defining Common Bluetooth Test Functionality
18"""
19
20import threading
21import time
22import traceback
23import os
24from acts.base_test import BaseTestClass
25from acts.signals import TestSignal
26from acts.utils import dump_string_to_file
27
28from acts.controllers import android_device
29from acts.libs.proto.proto_utils import compile_import_proto
30from acts.libs.proto.proto_utils import parse_proto_to_ascii
31from acts.test_utils.bt.bt_metrics_utils import get_bluetooth_metrics
32from acts.test_utils.bt.bt_test_utils import get_device_selector_dictionary
33from acts.test_utils.bt.bt_test_utils import reset_bluetooth
34from acts.test_utils.bt.bt_test_utils import setup_multiple_devices_for_bt_test
35from acts.test_utils.bt.bt_test_utils import take_btsnoop_logs
36from acts.test_utils.bt.ble_lib import BleLib
37from acts.test_utils.bt.bta_lib import BtaLib
38from acts.test_utils.bt.config_lib import ConfigLib
39from acts.test_utils.bt.gattc_lib import GattClientLib
40from acts.test_utils.bt.gatts_lib import GattServerLib
41from acts.test_utils.bt.rfcomm_lib import RfcommLib
42from acts.test_utils.bt.shell_commands_lib import ShellCommands
43
44
45class BluetoothBaseTest(BaseTestClass):
46    DEFAULT_TIMEOUT = 10
47    start_time = 0
48    timer_list = []
49
50    def collect_bluetooth_manager_metrics_logs(self, ads, test_name):
51        """
52        Collect Bluetooth metrics logs, save an ascii log to disk and return
53        both binary and ascii logs to caller
54        :param ads: list of active Android devices
55        :return: List of binary metrics logs,
56                List of ascii metrics logs
57        """
58        bluetooth_logs = []
59        bluetooth_logs_ascii = []
60        for ad in ads:
61            serial = ad.serial
62            out_name = "{}_{}_{}".format(serial, test_name,
63                                         "bluetooth_metrics.txt")
64            bluetooth_log = get_bluetooth_metrics(ad,
65                                                  ad.bluetooth_proto_module)
66            bluetooth_log_ascii = parse_proto_to_ascii(bluetooth_log)
67            dump_string_to_file(bluetooth_log_ascii,
68                                os.path.join(ad.metrics_path, out_name))
69            bluetooth_logs.append(bluetooth_log)
70            bluetooth_logs_ascii.append(bluetooth_log_ascii)
71        return bluetooth_logs, bluetooth_logs_ascii
72
73    # Use for logging in the test cases to facilitate
74    # faster log lookup and reduce ambiguity in logging.
75    @staticmethod
76    def bt_test_wrap(fn):
77        def _safe_wrap_test_case(self, *args, **kwargs):
78            test_id = "{}:{}:{}".format(self.__class__.__name__, fn.__name__,
79                                        time.time())
80            log_string = "[Test ID] {}".format(test_id)
81            self.log.info(log_string)
82            try:
83                for ad in self.android_devices:
84                    ad.droid.logI("Started " + log_string)
85                result = fn(self, *args, **kwargs)
86                for ad in self.android_devices:
87                    ad.droid.logI("Finished " + log_string)
88                if result is not True and "bt_auto_rerun" in self.user_params:
89                    self.teardown_test()
90                    log_string = "[Rerun Test ID] {}. 1st run failed.".format(
91                        test_id)
92                    self.log.info(log_string)
93                    self.setup_test()
94                    for ad in self.android_devices:
95                        ad.droid.logI("Rerun Started " + log_string)
96                    result = fn(self, *args, **kwargs)
97                    if result is True:
98                        self.log.info("Rerun passed.")
99                    elif result is False:
100                        self.log.info("Rerun failed.")
101                    else:
102                        # In the event that we have a non-bool or null
103                        # retval, we want to clearly distinguish this in the
104                        # logs from an explicit failure, though the test will
105                        # still be considered a failure for reporting purposes.
106                        self.log.info("Rerun indeterminate.")
107                        result = False
108                return result
109            except TestSignal:
110                raise
111            except Exception as e:
112                self.log.error(traceback.format_exc())
113                self.log.error(str(e))
114                raise
115            return fn(self, *args, **kwargs)
116
117        return _safe_wrap_test_case
118
119    def setup_class(self):
120        super().setup_class()
121        for ad in self.android_devices:
122            self._setup_bt_libs(ad)
123        if 'preferred_device_order' in self.user_params:
124            prefered_device_order = self.user_params['preferred_device_order']
125            for i, ad in enumerate(self.android_devices):
126                if ad.serial in prefered_device_order:
127                    index = prefered_device_order.index(ad.serial)
128                    self.android_devices[i], self.android_devices[index] = \
129                        self.android_devices[index], self.android_devices[i]
130
131        if "reboot_between_test_class" in self.user_params:
132            threads = []
133            for a in self.android_devices:
134                thread = threading.Thread(
135                    target=self._reboot_device, args=([a]))
136                threads.append(thread)
137                thread.start()
138            for t in threads:
139                t.join()
140        if not setup_multiple_devices_for_bt_test(self.android_devices):
141            return False
142        self.device_selector = get_device_selector_dictionary(
143            self.android_devices)
144        if "bluetooth_proto_path" in self.user_params:
145            from google import protobuf
146
147            self.bluetooth_proto_path = self.user_params[
148                "bluetooth_proto_path"][0]
149            if not os.path.isfile(self.bluetooth_proto_path):
150                try:
151                    self.bluetooth_proto_path = "{}/bluetooth.proto".format(
152                        os.path.dirname(os.path.realpath(__file__)))
153                except Exception:
154                    self.log.error("File not found.")
155                if not os.path.isfile(self.bluetooth_proto_path):
156                    self.log.error("Unable to find Bluetooth proto {}.".format(
157                        self.bluetooth_proto_path))
158                    return False
159            for ad in self.android_devices:
160                ad.metrics_path = os.path.join(ad.log_path, "BluetoothMetrics")
161                os.makedirs(ad.metrics_path, exist_ok=True)
162                ad.bluetooth_proto_module = \
163                    compile_import_proto(ad.metrics_path, self.bluetooth_proto_path)
164                if not ad.bluetooth_proto_module:
165                    self.log.error("Unable to compile bluetooth proto at " +
166                                   self.bluetooth_proto_path)
167                    return False
168                # Clear metrics.
169                get_bluetooth_metrics(ad, ad.bluetooth_proto_module)
170        return True
171
172    def teardown_class(self):
173        if "bluetooth_proto_path" in self.user_params:
174            # Collect metrics here bassed off class name
175            bluetooth_logs, bluetooth_logs_ascii = \
176                self.collect_bluetooth_manager_metrics_logs(
177                    self.android_devices, self.__class__.__name__)
178
179    def setup_test(self):
180        self.timer_list = []
181        for a in self.android_devices:
182            a.ed.clear_all_events()
183            a.droid.setScreenTimeout(500)
184            a.droid.wakeUpNow()
185        return True
186
187    def on_fail(self, test_name, begin_time):
188        self.log.debug(
189            "Test {} failed. Gathering bugreport and btsnoop logs".format(
190                test_name))
191        take_btsnoop_logs(self.android_devices, self, test_name)
192        self._take_bug_report(test_name, begin_time)
193        for _ in range(5):
194            if reset_bluetooth(self.android_devices):
195                break
196            else:
197                self.log.error("Failed to reset Bluetooth... retrying.")
198        return
199
200    def _get_time_in_milliseconds(self):
201        return int(round(time.time() * 1000))
202
203    def start_timer(self):
204        self.start_time = self._get_time_in_milliseconds()
205
206    def end_timer(self):
207        total_time = self._get_time_in_milliseconds() - self.start_time
208        self.timer_list.append(total_time)
209        self.start_time = 0
210        return total_time
211
212    def log_stats(self):
213        if self.timer_list:
214            self.log.info("Overall list {}".format(self.timer_list))
215            self.log.info("Average of list {}".format(
216                sum(self.timer_list) / float(len(self.timer_list))))
217            self.log.info("Maximum of list {}".format(max(self.timer_list)))
218            self.log.info("Minimum of list {}".format(min(self.timer_list)))
219            self.log.info("Total items in list {}".format(
220                len(self.timer_list)))
221        self.timer_list = []
222
223    def _setup_bt_libs(self, android_device):
224        # Bluetooth Low Energy library.
225        setattr(android_device, "ble", BleLib(
226            log=self.log, dut=android_device))
227        # Bluetooth Adapter library.
228        setattr(android_device, "bta", BtaLib(
229            log=self.log, dut=android_device))
230        # Bluetooth stack config library.
231        setattr(android_device, "config",
232                ConfigLib(log=self.log, dut=android_device))
233        # GATT Client library.
234        setattr(android_device, "gattc",
235                GattClientLib(log=self.log, dut=android_device))
236        # GATT Server library.
237        setattr(android_device, "gatts",
238                GattServerLib(log=self.log, dut=android_device))
239        # RFCOMM library.
240        setattr(android_device, "rfcomm",
241                RfcommLib(log=self.log, dut=android_device))
242        # Shell command library
243        setattr(android_device, "shell",
244                ShellCommands(log=self.log, dut=android_device))
245        # Setup Android Device feature list
246        setattr(android_device, "features",
247                android_device.adb.shell("pm list features").split("\n"))
248