1#!/usr/bin/env python3
2#
3#   Copyright 2021 - The Android Open Source Project
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
17import traceback
18import os
19import logging
20
21from functools import wraps
22from grpc import RpcError
23
24from mobly import signals
25from mobly.base_test import BaseTestClass
26from mobly.controllers.android_device_lib.adb import AdbError
27from mobly.controllers import android_device
28from mobly.controllers.android_device import MOBLY_CONTROLLER_CONFIG_NAME as ANDROID_DEVICE_COFNIG_NAME
29
30from blueberry.utils.mobly_sl4a_utils import setup_sl4a
31from blueberry.utils.mobly_sl4a_utils import teardown_sl4a
32from blueberry.tests.gd.cert.context import get_current_context
33from blueberry.tests.gd.cert.gd_device import MOBLY_CONTROLLER_CONFIG_NAME as GD_DEVICE_CONFIG_NAME
34from blueberry.tests.gd_sl4a.lib.ble_lib import enable_bluetooth, disable_bluetooth, BleLib
35from blueberry.facade import rootservice_pb2 as facade_rootservice
36from blueberry.tests.gd.cert import gd_device
37from blueberry.utils.bt_test_utils import clear_bonded_devices
38
39
40class GdSl4aBaseTestClass(BaseTestClass):
41
42    SUBPROCESS_WAIT_TIMEOUT_SECONDS = 10
43
44    def setup_class(self, cert_module):
45        self.log_path_base = get_current_context().get_full_output_path()
46        self.verbose_mode = bool(self.user_params.get('verbose_mode', False))
47        for config in self.controller_configs[GD_DEVICE_CONFIG_NAME]:
48            config['verbose_mode'] = self.verbose_mode
49        self.cert_module = cert_module
50
51        # Parse and construct GD device objects
52        self.gd_devices = self.register_controller(gd_device, required=True)
53        self.cert = self.gd_devices[0]
54
55        # Parse and construct Android device objects
56        self.android_devices = self.register_controller(android_device, required=True)
57        server_port = int(self.controller_configs[ANDROID_DEVICE_COFNIG_NAME][0]['server_port'])
58        forwarded_port = int(self.controller_configs[ANDROID_DEVICE_COFNIG_NAME][0]['forwarded_port'])
59        self.dut = self.android_devices[0]
60        setup_sl4a(self.dut, server_port, forwarded_port)
61
62        # Enable full btsnoop log
63        self.dut.adb.root()
64        self.dut.adb.shell("setprop persist.bluetooth.btsnooplogmode full")
65        getprop_result = self.dut.adb.getprop("persist.bluetooth.btsnooplogmode")
66        if getprop_result is None or ("full" not in getprop_result.lower()):
67            self.dut.log.warning(
68                "Failed to enable Bluetooth Hci Snoop Logging, getprop returned {}".format(getprop_result))
69
70        self.ble = BleLib(dut=self.dut)
71
72    def teardown_class(self):
73        teardown_sl4a(self.dut)
74
75    def setup_test(self):
76        self.cert.rootservice.StartStack(
77            facade_rootservice.StartStackRequest(module_under_test=facade_rootservice.BluetoothModule.Value(
78                self.cert_module),))
79        self.cert.wait_channel_ready()
80
81        self.timer_list = []
82        self.dut.ed.clear_all_events()
83        self.dut.sl4a.setScreenTimeout(500)
84        self.dut.sl4a.wakeUpNow()
85
86        # Always start tests with Bluetooth enabled and BLE disabled.
87        self.dut.sl4a.bluetoothDisableBLE()
88        disable_bluetooth(self.dut.sl4a, self.dut.ed)
89        # Enable full verbose logging for Bluetooth
90        self.dut.adb.shell("setprop log.tag.bluetooth VERBOSE")
91        # Then enable Bluetooth
92        enable_bluetooth(self.dut.sl4a, self.dut.ed)
93        self.dut.sl4a.bluetoothDisableBLE()
94        clear_bonded_devices(self.dut)
95        return True
96
97    def teardown_test(self):
98        clear_bonded_devices(self.dut)
99        # Make sure BLE is disabled and Bluetooth is disabled after test
100        self.dut.sl4a.bluetoothDisableBLE()
101        disable_bluetooth(self.dut.sl4a, self.dut.ed)
102        try:
103            self.cert.rootservice.StopStack(facade_rootservice.StopStackRequest())
104        except Exception:
105            logging.error("Failed to stop CERT stack")
106
107        # TODO: split cert logcat logs into individual tests
108        current_test_dir = get_current_context().get_full_output_path()
109
110        # Pull DUT logs
111        self.pull_dut_logs(current_test_dir)
112
113        # Pull CERT logs
114        self.cert.pull_logs(current_test_dir)
115
116    def pull_dut_logs(self, base_dir):
117        try:
118            self.dut.adb.pull([
119                "/data/misc/bluetooth/logs/btsnoop_hci.log",
120                os.path.join(base_dir, "DUT_%s_btsnoop_hci.log" % self.dut.serial)
121            ])
122            self.dut.adb.pull([
123                "/data/misc/bluedroid/bt_config.conf",
124                os.path.join(base_dir, "DUT_%s_bt_config.conf" % self.dut.serial)
125            ])
126        except AdbError as error:
127            logging.warning("Failed to pull logs from DUT: " + str(error))
128
129    def __getattribute__(self, name):
130        attr = super().__getattribute__(name)
131        if not callable(attr) or not GdSl4aBaseTestClass.__is_entry_function(name):
132            return attr
133
134        @wraps(attr)
135        def __wrapped(*args, **kwargs):
136            try:
137                return attr(*args, **kwargs)
138            except RpcError as e:
139                exception_info = "".join(traceback.format_exception(e.__class__, e, e.__traceback__))
140                raise signals.TestFailure("RpcError during test\n\nRpcError:\n\n%s" % (exception_info))
141
142        return __wrapped
143
144    __ENTRY_METHODS = {"setup_class", "teardown_class", "setup_test", "teardown_test"}
145
146    @staticmethod
147    def __is_entry_function(name):
148        return name.startswith("test_") or name in GdSl4aBaseTestClass.__ENTRY_METHODS
149