client/site_tests/policy_AudioOutputAllowed/policy_AudioOutputAllowed.py

# Copyright 2018 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 os
import utils

from autotest_lib.client.common_lib import error
from autotest_lib.client.cros.audio import audio_helper
from autotest_lib.client.cros.audio import cmd_utils
from autotest_lib.client.cros.audio import cras_utils
from autotest_lib.client.cros.enterprise import enterprise_policy_base
from autotest_lib.client.cros.input_playback import input_playback


class policy_AudioOutputAllowed(
        enterprise_policy_base.EnterprisePolicyTest):
    version = 1

    POLICY_NAME = 'AudioOutputAllowed'
    # How long (sec) to capture output for
    SAMPLE_DURATION = 1

    TEST_CASES = {
        'NotSet_Allow': None,
        'True_Allow': True,
        'False_Block': False
    }

    def initialize(self, **kwargs):
        """Initialize objects for test."""
        super(policy_AudioOutputAllowed, self).initialize(**kwargs)
        audio_helper.cras_rms_test_setup()

    def wait_for_active_stream_count(self, expected_count):
        """
        Waits for there to be the expected number of audio streams.

        @param expected_count: Number of audio streams to wait for.

        @raises error.TestError: if there is a timeout before the there is the
        desired number of audio streams.

        """
        utils.poll_for_condition(
            lambda: cras_utils.get_active_stream_count() == expected_count,
            exception=error.TestError(
                'Timeout waiting active stream count to become %d' %
                 expected_count))


    def is_muted(self):
        """
        Returns mute status of system.

        @returns: True if system muted, False if not.

        """
        MUTE_STATUS = 'Muted'
        CTC_GREP_FOR_MUTED = 'cras_test_client --dump_server_info | grep muted'

        output = utils.system_output(CTC_GREP_FOR_MUTED)
        muted = output.split(':')[-1].strip()
        return muted == MUTE_STATUS


    def _test_audio_disabled(self, policy_value):
        """
        Verify the AudioOutputAllowed policy behaves as expected.

        Generate and play a sample audio file. When enabled, the difference
        between the muted and unmuted RMS should be greater than 0.75. When
        disabled, the RMS difference should be less than 0.05.

        @param policy_value: policy value for this case.

        @raises error.TestFail: In the case where the audio behavior
            does not match the policy value.

        """
        audio_allowed = policy_value or policy_value is None

        RAW_FILE = os.path.join(self.enterprise_dir, 'test_audio.raw')
        noise_file = os.path.join(self.resultsdir, 'noise.wav')
        recorded_file = os.path.join(self.resultsdir, 'recorded-cras.raw')
        recorded_rms = []

        # Record a sample of silence to use as a noise profile.
        cras_utils.capture(noise_file, duration=2)
        logging.info('NOISE: %s', audio_helper.get_rms(noise_file))

        # Get two RMS samples: one when muted and one when not
        for muted in [False, True]:
            cras_utils.set_system_mute(muted)

            # Play the audio file and capture the output
            self.wait_for_active_stream_count(0)
            p = cmd_utils.popen(cras_utils.playback_cmd(RAW_FILE))
            try:
                self.wait_for_active_stream_count(1)
                cras_utils.capture(recorded_file, duration=self.SAMPLE_DURATION)

                if p.poll() is not None:
                    raise error.TestError('Audio playback stopped prematurely')
            finally:
                cmd_utils.kill_or_log_returncode(p)

            rms_value = audio_helper.reduce_noise_and_get_rms(
                recorded_file, noise_file)[0]

            logging.info('muted (%s): %s' % (muted, rms_value))
            recorded_rms.append(rms_value)

        rms_diff = recorded_rms[0] - recorded_rms[1]
        self.write_perf_keyval({'rms_diff': rms_diff})

        if audio_allowed:
            if rms_diff < 0.4:
                raise error.TestFail('RMS difference not large enough between '
                                     'mute and ummute: %s' % rms_diff)
        else:
            if abs(rms_diff) > 0.05:
                raise error.TestFail('RMS difference too wide while audio '
                                     'disabled: %s' % rms_diff)


    def _test_unmute_disabled(self, policy_value):
        """
        Verify AudioOutputAllowed does not allow unmuting when disabled.

        Attempt to unmute the system with CRAS and check the system state
        after.

        @param policy_value: policy value for this case.

        @raises error.TestFail: In the case where the audio behavior
            does not match the policy value.

        """
        audio_allowed = policy_value or policy_value is None

        cras_utils.set_system_mute(False)

        if not audio_allowed and not self.is_muted():
            raise error.TestFail('System should be muted, but is not')
        elif audio_allowed and self.is_muted():
            raise error.TestFail('System is muted but should not be')


    def run_once(self, case):
        """
        Setup and run the test configured for the specified test case.

        @param case: Name of the test case to run.

        """
        case_value = self.TEST_CASES[case]
        self.setup_case(user_policies={self.POLICY_NAME: case_value})
        self._test_audio_disabled(case_value)
        self._test_unmute_disabled(case_value)