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