| # Copyright 2017 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import logging |
| import random |
| import re |
| import time |
| |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.common_lib.cros import power_cycle_usb_util |
| from autotest_lib.client.common_lib.cros.cfm import cras_node_collector |
| from autotest_lib.client.common_lib.cros.cfm.usb import cfm_usb_devices |
| from autotest_lib.client.common_lib.cros.cfm.usb import usb_device_collector |
| from autotest_lib.server.cros.cfm import cfm_base_test |
| |
| |
| # CFMs have a base volume level threshold. Setting the level below 2 |
| # is interpreted by the CFM as 0. |
| CFM_VOLUME_LEVEL_LOWER_LIMIT = 2 |
| CFM_VOLUME_LEVEL_UPPER_LIMIT = 100 |
| JABRA = cfm_usb_devices.JABRA_SPEAK_410 |
| DUAL_SPEAKER_DEVICE_NAME = JABRA.product |
| TIMEOUT_SECS = 10 |
| |
| |
| class enterprise_CFM_DualSpeaker(cfm_base_test.CfmBaseTest): |
| """ |
| Tests that the following functionality works on CfM enrolled devices: |
| |
| 1. Mixer mute/umute state should be in sync between CfM and 2 speakers. |
| 2. Volume of two speakers should be in sync with the volume set by CfM. |
| 3. When muting/unmuting speakers from CfM, #1 still holds. |
| 4. When changing volume from CfM, #2 still holds. |
| 5. After disconnect/re-connecting any speaker #1-4 still holds. |
| """ |
| version = 1 |
| |
| |
| def _get_cras_jabra_speaker_nodes(self): |
| """ |
| Gets jabra speaker nodes. |
| |
| @returns A list cras input nodes representing Jabra speakers. |
| """ |
| nodes = self.cras_collector.get_output_nodes() |
| return [n for n in nodes if |
| DUAL_SPEAKER_DEVICE_NAME in n.device_name] |
| |
| def _get_amixer_jabra_mic_node_ids(self): |
| """ |
| Gets jabra mixer (microphone) card IDs from arecord. |
| |
| @returns A list of mixer card IDs or [] if no mixer is found. |
| """ |
| cmd = ("arecord -l" |
| " | grep \"%s\"" |
| " | awk -v N=2 '{print $N}'" % DUAL_SPEAKER_DEVICE_NAME) |
| mixer_cards = [s.strip().split(':')[0] for s in |
| self._host.run_output(cmd).splitlines()] |
| return mixer_cards |
| |
| def _get_cras_default_speakers(self): |
| """ |
| Gets the default speakers from cras_test_client. |
| |
| @returns A list of speaker nodes. |
| """ |
| nodes = self.cras_collector.get_output_nodes() |
| return [n for n in nodes if '*(default)' in n.node_name] |
| |
| def _get_cras_default_mixers(self): |
| """ |
| Gets the default mixers from cras_test_client. |
| |
| @returns A list of speaker node IDs or [] if no device is found. |
| """ |
| nodes = self.cras_collector.get_input_nodes() |
| return [n for n in nodes if n.node_name.startswith('*')] |
| |
| def _get_cras_jabra_mixer_nodes(self): |
| """ |
| Gets the mixer nodes from cras_test_client. |
| |
| @returns A list of mixer node IDs or [] if no device is found. |
| """ |
| nodes = self.cras_collector.get_input_nodes() |
| return [n for n in nodes if DUAL_SPEAKER_DEVICE_NAME in n.node_name] |
| |
| def _get_cras_speaker_volume(self, node_id): |
| """ |
| Gets the speaker volume for a node from cras_test_client. |
| |
| @param node_id: the node ID to query. |
| @returns the volume of speaker. |
| """ |
| for node in self.cras_collector.get_output_nodes(): |
| if node.node_id == node_id: |
| return node.volume |
| |
| def _get_mixer_mute_state(self, node_id): |
| """ |
| Gets the speaker mute state from cras_test_client. |
| |
| @param node_id: the node to query. |
| @returns True if speakers is muted, otherwise False. |
| """ |
| cmd = ("amixer -c %s" |
| " | grep \"Mono: Capture\"" |
| " | awk -v N=6 '{print $N}'" % node_id) |
| mute = [s.strip() for s in |
| self._host.run_output(cmd).splitlines()][0] |
| |
| if re.search(r"\[(off)\]", mute): |
| return True |
| else: |
| return False |
| |
| |
| def _test_volume_sync_for_dual_speakers(self): |
| """Checks whether volume is synced between dual speakers and CfM.""" |
| cfm_volume = self.cfm_facade.get_speaker_volume() |
| # There is case where cfm_facade.get_speaker_volume() returns empty |
| # string. Script polls volume from App up to 15 seconds or until |
| # non-zero value is returned. |
| poll_time = 0 |
| while not cfm_volume and poll_time < TIMEOUT_SECS: |
| cfm_volume = self.cfm_facade.get_speaker_volume() |
| time.sleep(1) |
| logging.info('Checking volume set by App on CfM: %s', cfm_volume) |
| poll_time += 1 |
| if not cfm_volume: |
| logging.info('Volume returned from App is Null') |
| return False |
| nodes = self._get_cras_default_speakers() |
| for node in nodes: |
| cras_volume = self._get_cras_speaker_volume(node.node_id) |
| logging.info('Volume in CfM and cras are sync for ' |
| 'node %s? cfm: %s, cras: %s', |
| str(node), cfm_volume, cras_volume) |
| if int(cfm_volume) != int(cras_volume): |
| logging.error('Test _test_volume_sync_for_dual_speakers fails' |
| ' for node %s', node.device_name) |
| return False |
| return True |
| |
| |
| def _test_mute_state_sync_for_dual_speakers(self): |
| """Checks whether mute/unmute is sync between dual speakers and CfM.""" |
| cfm_mute = self.cfm_facade.is_mic_muted() |
| if cfm_mute: |
| logging.info('Mixer is muted from CfM.') |
| else: |
| logging.info('Mixer is not muted from CfM.') |
| |
| nodes = self._get_amixer_jabra_mic_node_ids() |
| for node_id in nodes: |
| amixer_mute = self._get_mixer_mute_state(node_id) |
| if amixer_mute: |
| logging.info('amixer shows mic is muted for node %s.', node_id) |
| else: |
| logging.info('amixer shows mix not muted for node %s', node_id) |
| if not cfm_mute == amixer_mute: |
| logging.error('Test _test_mute_state_sync_for_dual_speakers ' |
| 'fails for node %s', node_id) |
| return False |
| return True |
| |
| |
| def _has_dual_speakers(self): |
| """ |
| Checks if there are dual speakers connected to the DUT. |
| |
| @returns True if there are dual speakers, false otherwise. |
| """ |
| collector = usb_device_collector.UsbDeviceCollector(self._host) |
| speakers = collector.get_devices_by_spec(JABRA) |
| return len(speakers) == 2 |
| |
| def _set_preferred_speaker(self, speaker_name): |
| """Set preferred speaker to Dual speaker.""" |
| logging.info('CfM sets preferred speaker to %s.', speaker_name) |
| self.cfm_facade.set_preferred_speaker(speaker_name) |
| time.sleep(TIMEOUT_SECS) |
| current_prefered_speaker = self.cfm_facade.get_preferred_speaker() |
| logging.info('Prefered speaker set to %s', current_prefered_speaker) |
| if speaker_name != current_prefered_speaker: |
| raise error.TestFail('Failed to set prefered speaker! ' |
| 'Expected %s, got %s', |
| speaker_name, current_prefered_speaker) |
| |
| |
| def _set_preferred_mixer(self, mixer_name): |
| """Set preferred mixer/microphone to Dual speaker.""" |
| logging.info('CfM sets preferred mixer to %s.', mixer_name) |
| self.cfm_facade.set_preferred_mic(mixer_name) |
| time.sleep(TIMEOUT_SECS) |
| current_prefered_mixer = self.cfm_facade.get_preferred_speaker() |
| logging.info('Prefered mixer set to %s by CfM.', current_prefered_mixer) |
| if mixer_name != current_prefered_mixer: |
| raise error.TestFail('Failed to set prefered mixer! ' |
| 'Expected %s, got %s', |
| mixer_name, current_prefered_mixer) |
| |
| |
| def _set_preferred_speaker_mixer(self): |
| """Sets preferred speaker and mixer to Dual speaker.""" |
| self._set_preferred_speaker(DUAL_SPEAKER_DEVICE_NAME) |
| self._set_preferred_mixer(DUAL_SPEAKER_DEVICE_NAME) |
| time.sleep(TIMEOUT_SECS) |
| default_speaker_ids = set([ |
| n.node_id for n in self._get_cras_default_speakers()]) |
| cras_speaker_ids = set([ |
| n.node_id for n in self._get_cras_jabra_speaker_nodes()]) |
| if default_speaker_ids != cras_speaker_ids: |
| raise error.TestFail('Dual speakers not set to preferred speaker ' |
| 'dual=%s; cras=%s' % (default_speaker_ids, |
| cras_speaker_ids)) |
| default_mixer_ids = set([ |
| n.node_id for n in self._get_cras_default_mixers()]) |
| jabra_mixer_ids = set([ |
| n.node_id for n in self._get_cras_jabra_mixer_nodes()]) |
| if default_mixer_ids != jabra_mixer_ids: |
| raise error.TestFail('Dual mixs is not set to preferred speaker.') |
| |
| |
| def _test_dual_speaker_sanity(self): |
| """ |
| Performs a speaker sanity check: |
| 1. Checks whether volume is sync between dual speakers and CfM |
| 2. Checks whether mute/unmute is sync between dual speakers and CfM. |
| @returns True if passed, otherwise False. |
| """ |
| volume = self._test_volume_sync_for_dual_speakers() |
| mute = self._test_mute_state_sync_for_dual_speakers() |
| return volume and mute |
| |
| |
| def _test_mute_sync(self): |
| """ |
| Mutes and unmutes speaker from CfM. |
| Check whether mute/unmute is sync between dual speakers and CfM. |
| @returns True if yes, else retrun False. |
| """ |
| self.cfm_facade.mute_mic() |
| if not self.cfm_facade.is_mic_muted(): |
| raise error.TestFail('CFM fails to mute mic') |
| time.sleep(TIMEOUT_SECS) |
| muted = self._test_mute_state_sync_for_dual_speakers() |
| self.cfm_facade.unmute_mic() |
| if self.cfm_facade.is_mic_muted(): |
| raise error.TestFail('CFM fails to unmute mic') |
| time.sleep(TIMEOUT_SECS) |
| unmuted = self._test_mute_state_sync_for_dual_speakers() |
| return muted and unmuted |
| |
| def _test_volume_sync(self): |
| """ |
| Changes speaker volume from CfM. |
| Checks whether volume is sync between dual speakers and CfM. |
| @returns True if check succeeds, otherwise False. |
| """ |
| test_volume = random.randrange(CFM_VOLUME_LEVEL_LOWER_LIMIT, |
| CFM_VOLUME_LEVEL_UPPER_LIMIT) |
| self.cfm_facade.set_speaker_volume(str(test_volume)) |
| time.sleep(TIMEOUT_SECS) |
| return self._test_volume_sync_for_dual_speakers() |
| |
| |
| def run_once(self): |
| """Runs the test.""" |
| logging.info('Sanity check and initilization:') |
| if not self._has_dual_speakers(): |
| raise error.TestFail('No dual speakers found on DUT.') |
| |
| self.cras_collector = cras_node_collector.CrasNodeCollector(self._host) |
| |
| # Remove 'board:' prefix. |
| board_name = self._host.get_board().split(':')[1] |
| gpio_list = power_cycle_usb_util.get_target_all_gpio( |
| self._host, board_name, JABRA.vendor_id, |
| JABRA.product_id) |
| if len(set(gpio_list)) == 1: |
| raise error.TestFail('Speakers have to be tied to different GPIO.') |
| |
| self.cfm_facade.wait_for_hangouts_telemetry_commands() |
| |
| self._set_preferred_speaker_mixer() |
| logging.info('1. Check CfM and dual speakers have the same setting ' |
| 'after joining meeting:') |
| self.cfm_facade.start_new_hangout_session('test_cfm_dual_speaker') |
| if not self._test_dual_speaker_sanity(): |
| raise error.TestFail('Dual speaker Sanity verification fails.') |
| |
| logging.info('1.1 Check CfM and dual microphones have the same ' |
| 'setting for Mute/unmute:') |
| if not self._test_mute_sync(): |
| raise error.TestFail('Dual speaker Mute-unmute test' |
| ' verification fails') |
| |
| logging.info('1.2 Check CfM and dual Speakers have same volume: ') |
| if not self._test_volume_sync(): |
| raise error.TestFail('Dual speaker volume test verification fails') |
| |
| for gpio in gpio_list: |
| |
| logging.info('2. Check CfM and speakers have the same setting ' |
| 'after flapping speaker:') |
| logging.info('Power cycle one of Speaker %s', gpio) |
| power_cycle_usb_util.power_cycle_usb_gpio(self._host, |
| gpio, TIMEOUT_SECS) |
| time.sleep(TIMEOUT_SECS) |
| |
| logging.info('2.1. Check CfM and dual speakers have same setting') |
| if not self._test_dual_speaker_sanity(): |
| raise error.TestFail('Dual speaker Sanity verification' |
| ' fails after disconnect/reconnect speaker') |
| |
| logging.info('2.1.1. Check CfM and microphones have same setting ' |
| 'for Mute/unmute:') |
| if not self._test_mute_sync(): |
| raise error.TestFail( |
| 'Dual speaker Mute-unmute test verification fails after ' |
| 'disconnect/reconnect speaker') |
| |
| logging.info('2.1.2. Check CfM and dual Speakers have same volume:') |
| if not self._test_volume_sync(): |
| raise error.TestFail('Dual speaker volume test verification' |
| ' fails after disconnect/reconnect speaker') |
| |
| self.cfm_facade.end_hangout_session() |
