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"""Server side Bluetooth audio tests.""" 7 8from __future__ import absolute_import 9from __future__ import division 10from __future__ import print_function 11 12import logging 13import os 14import re 15import subprocess 16import time 17 18import common 19from autotest_lib.client.bin import utils 20from autotest_lib.client.common_lib import error 21from autotest_lib.client.cros.bluetooth.bluetooth_audio_test_data import ( 22 A2DP, HFP_NBS, HFP_NBS_MEDIUM, HFP_WBS, HFP_WBS_MEDIUM, 23 AUDIO_DATA_TARBALL_PATH, VISQOL_BUFFER_LENGTH, DATA_DIR, VISQOL_PATH, 24 VISQOL_SIMILARITY_MODEL, VISQOL_TEST_DIR, AUDIO_RECORD_DIR, 25 audio_test_data, get_audio_test_data, get_visqol_binary) 26from autotest_lib.server.cros.bluetooth.bluetooth_adapter_tests import ( 27 BluetoothAdapterTests, test_retry_and_log) 28from six.moves import range 29 30 31class BluetoothAdapterAudioTests(BluetoothAdapterTests): 32 """Server side Bluetooth adapter audio test class.""" 33 34 DEVICE_TYPE = 'BLUETOOTH_AUDIO' 35 FREQUENCY_TOLERANCE_RATIO = 0.01 36 WAIT_DAEMONS_READY_SECS = 1 37 DEFAULT_CHUNK_IN_SECS = 1 38 IGNORE_LAST_FEW_CHUNKS = 2 39 40 # Useful constant for upsampling NBS files for compatibility with ViSQOL 41 MIN_VISQOL_SAMPLE_RATE = 16000 42 43 # The node types of the bluetooth output nodes in cras are the same for both 44 # A2DP and HFP. 45 CRAS_BLUETOOTH_OUTPUT_NODE_TYPE = 'BLUETOOTH' 46 CRAS_INTERNAL_SPEAKER_OUTPUT_NODE_TYPE = 'INTERNAL_SPEAKER' 47 # The node types of the bluetooth input nodes in cras are different for WBS 48 # and NBS. 49 CRAS_HFP_BLUETOOTH_INPUT_NODE_TYPE = {HFP_WBS: 'BLUETOOTH', 50 HFP_NBS: 'BLUETOOTH_NB_MIC'} 51 52 def _get_pulseaudio_bluez_source(self, get_source_method, device, 53 test_profile): 54 """Get the specified bluez device number in the pulseaudio source list. 55 56 @param get_source_method: the method to get distinct bluez source 57 @param device: the bluetooth peer device 58 @param test_profile: the test profile used, A2DP, HFP_WBS or HFP_NBS 59 60 @returns: True if the specified bluez source is derived 61 """ 62 sources = device.ListSources(test_profile) 63 logging.debug('ListSources()\n%s', sources) 64 self.bluez_source = get_source_method(test_profile) 65 result = bool(self.bluez_source) 66 if result: 67 logging.debug('bluez_source device number: %s', self.bluez_source) 68 else: 69 logging.debug('waiting for bluez_source ready in pulseaudio...') 70 return result 71 72 73 def _get_pulseaudio_bluez_sink(self, get_sink_method, device, test_profile): 74 """Get the specified bluez device number in the pulseaudio sink list. 75 76 @param get_sink_method: the method to get distinct bluez sink 77 @param device: the bluetooth peer device 78 @param test_profile: the test profile used, A2DP, HFP_WBS or HFP_NBS 79 80 @returns: True if the specified bluez sink is derived 81 """ 82 sinks = device.ListSinks(test_profile) 83 logging.debug('ListSinks()\n%s', sinks) 84 self.bluez_sink = get_sink_method(test_profile) 85 result = bool(self.bluez_sink) 86 if result: 87 logging.debug('bluez_sink device number: %s', self.bluez_sink) 88 else: 89 logging.debug('waiting for bluez_sink ready in pulseaudio...') 90 return result 91 92 93 def _get_pulseaudio_bluez_source_a2dp(self, device, test_profile): 94 """Get the a2dp bluez source device number. 95 96 @param device: the bluetooth peer device 97 @param test_profile: the test profile used, A2DP, HFP_WBS or HFP_NBS 98 99 @returns: True if the specified a2dp bluez source is derived 100 """ 101 return self._get_pulseaudio_bluez_source( 102 device.GetBluezSourceA2DPDevice, device, test_profile) 103 104 105 def _get_pulseaudio_bluez_source_hfp(self, device, test_profile): 106 """Get the hfp bluez source device number. 107 108 @param device: the bluetooth peer device 109 @param test_profile: the test profile used, A2DP, HFP_WBS or HFP_NBS 110 111 @returns: True if the specified hfp bluez source is derived 112 """ 113 return self._get_pulseaudio_bluez_source( 114 device.GetBluezSourceHFPDevice, device, test_profile) 115 116 117 def _get_pulseaudio_bluez_sink_hfp(self, device, test_profile): 118 """Get the hfp bluez sink device number. 119 120 @param device: the bluetooth peer device 121 @param test_profile: the test profile used, A2DP, HFP_WBS or HFP_NBS 122 123 @returns: True if the specified hfp bluez sink is derived 124 """ 125 return self._get_pulseaudio_bluez_sink( 126 device.GetBluezSinkHFPDevice, device, test_profile) 127 128 129 def _check_audio_frames_legitimacy(self, audio_test_data, recording_device, 130 recorded_file=None): 131 """Check if audio frames in the recorded file are legitimate. 132 133 For a wav file, a simple check is to make sure the recorded audio file 134 is not empty. 135 136 For a raw file, a simple check is to make sure the recorded audio file 137 are not all zeros. 138 139 @param audio_test_data: a dictionary about the audio test data 140 defined in client/cros/bluetooth/bluetooth_audio_test_data.py 141 @param recording_device: which device recorded the audio, 142 possible values are 'recorded_by_dut' or 'recorded_by_peer' 143 @param recorded_file: the recorded file name 144 145 @returns: True if audio frames are legitimate. 146 """ 147 result = self.bluetooth_facade.check_audio_frames_legitimacy( 148 audio_test_data, recording_device, recorded_file) 149 if not result: 150 self.results = {'audio_frames_legitimacy': 'empty or all zeros'} 151 logging.error('The recorded audio file is empty or all zeros.') 152 return result 153 154 155 def _check_frequency(self, test_profile, recorded_freq, expected_freq): 156 """Check if the recorded frequency is within tolerance. 157 158 @param test_profile: the test profile used, A2DP, HFP_WBS or HFP_NBS 159 @param recorded_freq: the frequency of recorded audio 160 @param expected_freq: the expected frequency 161 162 @returns: True if the recoreded frequency falls within the tolerance of 163 the expected frequency 164 """ 165 tolerance = expected_freq * self.FREQUENCY_TOLERANCE_RATIO 166 return abs(expected_freq - recorded_freq) <= tolerance 167 168 169 def _check_primary_frequencies(self, test_profile, audio_test_data, 170 recording_device, recorded_file=None): 171 """Check if the recorded frequencies meet expectation. 172 173 @param test_profile: the test profile used, A2DP, HFP_WBS or HFP_NBS 174 @param audio_test_data: a dictionary about the audio test data 175 defined in client/cros/bluetooth/bluetooth_audio_test_data.py 176 @param recording_device: which device recorded the audio, 177 possible values are 'recorded_by_dut' or 'recorded_by_peer' 178 @param recorded_file: the recorded file name 179 180 @returns: True if the recorded frequencies of all channels fall within 181 the tolerance of expected frequencies 182 """ 183 recorded_frequencies = self.bluetooth_facade.get_primary_frequencies( 184 audio_test_data, recording_device, recorded_file) 185 expected_frequencies = audio_test_data['frequencies'] 186 final_result = True 187 self.results = dict() 188 189 if len(recorded_frequencies) < len(expected_frequencies): 190 logging.error('recorded_frequencies: %s, expected_frequencies: %s', 191 str(recorded_frequencies), str(expected_frequencies)) 192 final_result = False 193 else: 194 for channel, expected_freq in enumerate(expected_frequencies): 195 recorded_freq = recorded_frequencies[channel] 196 ret_val = self._check_frequency( 197 test_profile, recorded_freq, expected_freq) 198 pass_fail_str = 'pass' if ret_val else 'fail' 199 result = ('primary frequency %d (expected %d): %s' % 200 (recorded_freq, expected_freq, pass_fail_str)) 201 self.results['Channel %d' % channel] = result 202 logging.info('Channel %d: %s', channel, result) 203 204 if not ret_val: 205 final_result = False 206 207 logging.debug(str(self.results)) 208 if not final_result: 209 logging.error('Failure at checking primary frequencies') 210 return final_result 211 212 213 def _poll_for_condition(self, condition, timeout=20, sleep_interval=1, 214 desc='waiting for condition'): 215 try: 216 utils.poll_for_condition(condition=condition, 217 timeout=timeout, 218 sleep_interval=sleep_interval, 219 desc=desc) 220 except Exception as e: 221 raise error.TestError('Exception occurred when %s (%s)' % (desc, e)) 222 223 def _scp_to_dut(self, device, src_file, dest_file): 224 """SCP file from peer device to DuT.""" 225 ip = self.host.ip 226 # Localhost is unlikely to be the correct ip target so take the local 227 # host ip if it exists. 228 if self.host.ip == '127.0.0.1' and self.local_host_ip: 229 ip = self.local_host_ip 230 logging.info('Using local host ip = %s', ip) 231 232 device.ScpToDut(src_file, dest_file, ip) 233 234 def check_wbs_capability(self): 235 """Check if the DUT supports WBS capability. 236 237 @raises: TestNAError if the dut does not support wbs. 238 """ 239 capabilities, err = self.bluetooth_facade.get_supported_capabilities() 240 return err is None and bool(capabilities.get('wide band speech')) 241 242 243 def initialize_bluetooth_audio(self, device, test_profile): 244 """Initialize the Bluetooth audio task. 245 246 Note: pulseaudio is not stable. Need to restart it in the beginning. 247 248 @param device: the bluetooth peer device 249 @param test_profile: the test profile used, A2DP, HFP_WBS or HFP_NBS 250 251 """ 252 if not self.bluetooth_facade.create_audio_record_directory( 253 AUDIO_RECORD_DIR): 254 raise error.TestError('Failed to create %s on the DUT' % 255 AUDIO_RECORD_DIR) 256 257 if not device.StartPulseaudio(test_profile): 258 raise error.TestError('Failed to start pulseaudio.') 259 logging.debug('pulseaudio is started.') 260 261 if test_profile in (HFP_WBS, HFP_NBS, HFP_NBS_MEDIUM, HFP_WBS_MEDIUM): 262 if device.StartOfono(): 263 logging.debug('ofono is started.') 264 else: 265 raise error.TestError('Failed to start ofono.') 266 elif device.StopOfono(): 267 logging.debug('ofono is stopped.') 268 else: 269 logging.warning('Failed to stop ofono. Ignored.') 270 271 # Need time to complete starting services. 272 time.sleep(self.WAIT_DAEMONS_READY_SECS) 273 274 275 def cleanup_bluetooth_audio(self, device, test_profile): 276 """Cleanup for Bluetooth audio. 277 278 @param device: the bluetooth peer device 279 @param test_profile: the test profile used, A2DP, HFP_WBS or HFP_NBS 280 281 """ 282 if device.StopPulseaudio(): 283 logging.debug('pulseaudio is stopped.') 284 else: 285 logging.warning('Failed to stop pulseaudio. Ignored.') 286 287 if device.StopOfono(): 288 logging.debug('ofono is stopped.') 289 else: 290 logging.warning('Failed to stop ofono. Ignored.') 291 292 293 def initialize_bluetooth_player(self, device): 294 """Initialize the Bluetooth media player. 295 296 @param device: the Bluetooth peer device. 297 298 """ 299 if not device.ExportMediaPlayer(): 300 raise error.TestError('Failed to export media player.') 301 logging.debug('mpris-proxy is started.') 302 303 # Wait for player to show up and observed by playerctl. 304 desc='waiting for media player' 305 self._poll_for_condition( 306 lambda: bool(device.GetExportedMediaPlayer()), desc=desc) 307 308 309 def cleanup_bluetooth_player(self, device): 310 """Cleanup for Bluetooth media player. 311 312 @param device: the bluetooth peer device. 313 314 """ 315 device.UnexportMediaPlayer() 316 317 318 def parse_visqol_output(self, stdout, stderr): 319 """ 320 Parse stdout and stderr string from VISQOL output and parse into 321 a float score. 322 323 On error, stderr will contain the error message, otherwise will be None. 324 On success, stdout will be a string, first line will be 325 VISQOL version, followed by indication of speech mode. Followed by 326 paths to reference and degraded file, and a float MOS-LQO score, which 327 is what we're interested in. Followed by more detailed charts about 328 specific scoring by segments of the files. Stdout is None on error. 329 330 @param stdout: The stdout bytes from commandline output of VISQOL. 331 @param stderr: The stderr bytes from commandline output of VISQOL. 332 333 @returns: A tuple of a float score and string representation of the 334 srderr or None if there was no error. 335 """ 336 string_out = stdout.decode('utf-8') or '' 337 stderr = stderr.decode('utf-8') 338 339 # Log verbose VISQOL output: 340 log_file = os.path.join(VISQOL_TEST_DIR, 'VISQOL_LOG.txt') 341 with open(log_file, 'w+') as f: 342 f.write('String Error:\n{}\n'.format(stderr)) 343 f.write('String Out:\n{}\n'.format(string_out)) 344 345 # pattern matches first float or int after 'MOS-LQO:' in stdout, 346 # e.g. it would match the line 'MOS-LQO 2.3' in the stdout 347 score_pattern = re.compile(r'.*MOS-LQO:\s*(\d+.?\d*)') 348 score_search = re.search(score_pattern, string_out) 349 350 # re.search returns None if no pattern match found, otherwise the score 351 # would be in the match object's group 1 matches just the float score 352 score = float(score_search.group(1)) if score_search else -1.0 353 return stderr, score 354 355 356 def get_visqol_score(self, ref_file, deg_file, speech_mode=True, 357 verbose=True): 358 """ 359 Runs VISQOL using the subprocess library on the provided reference file 360 and degraded file and returns the VISQOL score. 361 362 @param ref_file: File path to the reference wav file. 363 @param deg_file: File path to the degraded wav file. 364 @param speech_mode: [Optional] Defaults to True, accepts 16k sample 365 rate files and ignores frequencies > 8kHz for scoring. 366 @param verbose: [Optional] Defaults to True, outputs more details. 367 368 @returns: A float score for the tested file. 369 """ 370 visqol_cmd = [VISQOL_PATH] 371 visqol_cmd += ['--reference_file', ref_file] 372 visqol_cmd += ['--degraded_file', deg_file] 373 visqol_cmd += ['--similarity_to_quality_model', VISQOL_SIMILARITY_MODEL] 374 375 if speech_mode: 376 visqol_cmd.append('--use_speech_mode') 377 if verbose: 378 visqol_cmd.append('--verbose') 379 380 visqol_process = subprocess.Popen(visqol_cmd, stdout=subprocess.PIPE, 381 stderr=subprocess.PIPE) 382 stdout, stderr = visqol_process.communicate() 383 384 err, score = self.parse_visqol_output(stdout, stderr) 385 386 if err: 387 raise error.TestError(err) 388 elif score < 0.0: 389 raise error.TestError('Failed to parse score, got {}'.format(score)) 390 391 return score 392 393 394 def get_ref_and_deg_files(self, trimmed_file, test_profile, test_data): 395 """Return path for reference and degraded files to run visqol on. 396 397 @param trimmed_file: Path to the trimmed audio file on DUT. 398 @param test_profile: The test profile used HFP_WBS or HFP_NBS. 399 @param test_data: A dictionary about the audio test data defined in 400 client/cros/bluetooth/bluetooth_audio_test_data.py. 401 402 @returns: A tuple of path to the reference file and degraded file if 403 they exist, otherwise False for the files that aren't available. 404 """ 405 # Path in autotest server in ViSQOL folder to store degraded file from 406 # retrieved from the DUT 407 deg_file = os.path.join(VISQOL_TEST_DIR, os.path.split(trimmed_file)[1]) 408 played_file = test_data['file'] 409 # If profile is WBS, no resampling required 410 if test_profile == HFP_WBS: 411 self.host.get_file(trimmed_file, deg_file) 412 return played_file, deg_file 413 414 # On NBS, degraded and reference files need to be resampled to 16 kHz 415 # Build path for the upsampled (us) reference (ref) file on DUT 416 ref_file = '{}_us_ref{}'.format(*os.path.splitext(played_file)) 417 # If resampled ref file already exists, don't need to do it again 418 if not os.path.isfile(ref_file): 419 if not self.bluetooth_facade.convert_audio_sample_rate( 420 played_file, ref_file, test_data, 421 self.MIN_VISQOL_SAMPLE_RATE): 422 return False, False 423 # Move upsampled reference file to autotest server 424 self.host.get_file(ref_file, ref_file) 425 426 # Build path for resampled degraded file on DUT 427 deg_on_dut = '{}_us{}'.format(*os.path.splitext(trimmed_file)) 428 # Resample degraded file to 16 kHz and move to autotest server 429 if not self.bluetooth_facade.convert_audio_sample_rate( 430 trimmed_file, deg_on_dut, test_data, 431 self.MIN_VISQOL_SAMPLE_RATE): 432 return ref_file, False 433 434 self.host.get_file(deg_on_dut, deg_file) 435 436 return ref_file, deg_file 437 438 439 def format_recorded_file(self, test_data, test_profile, recording_device): 440 """Format recorded files to be compatible with ViSQOL. 441 442 Convert raw files to wav if recorded file is a raw file, trim file to 443 duration, if required, resample the file, then lastly return the paths 444 for the reference file and degraded file on the autotest server. 445 446 @param test_data: A dictionary about the audio test data defined in 447 client/cros/bluetooth/bluetooth_audio_test_data.py. 448 @param test_profile: The test profile used, HFP_WBS or HFP_NBS. 449 @param recording_device: Which device recorded the audio, either 450 'recorded_by_dut' or 'recorded_by_peer'. 451 452 @returns: A tuple of path to the reference file and degraded file if 453 they exist, otherwise False for the files that aren't available. 454 """ 455 # Path to recorded file either on DUT or BT peer 456 recorded_file = test_data[recording_device] 457 untrimmed_file = recorded_file 458 if recorded_file.endswith('.raw'): 459 # build path for file converted from raw to wav, i.e. change the ext 460 untrimmed_file = os.path.splitext(recorded_file)[0] + '.wav' 461 if not self.bluetooth_facade.convert_raw_to_wav( 462 recorded_file, untrimmed_file, test_data): 463 raise error.TestError('Could not convert raw file to wav') 464 465 # Compute the duration of played file without added buffer 466 new_duration = (test_data['chunk_checking_duration'] - 467 VISQOL_BUFFER_LENGTH) 468 # build path for file resulting from trimming to desired duration 469 trimmed_file = '{}_t{}'.format(*os.path.splitext(untrimmed_file)) 470 if not self.bluetooth_facade.trim_wav_file( 471 untrimmed_file, trimmed_file, new_duration, test_data): 472 raise error.TestError('Failed to trim recorded file') 473 474 return self.get_ref_and_deg_files(trimmed_file, test_profile, test_data) 475 476 477 def handle_one_chunk(self, device, chunk_in_secs, index, test_profile): 478 """Handle one chunk of audio data by calling chameleon api.""" 479 480 ip = self.host.ip 481 # Localhost is unlikely to be the correct ip target so take the local 482 # host ip if it exists. 483 if self.host.ip == '127.0.0.1' and self.local_host_ip: 484 ip = self.local_host_ip 485 logging.info('Using local host ip = %s', ip) 486 487 # TODO(b/207046142): Remove the old version fallback after the new 488 # Chameleon bundle is deployed. 489 try: 490 recorded_file = device.HandleOneChunk(chunk_in_secs, index, ip) 491 except Exception as e: 492 logging.debug("Unable to use new version of HandleOneChunk;" 493 "fall back to use the old one.") 494 try: 495 recorded_file = device.HandleOneChunk(chunk_in_secs, index, 496 test_profile, ip) 497 except Exception as e: 498 raise error.TestError('Failed to handle chunk (%s)', e) 499 500 return recorded_file 501 502 503 # --------------------------------------------------------------- 504 # Definitions of all bluetooth audio test cases 505 # --------------------------------------------------------------- 506 507 508 @test_retry_and_log(False) 509 def test_select_audio_input_device(self, device_name): 510 """Select the audio input device for the DUT. 511 512 @param: device_name: the audio input device to be selected. 513 514 @returns: True on success. Raise otherwise. 515 """ 516 desc = 'waiting for cras to select audio input device' 517 logging.debug(desc) 518 self._poll_for_condition( 519 lambda: self.bluetooth_facade.select_input_device(device_name), 520 desc=desc) 521 return True 522 523 524 @test_retry_and_log(False) 525 def test_select_audio_output_node_bluetooth(self): 526 """Select the Bluetooth device as output node. 527 528 @returns: True on success. False otherwise. 529 """ 530 return self._test_select_audio_output_node( 531 self.CRAS_BLUETOOTH_OUTPUT_NODE_TYPE) 532 533 534 @test_retry_and_log(False) 535 def test_select_audio_output_node_internal_speaker(self): 536 """Select the internal speaker as output node. 537 538 @returns: True on success. False otherwise. 539 """ 540 return self._test_select_audio_output_node( 541 self.CRAS_INTERNAL_SPEAKER_OUTPUT_NODE_TYPE) 542 543 544 def _test_select_audio_output_node(self, node_type=None): 545 """Select the audio output node through cras. 546 547 @param node_type: a str representing node type defined in 548 CRAS_NODE_TYPES. 549 @raises: error.TestError if failed. 550 551 @return True if select given node success. 552 """ 553 def node_type_selected(node_type): 554 """Check if the given node type is selected.""" 555 selected = self.bluetooth_facade.get_selected_output_device_type() 556 logging.debug('active output node type: %s, expected %s', selected, 557 node_type) 558 return selected == node_type 559 560 desc = 'waiting for bluetooth_facade.select_output_node()' 561 self._poll_for_condition( 562 lambda: self.bluetooth_facade.select_output_node(node_type), 563 desc=desc) 564 565 desc = 'waiting for %s as active cras audio output node type' % node_type 566 logging.debug(desc) 567 self._poll_for_condition(lambda: node_type_selected(node_type), 568 desc=desc) 569 570 return True 571 572 573 @test_retry_and_log(False) 574 def test_audio_is_alive_on_dut(self): 575 """Test that if the audio stream is alive on the DUT. 576 577 @returns: True if the audio summary is found on the DUT. 578 """ 579 summary = self.bluetooth_facade.get_audio_thread_summary() 580 result = bool(summary) 581 582 # If we can find something starts with summary like: "Summary: Output 583 # device [Silent playback device.] 4096 48000 2 Summary: Output stream 584 # CRAS_CLIENT_TYPE_TEST CRAS_STREAM_TYPE_DEFAULT 480 240 0x0000 48000 585 # 2 0" this means that there's an audio stream alive on the DUT. 586 desc = " ".join(str(line) for line in summary) 587 logging.debug('find summary: %s', desc) 588 589 self.results = {'test_audio_is_alive_on_dut': result} 590 return all(self.results.values()) 591 592 593 @test_retry_and_log(False) 594 def test_check_chunks(self, 595 device, 596 test_profile, 597 test_data, 598 duration, 599 check_legitimacy=True, 600 check_frequencies=True): 601 """Check chunks of recorded streams and verify the primary frequencies. 602 603 @param device: the bluetooth peer device 604 @param test_profile: the a2dp test profile; 605 choices are A2DP and A2DP_LONG 606 @param test_data: the test data of the test profile 607 @param duration: the duration of the audio file to test 608 @param check_legitimacy: specify this to True to run 609 _check_audio_frames_legitimacy test 610 @param check_frequencies: specify this to True to run 611 _check_primary_frequencies test 612 613 @returns: True if all chunks pass the frequencies check. 614 """ 615 chunk_in_secs = test_data['chunk_in_secs'] 616 if not bool(chunk_in_secs): 617 chunk_in_secs = self.DEFAULT_CHUNK_IN_SECS 618 nchunks = duration // chunk_in_secs 619 logging.info('Number of chunks: %d', nchunks) 620 621 check_audio_frames_legitimacy = True 622 check_primary_frequencies = True 623 for i in range(nchunks): 624 logging.debug('Check chunk %d', i) 625 626 recorded_file = self.handle_one_chunk(device, chunk_in_secs, i, 627 test_profile) 628 if recorded_file is None: 629 raise error.TestError('Failed to handle chunk %d' % i) 630 631 if check_legitimacy: 632 # Check if the audio frames in the recorded file are legitimate. 633 if not self._check_audio_frames_legitimacy( 634 test_data, 'recorded_by_peer', recorded_file=recorded_file): 635 if (i > self.IGNORE_LAST_FEW_CHUNKS and 636 i >= nchunks - self.IGNORE_LAST_FEW_CHUNKS): 637 logging.info('empty chunk %d ignored for last %d chunks', 638 i, self.IGNORE_LAST_FEW_CHUNKS) 639 else: 640 check_audio_frames_legitimacy = False 641 break 642 643 if check_frequencies: 644 # Check if the primary frequencies of the recorded file 645 # meet expectation. 646 if not self._check_primary_frequencies( 647 test_profile, 648 test_data, 649 'recorded_by_peer', 650 recorded_file=recorded_file): 651 if (i > self.IGNORE_LAST_FEW_CHUNKS and 652 i >= nchunks - self.IGNORE_LAST_FEW_CHUNKS): 653 msg = 'partially filled chunk %d ignored for last %d chunks' 654 logging.info(msg, i, self.IGNORE_LAST_FEW_CHUNKS) 655 else: 656 check_primary_frequencies = False 657 break 658 659 self.results = dict() 660 if check_legitimacy: 661 self.results['check_audio_frames_legitimacy'] = ( 662 check_audio_frames_legitimacy) 663 664 if check_frequencies: 665 self.results['check_primary_frequencies'] = ( 666 check_primary_frequencies) 667 668 return all(self.results.values()) 669 670 671 @test_retry_and_log(False) 672 def test_check_empty_chunks(self, device, test_data, duration, 673 test_profile): 674 """Check if all the chunks are empty. 675 676 @param device: The Bluetooth peer device. 677 @param test_data: The test data of the test profile. 678 @param duration: The duration of the audio file to test. 679 @param test_profile: Which audio profile is used. Profiles are defined 680 in bluetooth_audio_test_data.py. 681 682 @returns: True if all the chunks are empty. 683 """ 684 chunk_in_secs = test_data['chunk_in_secs'] 685 if not bool(chunk_in_secs): 686 chunk_in_secs = self.DEFAULT_CHUNK_IN_SECS 687 nchunks = duration // chunk_in_secs 688 logging.info('Number of chunks: %d', nchunks) 689 690 all_chunks_empty = True 691 for i in range(nchunks): 692 logging.info('Check chunk %d', i) 693 694 recorded_file = self.handle_one_chunk(device, chunk_in_secs, i, 695 test_profile) 696 if recorded_file is None: 697 raise error.TestError('Failed to handle chunk %d' % i) 698 699 700 # Check if the audio frames in the recorded file are legitimate. 701 if self._check_audio_frames_legitimacy( 702 test_data, 'recorded_by_peer', recorded_file): 703 if (i > self.IGNORE_LAST_FEW_CHUNKS and 704 i >= nchunks - self.IGNORE_LAST_FEW_CHUNKS): 705 logging.info('empty chunk %d ignored for last %d chunks', 706 i, self.IGNORE_LAST_FEW_CHUNKS) 707 else: 708 all_chunks_empty = False 709 break 710 711 self.results = {'all chunks are empty': all_chunks_empty} 712 713 return all(self.results.values()) 714 715 716 @test_retry_and_log(False) 717 def test_check_audio_file(self, 718 device, 719 test_profile, 720 test_data, 721 recording_device, 722 check_legitimacy=True, 723 check_frequencies=True): 724 """Check the audio file and verify the primary frequencies. 725 726 @param device: the Bluetooth peer device. 727 @param test_profile: A2DP or HFP test profile. 728 @param test_data: the test data of the test profile. 729 @param recording_device: which device recorded the audio, 730 possible values are 'recorded_by_dut' or 'recorded_by_peer'. 731 @param check_legitimacy: if set this to True, run 732 _check_audio_frames_legitimacy test. 733 @param check_frequencies: if set this to True, run 734 _check_primary_frequencies test. 735 736 @returns: True if audio file passes the frequencies check. 737 """ 738 if recording_device == 'recorded_by_peer': 739 logging.debug('Scp to DUT') 740 try: 741 recorded_file = test_data[recording_device] 742 self._scp_to_dut(device, recorded_file, recorded_file) 743 logging.debug('Recorded {} successfully'.format(recorded_file)) 744 except Exception as e: 745 raise error.TestError('Exception occurred when (%s)' % (e)) 746 747 self.results = dict() 748 if check_legitimacy: 749 self.results['check_audio_frames_legitimacy'] = ( 750 self._check_audio_frames_legitimacy( 751 test_data, recording_device)) 752 753 if check_frequencies: 754 self.results['check_primary_frequencies'] = ( 755 self._check_primary_frequencies( 756 test_profile, test_data, recording_device)) 757 758 return all(self.results.values()) 759 760 761 @test_retry_and_log(False) 762 def test_dut_to_start_playing_audio_subprocess(self, 763 test_data, 764 pin_device=None): 765 """Start playing audio in a subprocess. 766 767 @param test_data: the audio test data 768 769 @returns: True on success. False otherwise. 770 """ 771 start_playing_audio = self.bluetooth_facade.start_playing_audio_subprocess( 772 test_data, pin_device) 773 self.results = { 774 'dut_to_start_playing_audio_subprocess': start_playing_audio 775 } 776 return all(self.results.values()) 777 778 @test_retry_and_log(False) 779 def test_dut_to_stop_playing_audio_subprocess(self): 780 """Stop playing audio in the subprocess. 781 782 @returns: True on success. False otherwise. 783 """ 784 stop_playing_audio = ( 785 self.bluetooth_facade.stop_playing_audio_subprocess()) 786 787 self.results = { 788 'dut_to_stop_playing_audio_subprocess': stop_playing_audio 789 } 790 return all(self.results.values()) 791 792 @test_retry_and_log(False) 793 def test_dut_to_start_capturing_audio_subprocess(self, audio_data, 794 recording_device): 795 """Start capturing audio in a subprocess. 796 797 @param audio_data: the audio test data 798 @param recording_device: which device recorded the audio, 799 possible values are 'recorded_by_dut' or 'recorded_by_peer' 800 801 @returns: True on success. False otherwise. 802 """ 803 # Let the dut capture audio stream until it is stopped explicitly by 804 # setting duration to None. This is required on some slower devices. 805 audio_data = audio_data.copy() 806 audio_data.update({'duration': None}) 807 808 start_capturing_audio = self.bluetooth_facade.start_capturing_audio_subprocess( 809 audio_data, recording_device) 810 self.results = { 811 'dut_to_start_capturing_audio_subprocess': 812 start_capturing_audio 813 } 814 return all(self.results.values()) 815 816 @test_retry_and_log(False) 817 def test_dut_to_stop_capturing_audio_subprocess(self): 818 """Stop capturing audio. 819 820 @returns: True on success. False otherwise. 821 """ 822 stop_capturing_audio = ( 823 self.bluetooth_facade.stop_capturing_audio_subprocess()) 824 825 self.results = { 826 'dut_to_stop_capturing_audio_subprocess': stop_capturing_audio 827 } 828 return all(self.results.values()) 829 830 @test_retry_and_log(False) 831 def test_device_to_start_playing_audio_subprocess(self, device, 832 test_profile, test_data): 833 """Start playing the audio file in a subprocess. 834 835 @param device: the bluetooth peer device 836 @param test_data: the audio file to play and data about the file 837 @param audio_profile: the audio profile, either a2dp, hfp_wbs, or hfp_nbs 838 839 @returns: True on success. False otherwise. 840 """ 841 start_playing_audio = device.StartPlayingAudioSubprocess( 842 test_profile, test_data) 843 self.results = { 844 'device_to_start_playing_audio_subprocess': start_playing_audio 845 } 846 return all(self.results.values()) 847 848 @test_retry_and_log(False) 849 def test_device_to_stop_playing_audio_subprocess(self, device): 850 """Stop playing the audio file in a subprocess. 851 852 @param device: the bluetooth peer device 853 854 @returns: True on success. False otherwise. 855 """ 856 stop_playing_audio = device.StopPlayingAudioSubprocess() 857 self.results = { 858 'device_to_stop_playing_audio_subprocess': stop_playing_audio 859 } 860 return all(self.results.values()) 861 862 @test_retry_and_log(False) 863 def test_device_to_start_recording_audio_subprocess( 864 self, device, test_profile, test_data): 865 """Start recording audio in a subprocess. 866 867 @param device: the bluetooth peer device 868 @param test_profile: the audio profile used to get the recording settings 869 @param test_data: the details of the file being recorded 870 871 @returns: True on success. False otherwise. 872 """ 873 start_recording_audio = device.StartRecordingAudioSubprocess( 874 test_profile, test_data) 875 self.results = { 876 'device_to_start_recording_audio_subprocess': 877 start_recording_audio 878 } 879 return all(self.results.values()) 880 881 @test_retry_and_log(False) 882 def test_device_to_stop_recording_audio_subprocess(self, device): 883 """Stop the recording subprocess. 884 885 @returns: True on success. False otherwise. 886 """ 887 stop_recording_audio = device.StopRecordingingAudioSubprocess() 888 self.results = { 889 'device_to_stop_recording_audio_subprocess': 890 stop_recording_audio 891 } 892 return all(self.results.values()) 893 894 895 @test_retry_and_log(False) 896 def test_device_a2dp_connected(self, device, timeout=15): 897 """ Tests a2dp profile is connected on device. """ 898 self.results = {} 899 check_connection = lambda: self._get_pulseaudio_bluez_source_a2dp( 900 device, A2DP) 901 is_connected = self._wait_for_condition(check_connection, 902 'test_device_a2dp_connected', 903 timeout=timeout) 904 self.results['peer a2dp connected'] = is_connected 905 906 return all(self.results.values()) 907 908 909 @test_retry_and_log(False) 910 def test_hfp_connected(self, 911 bluez_function, 912 device, 913 test_profile, 914 timeout=15): 915 """Tests HFP profile is connected. 916 917 @param bluez_function: the appropriate bluez HFP function either 918 _get_pulseaudio_bluez_source_hfp or 919 _get_pulseaudio_bluez_sink_hfp depending on the role of the DUT. 920 @param device: the Bluetooth peer device. 921 @param test_profile: which test profile is used, HFP_WBS or HFP_NBS. 922 @param timeout: number of seconds to wait before giving up connecting 923 to HFP profile. 924 925 @returns: True on success. False otherwise. 926 """ 927 check_connection = lambda: bluez_function(device, test_profile) 928 is_connected = self._wait_for_condition(check_connection, 929 'test_hfp_connected', 930 timeout=timeout) 931 self.results = {'peer hfp connected': is_connected} 932 933 return all(self.results.values()) 934 935 936 @test_retry_and_log(False) 937 def test_send_audio_to_dut_and_unzip(self): 938 """Send the audio file to the DUT and unzip it. 939 940 @returns: True on success. False otherwise. 941 """ 942 try: 943 self.host.send_file(AUDIO_DATA_TARBALL_PATH, 944 AUDIO_DATA_TARBALL_PATH) 945 except Exception as e: 946 raise error.TestError('Fail to send file to the DUT: (%s)', e) 947 948 unzip_success = self.bluetooth_facade.unzip_audio_test_data( 949 AUDIO_DATA_TARBALL_PATH, DATA_DIR) 950 951 self.results = {'unzip audio file': unzip_success} 952 953 return all(self.results.values()) 954 955 956 @test_retry_and_log(False) 957 def test_get_visqol_score(self, test_file, test_profile, recording_device): 958 """Test that if the recorded audio file meets the passing score. 959 960 This function also records the visqol performance. 961 962 @param device: the Bluetooth peer device. 963 @param test_profile: which test profile is used, HFP_WBS or HFP_NBS. 964 @param recording_device: which device recorded the audio, 965 possible values are 'recorded_by_dut' or 'recorded_by_peer'. 966 967 @returns: True if the test files score at or above the 968 source_passing_score value as defined in 969 bluetooth_audio_test_data.py. 970 """ 971 dut_role = 'sink' if recording_device == 'recorded_by_dut' else 'source' 972 filename = os.path.split(test_file['file'])[1] 973 974 ref_file, deg_file = self.format_recorded_file(test_file, test_profile, 975 recording_device) 976 if not ref_file or not deg_file: 977 desc = 'Failed to get ref and deg file: ref {}, deg {}'.format( 978 ref_file, deg_file) 979 raise error.TestError(desc) 980 981 score = self.get_visqol_score(ref_file, 982 deg_file, 983 speech_mode=test_file['speech_mode']) 984 985 key = ''.join((dut_role, '_passing_score')) 986 logging.info('{} scored {}, min passing score: {}'.format( 987 filename, score, test_file[key])) 988 passed = score >= test_file[key] 989 self.results = {filename: passed} 990 991 # Track visqol performance 992 test_desc = '{}_{}_{}'.format(test_profile, dut_role, 993 test_file['reporting_type']) 994 self.write_perf_keyval({test_desc: score}) 995 996 if not passed: 997 logging.warning('Failed: {}'.format(filename)) 998 999 return all(self.results.values()) 1000 1001 1002 @test_retry_and_log(False) 1003 def test_avrcp_commands(self, device): 1004 """Test Case: Test AVRCP commands issued by peer can be received at DUT 1005 1006 The very first AVRCP command (Linux evdev event) the DUT receives 1007 contains extra information than just the AVRCP event, e.g. EV_REP 1008 report used to specify delay settings. Send the first command before 1009 the actual test starts to avoid dealing with them during test. 1010 1011 The peer device name is required to monitor the event reception on the 1012 DUT. However, as the peer device itself already registered with the 1013 kernel as an udev input device. The AVRCP profile will register as an 1014 separate input device with the name pattern: name + (AVRCP), e.g. 1015 RASPI_AUDIO (AVRCP). Using 'AVRCP' as device name to help search for 1016 the device. 1017 1018 @param device: the Bluetooth peer device 1019 1020 @returns: True if the all AVRCP commands received by DUT, false 1021 otherwise 1022 1023 """ 1024 device.SendMediaPlayerCommand('play') 1025 1026 name = device.name 1027 device.name = 'AVRCP' 1028 1029 result_pause = self.test_avrcp_event(device, 1030 device.SendMediaPlayerCommand, 'pause') 1031 result_play = self.test_avrcp_event(device, 1032 device.SendMediaPlayerCommand, 'play') 1033 result_stop = self.test_avrcp_event(device, 1034 device.SendMediaPlayerCommand, 'stop') 1035 result_next = self.test_avrcp_event(device, 1036 device.SendMediaPlayerCommand, 'next') 1037 result_previous = self.test_avrcp_event(device, 1038 device.SendMediaPlayerCommand, 'previous') 1039 1040 device.name = name 1041 self.results = {'pause': result_pause, 'play': result_play, 1042 'stop': result_stop, 'next': result_next, 1043 'previous': result_previous} 1044 return all(self.results.values()) 1045 1046 1047 @test_retry_and_log(False) 1048 def test_avrcp_media_info(self, device): 1049 """Test Case: Test AVRCP media info sent by DUT can be received by peer 1050 1051 The test update all media information twice to prevent previous 1052 leftover data affect the current iteration of test. Then compare the 1053 expected results against the information received on the peer device. 1054 1055 This test verifies media information including: playback status, 1056 length, title, artist, and album. Position of the media is not 1057 currently support as playerctl on the peer side cannot correctly 1058 retrieve such information. 1059 1060 Length and position information are transmitted in the unit of 1061 microsecond. However, BlueZ process those time data in the resolution 1062 of millisecond. Discard microsecond detail when comparing those media 1063 information. 1064 1065 @param device: the Bluetooth peer device 1066 1067 @returns: True if the all AVRCP media info received by DUT, false 1068 otherwise 1069 1070 """ 1071 # First round of updating media information to overwrite all leftovers. 1072 init_status = 'stopped' 1073 init_length = 20200414 1074 init_position = 8686868 1075 init_metadata = {'album': 'metadata_album_init', 1076 'artist': 'metadata_artist_init', 1077 'title': 'metadata_title_init'} 1078 self.bluetooth_facade.set_player_playback_status(init_status) 1079 self.bluetooth_facade.set_player_length(init_length) 1080 self.bluetooth_facade.set_player_position(init_position) 1081 self.bluetooth_facade.set_player_metadata(init_metadata) 1082 1083 # Second round of updating for actual testing. 1084 expected_status = 'playing' 1085 expected_length = 68686868 1086 expected_position = 20200414 1087 expected_metadata = {'album': 'metadata_album_expected', 1088 'artist': 'metadata_artist_expected', 1089 'title': 'metadata_title_expected'} 1090 self.bluetooth_facade.set_player_playback_status(expected_status) 1091 self.bluetooth_facade.set_player_length(expected_length) 1092 self.bluetooth_facade.set_player_position(expected_position) 1093 self.bluetooth_facade.set_player_metadata(expected_metadata) 1094 1095 received_media_info = device.GetMediaPlayerMediaInfo() 1096 logging.debug(received_media_info) 1097 1098 try: 1099 actual_length = int(received_media_info.get('length')) 1100 except: 1101 actual_length = 0 1102 1103 result_status = bool(expected_status == 1104 received_media_info.get('status').lower()) 1105 result_album = bool(expected_metadata['album'] == 1106 received_media_info.get('album')) 1107 result_artist = bool(expected_metadata['artist'] == 1108 received_media_info.get('artist')) 1109 result_title = bool(expected_metadata['title'] == 1110 received_media_info.get('title')) 1111 # The AVRCP time information is in the unit of microseconds but with 1112 # milliseconds resolution. Convert both send and received length into 1113 # milliseconds for comparison. 1114 result_length = bool(expected_length // 1000 == actual_length // 1000) 1115 1116 self.results = {'status': result_status, 'album': result_album, 1117 'artist': result_artist, 'title': result_title, 1118 'length': result_length} 1119 return all(self.results.values()) 1120 1121 1122 # --------------------------------------------------------------- 1123 # Definitions of all bluetooth audio test sequences 1124 # --------------------------------------------------------------- 1125 1126 def test_a2dp_sinewaves(self, device, test_profile, duration): 1127 """Test Case: a2dp sinewaves 1128 1129 @param device: the bluetooth peer device 1130 @param test_profile: the a2dp test profile; 1131 choices are A2DP and A2DP_LONG 1132 @param duration: the duration of the audio file to test 1133 0 means to use the default value in the test profile 1134 1135 """ 1136 # Make a copy since the test_data may be formatted with distinct 1137 # arguments in the follow-up tests. 1138 test_data = audio_test_data[test_profile].copy() 1139 if bool(duration): 1140 test_data['duration'] = duration 1141 else: 1142 duration = test_data['duration'] 1143 1144 test_data['file'] %= duration 1145 logging.info('%s test for %d seconds.', test_profile, duration) 1146 1147 # Wait for pulseaudio a2dp bluez source 1148 self.test_device_a2dp_connected(device) 1149 1150 # Select audio output node so that we do not rely on chrome to do it. 1151 self.test_select_audio_output_node_bluetooth() 1152 1153 # Start recording audio on the peer Bluetooth audio device. 1154 self.test_device_to_start_recording_audio_subprocess( 1155 device, test_profile, test_data) 1156 1157 # Play audio on the DUT in a non-blocked way and check the recorded 1158 # audio stream in a real-time manner. 1159 self.test_dut_to_start_playing_audio_subprocess(test_data) 1160 1161 # Check chunks of recorded streams and verify the primary frequencies. 1162 # This is a blocking call until all chunks are completed. 1163 self.test_check_chunks(device, test_profile, test_data, duration) 1164 1165 # Stop recording audio on the peer Bluetooth audio device. 1166 self.test_device_to_stop_recording_audio_subprocess(device) 1167 1168 # Stop playing audio on DUT. 1169 self.test_dut_to_stop_playing_audio_subprocess() 1170 1171 1172 def playback_and_connect(self, device, test_profile): 1173 """Connect then disconnect an A2DP device while playing stream. 1174 1175 This test first plays the audio stream and then selects the BT device 1176 as output node, checking if the stream has routed to the BT device. 1177 After that, disconnect the BT device and also check whether the stream 1178 closes on it gracefully. 1179 1180 @param device: the Bluetooth peer device. 1181 @param test_profile: to select which A2DP test profile is used. 1182 """ 1183 test_data = audio_test_data[test_profile] 1184 1185 # TODO(b/207046142): Remove the old version fallback after the new 1186 # Chameleon bundle is deployed. 1187 # Currently the BT audio tests store test profile parameters in 1188 # Chameleon bundle. However, we decide to move the test profiles to 1189 # server test. During the transition, the new test code may interact 1190 # with old/existing Chameleon bundle, which does not have A2DP_MEDIUM 1191 # profile. We use a trick here: override the passing-in test_profile 1192 # with A2DP so that Chameleon can look up the profile, and override the 1193 # three parameters locally to make it a A2DP_MEDIUM profile. 1194 test_profile = A2DP 1195 test_data = audio_test_data[test_profile].copy() 1196 test_data['duration'] = 60 1197 test_data['chunk_checking_duration'] = 5 1198 test_data['chunk_in_secs'] = 1 1199 1200 # Start playing audio on the Dut. 1201 self.test_dut_to_start_playing_audio_subprocess(test_data) 1202 1203 # Connect the Bluetooth device. 1204 self.test_device_set_discoverable(device, True) 1205 self.test_discover_device(device.address) 1206 self.test_pairing(device.address, device.pin, trusted=True) 1207 self.test_connection_by_adapter(device.address) 1208 self.test_device_a2dp_connected(device) 1209 1210 # Select Bluetooth as output node. 1211 self.test_select_audio_output_node_bluetooth() 1212 1213 self.test_device_to_start_recording_audio_subprocess( 1214 device, test_profile, test_data) 1215 1216 # Handle chunks of recorded streams and verify the primary frequencies. 1217 # This is a blocking call until all chunks are completed. 1218 self.test_check_chunks(device, test_profile, test_data, 1219 test_data['chunk_checking_duration']) 1220 1221 self.test_device_to_stop_recording_audio_subprocess(device) 1222 1223 self.test_select_audio_output_node_internal_speaker() 1224 1225 # Check if the device disconnects successfully. 1226 self.expect_test(False, self.test_device_a2dp_connected, device) 1227 1228 self.test_dut_to_stop_playing_audio_subprocess() 1229 1230 1231 def playback_and_disconnect(self, device, test_profile): 1232 """Disconnect the Bluetooth device while the stream is playing. 1233 1234 This test will keep the stream playing and then disconnect the 1235 Bluetooth device. The goal is to check the stream is still alive 1236 after the Bluetooth device disconnected. 1237 1238 @param device: the Bluetooth peer device. 1239 @param test_profile: to select which A2DP test profile is used. 1240 """ 1241 test_data = audio_test_data[test_profile] 1242 1243 # TODO(b/207046142): Remove the old version fallback after the new 1244 # Chameleon bundle is deployed. 1245 # Currently the BT audio tests store test profile parameters in 1246 # Chameleon bundle. However, we decide to move the test profiles to 1247 # server test. During the transition, the new test code may interact 1248 # with old/existing Chameleon bundle, which does not have A2DP_MEDIUM 1249 # profile. We use a trick here: override the passing-in test_profile 1250 # with A2DP so that Chameleon can look up the profile, and override the 1251 # three parameters locally to make it a A2DP_MEDIUM profile. 1252 test_profile = A2DP 1253 test_data = audio_test_data[test_profile].copy() 1254 test_data['duration'] = 60 1255 test_data['chunk_checking_duration'] = 5 1256 test_data['chunk_in_secs'] = 1 1257 1258 # Connect the Bluetooth device. 1259 self.test_device_set_discoverable(device, True) 1260 self.test_discover_device(device.address) 1261 self.test_pairing(device.address, device.pin, trusted=True) 1262 self.test_connection_by_adapter(device.address) 1263 self.test_device_a2dp_connected(device) 1264 1265 # Select Bluetooth as output node. 1266 self.test_select_audio_output_node_bluetooth() 1267 1268 self.test_device_to_start_recording_audio_subprocess( 1269 device, test_profile, test_data) 1270 1271 # Start playing audio on the DUT. 1272 self.test_dut_to_start_playing_audio_subprocess(test_data) 1273 1274 # Handle chunks of recorded streams and verify the primary frequencies. 1275 # This is a blocking call until all chunks are completed. 1276 self.test_check_chunks(device, test_profile, test_data, 1277 test_data['chunk_checking_duration']) 1278 1279 self.test_device_to_stop_recording_audio_subprocess(device) 1280 1281 # Disconnect the Bluetooth device. 1282 self.test_disconnection_by_adapter(device.address) 1283 1284 # Obtain audio thread summary to check if the audio stream is still 1285 # alive. 1286 self.test_audio_is_alive_on_dut() 1287 1288 # Stop playing audio on the DUT. 1289 self.test_dut_to_stop_playing_audio_subprocess() 1290 1291 1292 def playback_back2back(self, device, test_profile): 1293 """Repeat to start and stop the playback stream several times. 1294 1295 This test repeats to start and stop the playback stream and verify 1296 that the Bluetooth device receives the stream correctly. 1297 1298 @param device: the Bluetooth peer device. 1299 @param test_profile: to select which A2DP test profile is used. 1300 """ 1301 test_data = audio_test_data[test_profile] 1302 1303 # TODO(b/207046142): Remove the old version fallback after the new 1304 # Chameleon bundle is deployed. 1305 # Currently the BT audio tests store test profile parameters in 1306 # Chameleon bundle. However, we decide to move the test profiles to 1307 # server test. During the transition, the new test code may interact 1308 # with old/existing Chameleon bundle, which does not have A2DP_MEDIUM 1309 # profile. We use a trick here: override the passing-in test_profile 1310 # with A2DP so that Chameleon can look up the profile, and override the 1311 # three parameters locally to make it a A2DP_MEDIUM profile. 1312 test_profile = A2DP 1313 test_data = audio_test_data[test_profile].copy() 1314 test_data['duration'] = 60 1315 test_data['chunk_checking_duration'] = 5 1316 test_data['chunk_in_secs'] = 1 1317 1318 self.test_device_set_discoverable(device, True) 1319 self.test_discover_device(device.address) 1320 self.test_pairing(device.address, device.pin, trusted=True) 1321 self.test_connection_by_adapter(device.address) 1322 1323 self.test_device_a2dp_connected(device) 1324 self.test_select_audio_output_node_bluetooth() 1325 1326 for _ in range(3): 1327 # TODO(b/208165757): In here if we record the audio stream before 1328 # playing that will cause an audio blank about 1~2 sec in the 1329 # beginning of the recorded file and make the chunks checking fail. 1330 # Need to fix this problem in the future. 1331 self.test_dut_to_start_playing_audio_subprocess(test_data) 1332 self.test_device_to_start_recording_audio_subprocess( 1333 device, test_profile, test_data) 1334 self.test_check_chunks(device, test_profile, test_data, 1335 test_data['chunk_checking_duration']) 1336 self.test_dut_to_stop_playing_audio_subprocess() 1337 self.test_device_to_stop_recording_audio_subprocess(device) 1338 1339 self.test_device_to_start_recording_audio_subprocess( 1340 device, test_profile, test_data) 1341 self.test_check_empty_chunks(device, test_data, 1342 test_data['chunk_checking_duration'], 1343 test_profile) 1344 self.test_device_to_stop_recording_audio_subprocess(device) 1345 1346 self.test_disconnection_by_adapter(device.address) 1347 1348 1349 def pinned_playback(self, device, test_profile): 1350 """Play an audio stream that is pinned to the Bluetooth device. 1351 1352 This test does not choose Bluetooth as the output node but directly 1353 plays the sound that is pinned to the Bluetooth device and check 1354 whether it receives the audio stream correctly. 1355 1356 @param device: the Bluetooth peer device. 1357 @param test_profile: to select which A2DP test profile is used. 1358 """ 1359 test_data = audio_test_data[test_profile] 1360 1361 self.test_device_set_discoverable(device, True) 1362 self.test_discover_device(device.address) 1363 self.test_pairing(device.address, device.pin, trusted=True) 1364 self.test_connection_by_adapter(device.address) 1365 1366 self.test_device_a2dp_connected(device) 1367 self.test_device_to_start_recording_audio_subprocess( 1368 device, test_profile, test_data) 1369 1370 # We do not select Bluetooth as output node but play audio pinned to 1371 # the Bluetooth device straight forward. 1372 device_id = self.bluetooth_facade.get_device_id_from_node_type( 1373 self.CRAS_BLUETOOTH_OUTPUT_NODE_TYPE, False) 1374 logging.info("Bluetooth device id for audio stream output: %s", 1375 device_id) 1376 self.test_dut_to_start_playing_audio_subprocess(test_data, device_id) 1377 self.test_check_chunks(device, test_profile, test_data, 1378 test_data['duration']) 1379 self.test_dut_to_stop_playing_audio_subprocess() 1380 self.test_device_to_stop_recording_audio_subprocess(device) 1381 self.test_disconnection_by_adapter(device.address) 1382 1383 1384 def hfp_dut_as_source_visqol_score(self, device, test_profile): 1385 """Test Case: HFP test files streaming from peer device to the DUT. 1386 1387 @param device: the Bluetooth peer device. 1388 @param test_profile: which test profile is used, HFP_WBS or HFP_NBS. 1389 """ 1390 # list of test wav files 1391 hfp_test_data = audio_test_data[test_profile] 1392 test_files = hfp_test_data['visqol_test_files'] 1393 1394 get_visqol_binary() 1395 get_audio_test_data() 1396 1397 # Download test data to the DUT. 1398 self.test_send_audio_to_dut_and_unzip() 1399 1400 for test_file in test_files: 1401 filename = os.path.split(test_file['file'])[1] 1402 logging.debug('Testing file: {}'.format(filename)) 1403 1404 self.test_select_audio_input_device(device.name) 1405 self.test_select_audio_output_node_bluetooth() 1406 1407 # Enable HFP profile. 1408 self.test_dut_to_start_capturing_audio_subprocess( 1409 test_file, 'recorded_by_peer') 1410 1411 # Wait for pulseaudio bluez hfp source/sink 1412 self.test_hfp_connected(self._get_pulseaudio_bluez_source_hfp, 1413 device, test_profile) 1414 1415 self.test_device_to_start_recording_audio_subprocess( 1416 device, test_profile, test_file) 1417 1418 # Play audio on the DUT in a non-blocked way. 1419 # If there are issues, cras_test_client playing back might be blocked 1420 # forever. We would like to avoid the testing procedure from that. 1421 self.test_dut_to_start_playing_audio_subprocess(test_file) 1422 time.sleep(test_file['chunk_checking_duration']) 1423 self.test_dut_to_stop_playing_audio_subprocess() 1424 self.test_device_to_stop_recording_audio_subprocess(device) 1425 1426 # Disable HFP profile. 1427 self.test_dut_to_stop_capturing_audio_subprocess() 1428 1429 # Copy the recorded audio file to the DUT for spectrum analysis. 1430 recorded_file = test_file['recorded_by_peer'] 1431 self._scp_to_dut(device, recorded_file, recorded_file) 1432 1433 self.test_get_visqol_score(test_file, test_profile, 1434 'recorded_by_peer') 1435 1436 1437 def hfp_dut_as_sink_visqol_score(self, device, test_profile): 1438 """Test Case: HFP test files streaming from peer device to the DUT. 1439 1440 @param device: the Bluetooth peer device. 1441 @param test_profile: which test profile is used, HFP_WBS or HFP_NBS. 1442 """ 1443 # list of test wav files 1444 hfp_test_data = audio_test_data[test_profile] 1445 test_files = hfp_test_data['visqol_test_files'] 1446 1447 get_visqol_binary() 1448 get_audio_test_data() 1449 1450 # Download test data to the DUT. 1451 self.test_send_audio_to_dut_and_unzip() 1452 1453 for test_file in test_files: 1454 filename = os.path.split(test_file['file'])[1] 1455 logging.debug('Testing file: {}'.format(filename)) 1456 1457 self.test_select_audio_input_device(device.name) 1458 self.test_select_audio_output_node_bluetooth() 1459 1460 # Enable HFP profile. 1461 self.test_dut_to_start_capturing_audio_subprocess( 1462 test_file, 'recorded_by_dut') 1463 1464 # Wait for pulseaudio bluez hfp source/sink. 1465 self.test_hfp_connected(self._get_pulseaudio_bluez_sink_hfp, 1466 device, test_profile) 1467 1468 self.test_select_audio_input_device(device.name) 1469 1470 self.test_device_to_start_playing_audio_subprocess( 1471 device, test_profile, test_file) 1472 time.sleep(test_file['chunk_checking_duration']) 1473 self.test_device_to_stop_playing_audio_subprocess(device) 1474 1475 # Disable HFP profile. 1476 self.test_dut_to_stop_capturing_audio_subprocess() 1477 logging.debug('Recorded {} successfully'.format(filename)) 1478 1479 self.test_get_visqol_score(test_file, test_profile, 1480 'recorded_by_dut') 1481 1482 1483 def hfp_dut_as_source(self, device, test_profile): 1484 """Test Case: HFP sinewave streaming from the DUT to peer device. 1485 1486 @param device: the Bluetooth peer device. 1487 @param test_profile: which test profile is used, HFP_WBS or HFP_NBS. 1488 """ 1489 hfp_test_data = audio_test_data[test_profile] 1490 1491 self.test_select_audio_input_device(device.name) 1492 self.test_select_audio_output_node_bluetooth() 1493 1494 # Enable HFP profile. 1495 self.test_dut_to_start_capturing_audio_subprocess( 1496 hfp_test_data, 'recorded_by_peer') 1497 1498 # Wait for pulseaudio bluez hfp source/sink 1499 self.test_hfp_connected(self._get_pulseaudio_bluez_source_hfp, device, 1500 test_profile) 1501 1502 self.test_device_to_start_recording_audio_subprocess( 1503 device, test_profile, hfp_test_data) 1504 self.test_dut_to_start_playing_audio_subprocess(hfp_test_data) 1505 time.sleep(hfp_test_data['chunk_checking_duration']) 1506 self.test_dut_to_stop_playing_audio_subprocess() 1507 self.test_device_to_stop_recording_audio_subprocess(device) 1508 self.test_check_audio_file(device, test_profile, hfp_test_data, 1509 'recorded_by_peer') 1510 1511 # Disable HFP profile. 1512 self.test_dut_to_stop_capturing_audio_subprocess() 1513 1514 1515 def hfp_dut_as_sink(self, device, test_profile): 1516 """Test Case: HFP sinewave streaming from peer device to the DUT. 1517 1518 @param device: the Bluetooth peer device. 1519 @param test_profile: which test profile is used, HFP_WBS or HFP_NBS. 1520 """ 1521 hfp_test_data = audio_test_data[test_profile] 1522 1523 self.test_select_audio_input_device(device.name) 1524 self.test_select_audio_output_node_bluetooth() 1525 1526 # Enable HFP profile. 1527 self.test_dut_to_start_capturing_audio_subprocess( 1528 hfp_test_data, 'recorded_by_dut') 1529 1530 # Wait for pulseaudio bluez hfp source/sink 1531 self.test_hfp_connected(self._get_pulseaudio_bluez_sink_hfp, device, 1532 test_profile) 1533 1534 self.test_select_audio_input_device(device.name) 1535 1536 self.test_device_to_start_playing_audio_subprocess( 1537 device, test_profile, hfp_test_data) 1538 time.sleep(hfp_test_data['chunk_checking_duration']) 1539 self.test_device_to_stop_playing_audio_subprocess(device) 1540 1541 # Disable HFP profile. 1542 self.test_dut_to_stop_capturing_audio_subprocess() 1543 self.test_check_audio_file(device, test_profile, hfp_test_data, 1544 'recorded_by_dut') 1545 1546 1547 def hfp_dut_as_source_back2back(self, device, test_profile): 1548 """Play and stop the audio stream from DUT to Bluetooth peer device. 1549 1550 The test starts then stops the stream playback for three times. In each 1551 iteration, it checks the Bluetooth device can successfully receive the 1552 stream when it is played; also check the absence of the streama when 1553 stop playing. 1554 1555 @param device: the Bluetooth peer device. 1556 @param test_profile: which test profile is used, HFP_WBS or HFP_NBS. 1557 """ 1558 hfp_test_data = audio_test_data[test_profile] 1559 1560 # Select audio input device. 1561 self.test_select_audio_input_device(device.name) 1562 1563 # Select audio output node so that we do not rely on chrome to do it. 1564 self.test_select_audio_output_node_bluetooth() 1565 1566 # Enable HFP profile. 1567 self.test_dut_to_start_capturing_audio_subprocess(hfp_test_data, 1568 'recorded_by_peer') 1569 1570 # Wait for pulseaudio bluez hfp source/sink 1571 self.test_hfp_connected( 1572 self._get_pulseaudio_bluez_source_hfp, device, test_profile) 1573 1574 for _ in range(3): 1575 # TODO(b/208165757): If we record the audio stream before playing 1576 # that will cause an audio blank about 1~2 sec in the beginning of 1577 # the recorded file and make the chunks checking fail. Need to fix 1578 # this problem in the future. 1579 self.test_dut_to_start_playing_audio_subprocess(hfp_test_data) 1580 self.test_device_to_start_recording_audio_subprocess( 1581 device, test_profile, hfp_test_data) 1582 time.sleep(hfp_test_data['chunk_checking_duration']) 1583 1584 self.test_dut_to_stop_playing_audio_subprocess() 1585 self.test_device_to_stop_recording_audio_subprocess(device) 1586 self.test_check_audio_file(device, test_profile, hfp_test_data, 1587 'recorded_by_peer') 1588 1589 self.test_device_to_start_recording_audio_subprocess( 1590 device, test_profile, hfp_test_data) 1591 time.sleep(hfp_test_data['chunk_checking_duration']) 1592 1593 self.test_device_to_stop_recording_audio_subprocess(device) 1594 self.test_check_audio_file(device, test_profile, hfp_test_data, 1595 recording_device='recorded_by_peer', 1596 check_frequencies=False) 1597 1598 # Disable HFP profile. 1599 self.test_dut_to_stop_capturing_audio_subprocess() 1600 1601 1602 def a2dp_to_hfp_dut_as_source(self, device, test_profile): 1603 """Play the audio from DUT to Bluetooth device and switch the profile. 1604 1605 This test first uses A2DP profile and plays the audio stream on the 1606 DUT, checking if the peer receives the audio stream correctly. And 1607 then switch to the HFP_NBS profile and check the audio stream again. 1608 1609 @param device: the Bluetooth peer device. 1610 @param test_profile: which test profile is used, HFP_WBS_MEDIUM or 1611 HFP_NBS_MEDIUM. 1612 """ 1613 hfp_test_data = audio_test_data[test_profile] 1614 1615 # Wait for pulseaudio a2dp bluez source. 1616 self.test_device_a2dp_connected(device) 1617 1618 # Select audio output node so that we do not rely on chrome to do it. 1619 self.test_select_audio_output_node_bluetooth() 1620 1621 self.test_device_to_start_recording_audio_subprocess( 1622 device, test_profile, hfp_test_data) 1623 1624 # Play audio on the DUT in a non-blocked way and check the recorded 1625 # audio stream in a real-time manner. 1626 self.test_dut_to_start_playing_audio_subprocess(hfp_test_data) 1627 1628 time.sleep(hfp_test_data['chunk_checking_duration']) 1629 1630 self.test_device_to_stop_recording_audio_subprocess(device) 1631 1632 self.test_check_audio_file(device, test_profile, hfp_test_data, 1633 'recorded_by_peer') 1634 1635 self.test_select_audio_input_device(device.name) 1636 1637 # Enable HFP profile. 1638 self.test_dut_to_start_capturing_audio_subprocess(hfp_test_data, 1639 'recorded_by_peer') 1640 1641 # Wait for pulseaudio bluez hfp source/sink. 1642 self.test_hfp_connected( 1643 self._get_pulseaudio_bluez_source_hfp, device, test_profile) 1644 1645 self.test_device_to_start_recording_audio_subprocess( 1646 device, test_profile, hfp_test_data) 1647 1648 time.sleep(hfp_test_data['chunk_checking_duration']) 1649 1650 self.test_dut_to_stop_playing_audio_subprocess() 1651 1652 self.test_device_to_stop_recording_audio_subprocess(device) 1653 1654 self.test_check_audio_file(device, test_profile, hfp_test_data, 1655 'recorded_by_peer') 1656 1657 # Disable HFP profile. 1658 self.test_dut_to_stop_capturing_audio_subprocess() 1659 1660 1661 def hfp_to_a2dp_dut_as_source(self, device, test_profile): 1662 """Play the audio from DUT to Bluetooth peer in A2DP then switch to HFP. 1663 1664 This test first uses HFP profile and plays the audio stream on the DUT, 1665 checking if the peer receives the audio stream correctly. And then 1666 switch to the A2DP profile and check the audio stream again. 1667 1668 @param device: the Bluetooth peer device. 1669 @param test_profile: which test profile is used, 1670 HFP_NBS_MEDIUM or HFP_WBS_MEDIUM. 1671 """ 1672 hfp_test_data = audio_test_data[test_profile] 1673 1674 self.test_select_audio_input_device(device.name) 1675 1676 # Select audio output node so that we do not rely on chrome to do it. 1677 self.test_select_audio_output_node_bluetooth() 1678 1679 # Enable HFP profile. 1680 self.test_dut_to_start_capturing_audio_subprocess(hfp_test_data, 1681 'recorded_by_peer') 1682 1683 # Wait for pulseaudio bluez hfp source/sink. 1684 self.test_hfp_connected( 1685 self._get_pulseaudio_bluez_source_hfp, device, test_profile) 1686 1687 # Play audio on the DUT in a non-blocked way and check the recorded 1688 # audio stream in a real-time manner. 1689 self.test_dut_to_start_playing_audio_subprocess(hfp_test_data) 1690 self.test_device_to_start_recording_audio_subprocess( 1691 device, test_profile, hfp_test_data) 1692 time.sleep(hfp_test_data['chunk_checking_duration']) 1693 1694 self.test_device_to_stop_recording_audio_subprocess(device) 1695 self.test_check_audio_file(device, test_profile, hfp_test_data, 1696 'recorded_by_peer') 1697 1698 # Disable HFP profile. 1699 self.test_dut_to_stop_capturing_audio_subprocess() 1700 1701 # Wait for pulseaudio a2dp bluez source. 1702 self.test_device_a2dp_connected(device) 1703 1704 self.test_device_to_start_recording_audio_subprocess( 1705 device, test_profile, hfp_test_data) 1706 time.sleep(hfp_test_data['chunk_checking_duration']) 1707 1708 self.test_dut_to_stop_playing_audio_subprocess() 1709 self.test_check_audio_file(device, test_profile, hfp_test_data, 1710 'recorded_by_peer') 1711 self.test_device_to_stop_recording_audio_subprocess(device) 1712