1"""Base test class for Blueberry."""
2
3import importlib
4import re
5from typing import Union
6
7from mobly import base_test
8from mobly import records
9from mobly import signals
10from mobly.controllers import android_device
11from mobly.controllers.android_device_lib import adb
12
13from blueberry.controllers import derived_bt_device
14from blueberry.decorators import android_bluetooth_client_decorator
15from blueberry.utils import android_bluetooth_decorator
16
17
18class BlueberryBaseTest(base_test.BaseTestClass):
19  """Base test class for all Blueberry tests to inherit from.
20
21  This class assists with device setup for device logging and other pre test
22  setup required for Bluetooth tests.
23  """
24
25  def __init__(self, configs):
26    super().__init__(configs)
27    self._upload_test_report = None
28    self.capture_bugreport_on_fail = None
29    self.android_devices = None
30    self.derived_bt_devices = None
31    self.ignore_device_setup_failures = None
32    self._test_metrics = []
33
34  def setup_generated_tests(self):
35    """Generates multiple the same tests for pilot run.
36
37    This is used to let developers can easily pilot run their tests many times,
38    help them to check test stability and reliability. If need to use this,
39    please add a flag called "test_iterations" to TestParams in Mobly test
40    configuration and its value is number of test methods to be generated. The
41    naming rule of test method on Sponge is such as the following example:
42      test_send_file_via_bluetooth_opp_1_of_50
43      test_send_file_via_bluetooth_opp_2_of_50
44      test_send_file_via_bluetooth_opp_3_of_50
45      ...
46      test_send_file_via_bluetooth_opp_50_of_50
47
48    Don't use "test_case_selector" when using "test_iterations", and please use
49    "test_method_selector" to replace it.
50    """
51    test_iterations = int(self.user_params.get('test_iterations', 0))
52    if test_iterations < 2:
53      return
54
55    test_method_selector = self.user_params.get('test_method_selector', 'all')
56    existing_test_names = self.get_existing_test_names()
57
58    selected_test_names = None
59    if test_method_selector == 'all':
60      selected_test_names = existing_test_names
61    else:
62      selected_test_names = test_method_selector.split(' ')
63      # Check if selected test methods exist in the test class.
64      for test_name in selected_test_names:
65        if test_name not in existing_test_names:
66          raise base_test.Error('%s does not have test method "%s".' %
67                                (self.TAG, test_name))
68
69    for test_name in selected_test_names:
70      test_method = getattr(self.__class__, test_name)
71      # List of (<new test name>, <test method>).
72      test_arg_sets = [('%s_%s_of_%s' % (test_name, i + 1, test_iterations),
73                        test_method) for i in range(test_iterations)]
74      # pylint: disable=cell-var-from-loop
75      self.generate_tests(
76          test_logic=lambda _, test: test(self),
77          name_func=lambda name, _: name,
78          arg_sets=test_arg_sets)
79
80    # Delete origin test methods in order to avoid below situation:
81    #   test_send_file_via_bluetooth_opp  <-- origin test method
82    #   test_send_file_via_bluetooth_opp_1_of_50
83    #   test_send_file_via_bluetooth_opp_2_of_50
84    for test_name in existing_test_names:
85      delattr(self.__class__, test_name)
86
87  def setup_class(self):
88    """Setup class is called before running any tests."""
89    super(BlueberryBaseTest, self).setup_class()
90    self._upload_test_report = int(self.user_params.get(
91        'upload_test_report', 0))
92    # Inits Spanner Utils if need to upload the test reports to Spanner.
93    if self._upload_test_report:
94      self._init_spanner_utils()
95    self.capture_bugreport_on_fail = int(self.user_params.get(
96        'capture_bugreport_on_fail', 0))
97    self.ignore_device_setup_failures = int(self.user_params.get(
98        'ignore_device_setup_failures', 0))
99    self.enable_bluetooth_verbose_logging = int(self.user_params.get(
100        'enable_bluetooth_verbose_logging', 0))
101    self.enable_hci_snoop_logging = int(self.user_params.get(
102        'enable_hci_snoop_logging', 0))
103    self.increase_logger_buffers = int(self.user_params.get(
104        'increase_logger_buffers', 0))
105    self.enable_all_bluetooth_logging = int(self.user_params.get(
106        'enable_all_bluetooth_logging', 0))
107
108    # base test should include the test between primary device with Bluetooth
109    # peripheral device.
110    self.android_devices = self.register_controller(
111        android_device, required=False)
112
113    # In the case of no android_device assigned, at least 2 derived_bt_device
114    # is required.
115    if self.android_devices is None:
116      self.derived_bt_devices = self.register_controller(
117          module=derived_bt_device, min_number=2)
118    else:
119      self.derived_bt_devices = self.register_controller(
120          module=derived_bt_device, required=False)
121
122    if self.derived_bt_devices is None:
123      self.derived_bt_devices = []
124    else:
125      for derived_device in self.derived_bt_devices:
126        derived_device.set_user_params(self.user_params)
127        derived_device.setup()
128
129    self.android_devices = [
130        android_bluetooth_decorator.AndroidBluetoothDecorator(device)
131        for device in self.android_devices
132    ]
133    for device in self.android_devices:
134      device.set_user_params(self.user_params)
135
136    for device in self.android_devices:
137      need_restart_bluetooth = False
138      if (self.enable_bluetooth_verbose_logging or
139          self.enable_all_bluetooth_logging):
140        if self.set_bt_trc_level_verbose(device):
141          need_restart_bluetooth = True
142      if self.enable_hci_snoop_logging or self.enable_all_bluetooth_logging:
143        if self.set_btsnooplogmode_full(device):
144          need_restart_bluetooth = True
145      if self.increase_logger_buffers or self.enable_all_bluetooth_logging:
146        self.set_logger_buffer_size_16m(device)
147
148      # Restarts Bluetooth to take BT VERBOSE and HCI Snoop logging effect.
149      if need_restart_bluetooth:
150        device.log.info('Restarting Bluetooth by airplane mode...')
151        self.restart_bluetooth_by_airplane_mode(device)
152
153    self.client_decorators = self.user_params.get('sync_decorator', [])
154    if self.client_decorators:
155      self.client_decorators = self.client_decorators.split(',')
156
157    self.target_decorators = self.user_params.get('target_decorator', [])
158    if self.target_decorators:
159      self.target_decorators = self.target_decorators.split(',')
160
161    for decorator in self.client_decorators:
162      self.android_devices[0] = android_bluetooth_client_decorator.decorate(
163          self.android_devices[0], decorator)
164
165    for num_devices in range(1, len(self.android_devices)):
166      for decorator in self.target_decorators:
167        self.android_devices[
168            num_devices] = android_bluetooth_client_decorator.decorate(
169                self.android_devices[num_devices], decorator)
170
171  def on_pass(self, record):
172    """This method is called when a test passed."""
173    if self._upload_test_report:
174      self._upload_test_report_to_spanner(record.result)
175
176  def on_fail(self, record):
177    """This method is called when a test failure."""
178    if self._upload_test_report:
179      self._upload_test_report_to_spanner(record.result)
180
181    # Capture bugreports on fail if enabled.
182    if self.capture_bugreport_on_fail:
183      devices = self.android_devices
184      # Also capture bugreport of AndroidBtTargetDevice.
185      for d in self.derived_bt_devices:
186        if hasattr(d, 'take_bug_report'):
187          devices = devices + [d]
188      android_device.take_bug_reports(
189          devices,
190          record.test_name,
191          record.begin_time,
192          destination=self.current_test_info.output_path)
193
194  def _init_spanner_utils(self) -> None:
195    """Imports spanner_utils and creates SpannerUtils object."""
196    spanner_utils_module = importlib.import_module(
197        'blueberry.utils.spanner_utils')
198    self._spanner_utils = spanner_utils_module.SpannerUtils(
199        test_class_name=self.__class__.__name__,
200        mh_sponge_link=self.user_params['mh_sponge_link'])
201
202  def _upload_test_report_to_spanner(
203      self,
204      result: records.TestResultEnums) -> None:
205    """Uploads the test report to Spanner.
206
207    Args:
208      result: Result of this test.
209    """
210    self._spanner_utils.create_test_report_proto(
211        current_test_info=self.current_test_info,
212        primary_device=self.android_devices[0],
213        companion_devices=self.derived_bt_devices,
214        test_metrics=self._test_metrics)
215    self._test_metrics.clear()
216    test_report = self._spanner_utils.write_test_report_proto(result=result)
217    # Shows the test report on Sponge properties for debugging.
218    self.record_data({
219        'Test Name': self.current_test_info.name,
220        'sponge_properties': {'test_report': test_report},
221    })
222
223  def record_test_metric(
224      self,
225      metric_name: str,
226      metric_value: Union[int, float]) -> None:
227    """Records a test metric to Spanner.
228
229    Args:
230      metric_name: Name of the metric.
231      metric_value: Value of the metric.
232    """
233    if not self._upload_test_report:
234      return
235    self._test_metrics.append(
236        self._spanner_utils.create_metric_proto(metric_name, metric_value))
237
238  def set_logger_buffer_size_16m(self, device):
239    """Sets all logger sizes per log buffer to 16M."""
240    device.log.info('Setting all logger sizes per log buffer to 16M...')
241    # Logger buffer info:
242    # https://developer.android.com/studio/command-line/logcat#alternativeBuffers
243    logger_buffers = ['main', 'system', 'crash', 'radio', 'events', 'kernel']
244    for buffer in logger_buffers:  # pylint: disable=redefined-builtin
245      device.adb.shell('logcat -b %s -G 16M' % buffer)
246      buffer_size = device.adb.shell('logcat -b %s -g' % buffer)
247      if isinstance(buffer_size, bytes):
248        buffer_size = buffer_size.decode()
249      if 'ring buffer is 16' in buffer_size:
250        device.log.info('Successfully set "%s" buffer size to 16M.' % buffer)
251      else:
252        msg = 'Failed to set "%s" buffer size to 16M.' % buffer
253        if not self.ignore_device_setup_failures:
254          raise signals.TestError(msg)
255        device.log.warning(msg)
256
257  def set_bt_trc_level_verbose(self, device):
258    """Modifies etc/bluetooth/bt_stack.conf to enable Bluetooth VERBOSE log."""
259    device.log.info('Enabling Bluetooth VERBOSE logging...')
260    bt_stack_conf = device.adb.shell('cat etc/bluetooth/bt_stack.conf')
261    if isinstance(bt_stack_conf, bytes):
262      bt_stack_conf = bt_stack_conf.decode()
263    # Check if 19 trace level settings are set to 6(VERBOSE). E.g. TRC_HCI=6.
264    if len(re.findall('TRC.*=[6]', bt_stack_conf)) == 19:
265      device.log.info('Bluetooth VERBOSE logging has already enabled.')
266      return False
267    # Suggest to use AndroidDeviceSettingsDecorator to disable verity and then
268    # reboot (b/140277443).
269    device.disable_verity_check()
270    device.adb.remount()
271    try:
272      device.adb.shell(r'sed -i "s/\(TRC.*=\)2/\16/g;s/#\(LoggingV=--v=\)0/\13'
273                       '/" etc/bluetooth/bt_stack.conf')
274      device.log.info('Successfully enabled Bluetooth VERBOSE Logging.')
275      return True
276    except adb.AdbError:
277      msg = 'Failed to enable Bluetooth VERBOSE Logging.'
278      if not self.ignore_device_setup_failures:
279        raise signals.TestError(msg)
280      device.log.warning(msg)
281      return False
282
283  def set_btsnooplogmode_full(self, device):
284    """Enables bluetooth snoop logging."""
285    device.log.info('Enabling Bluetooth HCI Snoop logging...')
286    device.adb.shell('setprop persist.bluetooth.btsnooplogmode full')
287    out = device.adb.shell('getprop persist.bluetooth.btsnooplogmode')
288    if isinstance(out, bytes):
289      out = out.decode()
290    # The expected output is "full/n".
291    if 'full' in out:
292      device.log.info('Successfully enabled Bluetooth HCI Snoop Logging.')
293      return True
294    msg = 'Failed to enable Bluetooth HCI Snoop Logging.'
295    if not self.ignore_device_setup_failures:
296      raise signals.TestError(msg)
297    device.log.warning(msg)
298    return False
299
300  def restart_bluetooth_by_airplane_mode(self, device):
301    """Restarts bluetooth by airplane mode."""
302    device.enable_airplane_mode(3)
303    device.disable_airplane_mode(3)
304