1# Lint as: python2, python3
2# Copyright 2020 The Chromium OS Authors. All rights reserved.
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5
6"""A Batch of Bluetooth AUdio Health tests"""
7
8import time
9import logging
10
11from autotest_lib.client.common_lib import error
12from autotest_lib.client.cros.bluetooth.bluetooth_audio_test_data import (
13        A2DP, A2DP_MEDIUM, A2DP_LONG, AVRCP, HFP_WBS, HFP_NBS, HFP_WBS_MEDIUM,
14        HFP_NBS_MEDIUM)
15from autotest_lib.server.cros.bluetooth.bluetooth_adapter_audio_tests import (
16        BluetoothAdapterAudioTests)
17from autotest_lib.server.cros.bluetooth.bluetooth_adapter_quick_tests import (
18        BluetoothAdapterQuickTests)
19from autotest_lib.client.cros.chameleon.audio_test_utils import (
20        has_internal_speaker)
21
22
23class bluetooth_AdapterAUHealth(BluetoothAdapterQuickTests,
24                                BluetoothAdapterAudioTests):
25    """A Batch of Bluetooth audio health tests."""
26
27    test_wrapper = BluetoothAdapterQuickTests.quick_test_test_decorator
28    batch_wrapper = BluetoothAdapterQuickTests.quick_test_batch_decorator
29
30
31    def au_run_method(self, device, test_method, test_profile):
32        """audio procedure of running a specified test method.
33
34        @param device: the bt peer device
35        @param test_method: the audio test method to run
36        @param test_profile: which test profile is used,
37                             A2DP, HFP_WBS or HFP_NBS
38        """
39        self.test_reset_on_adapter()
40        self.test_bluetoothd_running()
41        self.initialize_bluetooth_audio(device, test_profile)
42        self.test_device_set_discoverable(device, True)
43        self.test_discover_device(device.address)
44        self.test_pairing(device.address, device.pin, trusted=True)
45        self.test_connection_by_adapter(device.address)
46        test_method()
47        self.test_disconnection_by_adapter(device.address)
48        self.cleanup_bluetooth_audio(device, test_profile)
49
50
51    def au_run_test_sequence(self, device, test_sequence, test_profile):
52        """Audio procedure of running a specified test sequence.
53
54        @param device: The Bluetooth peer device.
55        @param test_sequence: The audio test sequence to run.
56        @param test_profile: Which test profile is used,
57                             A2DP, A2DP_MEDIUM, HFP_WBS or HFP_NBS.
58        """
59        # Setup the Bluetooth device.
60        self.test_reset_on_adapter()
61        self.test_bluetoothd_running()
62        self.initialize_bluetooth_audio(device, test_profile)
63
64        test_sequence()
65
66        self.cleanup_bluetooth_audio(device, test_profile)
67
68
69    def _au_a2dp_test(self, test_profile, duration=0):
70        """A2DP test with sinewaves on the two channels.
71
72        @param test_profile: which test profile is used, A2DP or A2DP_LONG.
73        @param duration: the duration to test a2dp. The unit is in seconds.
74                if duration is 0, use the default duration in test_profile.
75        """
76        device = self.devices['BLUETOOTH_AUDIO'][0]
77        self.au_run_method(device,
78                           lambda: self.test_a2dp_sinewaves(
79                                   device, test_profile, duration),
80                           test_profile)
81
82
83    @test_wrapper('A2DP sinewave test',
84                  devices={'BLUETOOTH_AUDIO': 1},
85                  supports_floss=True)
86    def au_a2dp_test(self):
87        """A2DP test with sinewaves on the two channels."""
88        self._au_a2dp_test(A2DP)
89
90    # The A2DP long test is a stress test. Exclude it from the AVL.
91    @test_wrapper('A2DP sinewave long test', devices={'BLUETOOTH_AUDIO':1},
92                  flags=['Quick Health'])
93    def au_a2dp_long_test(self, duration=600):
94        """A2DP long test with sinewaves on the two channels.
95
96        @param duration: the duration to test a2dp. The unit is in seconds.
97        """
98        self._au_a2dp_test(A2DP_LONG, duration=duration)
99
100
101    @test_wrapper('A2DP playback and connect test',
102                  devices={'BLUETOOTH_AUDIO': 1})
103    def au_a2dp_playback_and_connect_test(self):
104        """Connect then disconnect an A2DP device while playing stream."""
105        if not has_internal_speaker(self.host):
106            logging.info('SKIPPING TEST A2DP playback and connect test')
107            raise error.TestNAError(
108                    'The DUT does not have an internal speaker')
109
110        device = self.devices['BLUETOOTH_AUDIO'][0]
111        test_profile = A2DP_MEDIUM
112        test_sequence = lambda: self.playback_and_connect(device, test_profile)
113        self.au_run_test_sequence(device, test_sequence, test_profile)
114
115
116    @test_wrapper('A2DP playback and disconnect test',
117                  devices={'BLUETOOTH_AUDIO': 1})
118    def au_a2dp_playback_and_disconnect_test(self):
119        """Check the playback stream is still alive after BT disconnected."""
120        device = self.devices['BLUETOOTH_AUDIO'][0]
121        test_profile = A2DP_MEDIUM
122        test_sequence = lambda: self.playback_and_disconnect(
123                device, test_profile)
124        self.au_run_test_sequence(device, test_sequence, test_profile)
125
126
127    @test_wrapper('A2DP playback back2back test',
128                  devices={'BLUETOOTH_AUDIO': 1})
129    def au_a2dp_playback_back2back_test(self):
130        """A2DP playback stream back to back test."""
131        device = self.devices['BLUETOOTH_AUDIO'][0]
132        test_profile = A2DP_MEDIUM
133        test_sequence = lambda: self.playback_back2back(device, test_profile)
134        self.au_run_test_sequence(device, test_sequence, test_profile)
135
136
137    @test_wrapper('A2DP pinned playback test', devices={'BLUETOOTH_AUDIO': 1})
138    def au_a2dp_pinned_playback_test(self):
139        """Pinned playback stream test."""
140        device = self.devices['BLUETOOTH_AUDIO'][0]
141        test_profile = A2DP
142        test_sequence = lambda: self.pinned_playback(device, test_profile)
143        self.au_run_test_sequence(device, test_sequence, test_profile)
144
145
146    def au_hfp_run_method(self, device, test_method, test_profile):
147        """Run an HFP test with the specified test method.
148
149        @param device: the bt peer device
150        @param test_method: the specific HFP WBS test method
151        @param test_profile: which test profile is used, HFP_WBS or HFP_NBS
152        """
153        if self.check_wbs_capability():
154            if test_profile in (HFP_WBS, HFP_WBS_MEDIUM):
155                # Restart cras to ensure that cras goes back to the default
156                # selection of either WBS or NBS.
157                # Any board that supports WBS should use WBS by default, unless
158                # it's overridden by CRAS' config.
159                # Do not enable WBS explicitly in the test so we can catch if
160                # the default selection goes wrong.
161                self.restart_cras()
162                # The audio team suggests a simple 2-second sleep.
163                time.sleep(2)
164            elif test_profile in (HFP_NBS, HFP_NBS_MEDIUM):
165                # Cras may be in either WBS or NBS mode. Disable WBS explicitly.
166                if not self.bluetooth_facade.enable_wbs(False):
167                    raise error.TestError('failed to disable wbs')
168        else:
169            if test_profile in (HFP_WBS, HFP_WBS_MEDIUM):
170                # Skip the WBS test on a board that does not support WBS.
171                raise error.TestNAError(
172                        'The DUT does not support WBS. Skip the test.')
173            elif test_profile in (HFP_NBS, HFP_NBS_MEDIUM):
174                # Restart cras to ensure that cras goes back to the default
175                # selection of either WBS or NBS.
176                # Any board that does not support WBS should use NBS by default.
177                # Do not enable NBS explicitly in the test so we can catch if
178                # the default selection goes wrong.
179                self.restart_cras()
180                # The audio team suggests a simple 2-second sleep.
181                time.sleep(2)
182
183        self.au_run_method(
184                device, lambda: test_method(device, test_profile), test_profile)
185
186
187    @test_wrapper('HFP WBS sinewave test with dut as source',
188                  devices={'BLUETOOTH_AUDIO':1})
189    def au_hfp_wbs_dut_as_source_test(self):
190        """HFP WBS test with sinewave streaming from dut to peer."""
191        device = self.devices['BLUETOOTH_AUDIO'][0]
192        self.au_hfp_run_method(device, self.hfp_dut_as_source, HFP_WBS)
193
194
195    @test_wrapper('HFP WBS sinewave test with dut as sink',
196                  devices={'BLUETOOTH_AUDIO':1})
197    def au_hfp_wbs_dut_as_sink_test(self):
198        """HFP WBS test with sinewave streaming from peer to dut."""
199        device = self.devices['BLUETOOTH_AUDIO'][0]
200        self.au_hfp_run_method(device, self.hfp_dut_as_sink, HFP_WBS)
201
202
203    @test_wrapper('HFP NBS sinewave test with dut as source',
204                  devices={'BLUETOOTH_AUDIO': 1},
205                  supports_floss=True)
206    def au_hfp_nbs_dut_as_source_test(self):
207        """HFP NBS test with sinewave streaming from dut to peer."""
208        device = self.devices['BLUETOOTH_AUDIO'][0]
209        self.au_hfp_run_method(device, self.hfp_dut_as_source, HFP_NBS)
210
211
212    @test_wrapper('HFP NBS sinewave test with dut as sink',
213                  devices={'BLUETOOTH_AUDIO': 1},
214                  supports_floss=True)
215    def au_hfp_nbs_dut_as_sink_test(self):
216        """HFP NBS test with sinewave streaming from peer to dut."""
217        device = self.devices['BLUETOOTH_AUDIO'][0]
218        self.au_hfp_run_method(device, self.hfp_dut_as_sink, HFP_NBS)
219
220
221    @test_wrapper('HFP WBS VISQOL test with dut as sink',
222                  devices={'BLUETOOTH_AUDIO':1})
223    def au_hfp_wbs_dut_as_sink_visqol_test(self):
224        """HFP WBS VISQOL test with audio streaming from peer to dut"""
225        device = self.devices['BLUETOOTH_AUDIO'][0]
226        self.au_hfp_run_method(device, self.hfp_dut_as_sink_visqol_score,
227                               HFP_WBS)
228
229
230    @test_wrapper('HFP WBS VISQOL test with dut as source',
231                  devices={'BLUETOOTH_AUDIO':1})
232    def au_hfp_wbs_dut_as_source_visqol_test(self):
233        """HFP WBS VISQOL test with audio streaming from dut to peer"""
234        device = self.devices['BLUETOOTH_AUDIO'][0]
235        self.au_hfp_run_method(device, self.hfp_dut_as_source_visqol_score,
236                               HFP_WBS)
237
238    @test_wrapper('HFP NBS VISQOL test with dut as sink',
239                  devices={'BLUETOOTH_AUDIO':1})
240    def au_hfp_nbs_dut_as_sink_visqol_test(self):
241        """HFP NBS VISQOL test with audio streaming from peer to dut"""
242        device = self.devices['BLUETOOTH_AUDIO'][0]
243        self.au_hfp_run_method(device, self.hfp_dut_as_sink_visqol_score,
244                               HFP_NBS)
245
246
247    @test_wrapper('HFP NBS VISQOL test with dut as source',
248                  devices={'BLUETOOTH_AUDIO':1})
249    def au_hfp_nbs_dut_as_source_visqol_test(self):
250        """HFP NBS VISQOL test with audio streaming from dut to peer"""
251        device = self.devices['BLUETOOTH_AUDIO'][0]
252        self.au_hfp_run_method(device, self.hfp_dut_as_source_visqol_score,
253                               HFP_NBS)
254
255
256    @test_wrapper('HFP NBS back2back test with dut as source',
257                  devices={'BLUETOOTH_AUDIO': 1})
258    def au_hfp_nbs_dut_as_source_back2back_test(self):
259        """HFP NBS back2back test from dut to peer"""
260        device = self.devices['BLUETOOTH_AUDIO'][0]
261        self.au_hfp_run_method(device, self.hfp_dut_as_source_back2back,
262                               HFP_NBS)
263
264
265    @test_wrapper('HFP WBS back2back test with dut as source',
266                  devices={'BLUETOOTH_AUDIO': 1})
267    def au_hfp_wbs_dut_as_source_back2back_test(self):
268        """HFP WBS back2back test from dut to peer"""
269        device = self.devices['BLUETOOTH_AUDIO'][0]
270        self.au_hfp_run_method(device, self.hfp_dut_as_source_back2back,
271                               HFP_WBS)
272
273
274    @test_wrapper('Switch A2DP to HFP NBS test with dut as source',
275                  devices={'BLUETOOTH_AUDIO': 1})
276    def au_a2dp_to_hfp_nbs_dut_as_source_test(self):
277        """Switch A2DP to HFP NBS test with dut as source."""
278        device = self.devices['BLUETOOTH_AUDIO'][0]
279        self.au_hfp_run_method(device, self.a2dp_to_hfp_dut_as_source,
280                               HFP_NBS_MEDIUM)
281
282
283    @test_wrapper('Switch A2DP to HFP WBS test with dut as source',
284                  devices={'BLUETOOTH_AUDIO': 1})
285    def au_a2dp_to_hfp_wbs_dut_as_source_test(self):
286        """Switch A2DP to HFP WBS test with dut as source."""
287        device = self.devices['BLUETOOTH_AUDIO'][0]
288        self.au_hfp_run_method(device, self.a2dp_to_hfp_dut_as_source,
289                               HFP_WBS_MEDIUM)
290
291
292    @test_wrapper('Switch HFP NBS to A2DP test with dut as source',
293                  devices={'BLUETOOTH_AUDIO': 1})
294    def au_hfp_nbs_to_a2dp_dut_as_source_test(self):
295        """Switch HFP NBS to A2DP test with dut as source."""
296        device = self.devices['BLUETOOTH_AUDIO'][0]
297        self.au_hfp_run_method(device, self.hfp_to_a2dp_dut_as_source,
298                               HFP_NBS_MEDIUM)
299
300
301    @test_wrapper('Switch HFP WBS to A2DP test with dut as source',
302                  devices={'BLUETOOTH_AUDIO': 1})
303    def au_hfp_wbs_to_a2dp_dut_as_source_test(self):
304        """Switch HFP WBS to A2DP test with dut as source."""
305        device = self.devices['BLUETOOTH_AUDIO'][0]
306        self.au_hfp_run_method(device, self.hfp_to_a2dp_dut_as_source,
307                               HFP_WBS_MEDIUM)
308
309
310    def au_run_avrcp_method(self, device, test_method):
311        """avrcp procedure of running a specified test method.
312
313        @param device: the bt peer device
314        @param test_method: the avrcp test method to run
315        """
316        def wrapped_test_method(device):
317            """A wrapper method to initialize and cleanup avrcp tests.
318
319            @param device: the bt peer device
320            """
321            self.initialize_bluetooth_player(device)
322            test_method(device)
323            self.cleanup_bluetooth_player(device)
324
325        self.au_run_method(
326                device, lambda: wrapped_test_method(device), AVRCP)
327
328
329    @test_wrapper('avrcp command test', devices={'BLUETOOTH_AUDIO':1})
330    def au_avrcp_command_test(self):
331        """AVRCP test to examine commands reception."""
332        device = self.devices['BLUETOOTH_AUDIO'][0]
333        self.au_run_avrcp_method(device, self.test_avrcp_commands)
334
335
336    @test_wrapper('avrcp media info test', devices={'BLUETOOTH_AUDIO': 1})
337    def au_avrcp_media_info_test(self):
338        """AVRCP test to examine metadata propgation."""
339        device = self.devices['BLUETOOTH_AUDIO'][0]
340        self.au_run_avrcp_method(device, self.test_avrcp_media_info)
341
342
343    @batch_wrapper('Bluetooth Audio Batch Health Tests')
344    def au_health_batch_run(self, num_iterations=1, test_name=None):
345        """Run the bluetooth audio health test batch or a specific given test.
346
347        @param num_iterations: how many iterations to run
348        @param test_name: specific test to run otherwise None to run the
349                whole batch
350        """
351        self.au_a2dp_test()
352        self.au_a2dp_long_test()
353        self.au_hfp_nbs_dut_as_source_test()
354        self.au_hfp_nbs_dut_as_sink_test()
355        self.au_hfp_wbs_dut_as_source_test()
356        self.au_hfp_wbs_dut_as_sink_test()
357        self.au_hfp_wbs_dut_as_source_visqol_test()
358        self.au_hfp_wbs_dut_as_sink_visqol_test()
359        self.au_hfp_nbs_dut_as_source_visqol_test()
360        self.au_hfp_nbs_dut_as_sink_visqol_test()
361        self.au_avrcp_command_test()
362        self.au_avrcp_media_info_test()
363        self.au_a2dp_playback_and_connect_test()
364        self.au_a2dp_playback_and_disconnect_test()
365        self.au_a2dp_playback_back2back_test()
366        self.au_a2dp_pinned_playback_test()
367        self.au_hfp_nbs_dut_as_source_back2back_test()
368        self.au_hfp_wbs_dut_as_source_back2back_test()
369        self.au_a2dp_to_hfp_nbs_dut_as_source_test()
370        self.au_a2dp_to_hfp_wbs_dut_as_source_test()
371        self.au_hfp_nbs_to_a2dp_dut_as_source_test()
372        self.au_hfp_wbs_to_a2dp_dut_as_source_test()
373
374
375    def run_once(self,
376                 host,
377                 num_iterations=1,
378                 args_dict=None,
379                 test_name=None,
380                 flag='Quick Health',
381                 floss=False):
382        """Run the batch of Bluetooth stand health tests
383
384        @param host: the DUT, usually a chromebook
385        @param num_iterations: the number of rounds to execute the test
386        @param test_name: the test to run, or None for all tests
387        """
388        self.host = host
389
390        self.quick_test_init(host,
391                             use_btpeer=True,
392                             flag=flag,
393                             args_dict=args_dict,
394                             floss=floss)
395        self.au_health_batch_run(num_iterations, test_name)
396        self.quick_test_cleanup()
397