1# Copyright 2018 The Chromium OS Authors. All rights reserved.
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5import logging
6import os
7import utils
8
9from autotest_lib.client.common_lib import error
10from autotest_lib.client.cros.audio import audio_helper
11from autotest_lib.client.cros.audio import cmd_utils
12from autotest_lib.client.cros.audio import cras_utils
13from autotest_lib.client.cros.enterprise import enterprise_policy_base
14from autotest_lib.client.cros.input_playback import input_playback
15
16
17class policy_AudioOutputAllowed(
18        enterprise_policy_base.EnterprisePolicyTest):
19    version = 1
20
21    POLICY_NAME = 'AudioOutputAllowed'
22    # How long (sec) to capture output for
23    SAMPLE_DURATION = 1
24
25    TEST_CASES = {
26        'NotSet_Allow': None,
27        'True_Allow': True,
28        'False_Block': False
29    }
30
31    def initialize(self, **kwargs):
32        """Initialize objects for test."""
33        super(policy_AudioOutputAllowed, self).initialize(**kwargs)
34        audio_helper.cras_rms_test_setup()
35
36    def wait_for_active_stream_count(self, expected_count):
37        """
38        Waits for there to be the expected number of audio streams.
39
40        @param expected_count: Number of audio streams to wait for.
41
42        @raises error.TestError: if there is a timeout before the there is the
43        desired number of audio streams.
44
45        """
46        utils.poll_for_condition(
47            lambda: cras_utils.get_active_stream_count() == expected_count,
48            exception=error.TestError(
49                'Timeout waiting active stream count to become %d' %
50                 expected_count))
51
52
53    def is_muted(self):
54        """
55        Returns mute status of system.
56
57        @returns: True if system muted, False if not.
58
59        """
60        MUTE_STATUS = 'Muted'
61        CTC_GREP_FOR_MUTED = 'cras_test_client --dump_server_info | grep muted'
62
63        output = utils.system_output(CTC_GREP_FOR_MUTED)
64        muted = output.split(':')[-1].strip()
65        return muted == MUTE_STATUS
66
67
68    def _test_audio_disabled(self, policy_value):
69        """
70        Verify the AudioOutputAllowed policy behaves as expected.
71
72        Generate and play a sample audio file. When enabled, the difference
73        between the muted and unmuted RMS should be greater than 0.75. When
74        disabled, the RMS difference should be less than 0.05.
75
76        @param policy_value: policy value for this case.
77
78        @raises error.TestFail: In the case where the audio behavior
79            does not match the policy value.
80
81        """
82        audio_allowed = policy_value or policy_value is None
83
84        RAW_FILE = os.path.join(self.enterprise_dir, 'test_audio.raw')
85        noise_file = os.path.join(self.resultsdir, 'noise.wav')
86        recorded_file = os.path.join(self.resultsdir, 'recorded-cras.raw')
87        recorded_rms = []
88
89        # Record a sample of silence to use as a noise profile.
90        cras_utils.capture(noise_file, duration=2)
91        logging.info('NOISE: %s', audio_helper.get_rms(noise_file))
92
93        # Get two RMS samples: one when muted and one when not
94        for muted in [False, True]:
95            cras_utils.set_system_mute(muted)
96
97            # Play the audio file and capture the output
98            self.wait_for_active_stream_count(0)
99            p = cmd_utils.popen(cras_utils.playback_cmd(RAW_FILE))
100            try:
101                self.wait_for_active_stream_count(1)
102                cras_utils.capture(recorded_file, duration=self.SAMPLE_DURATION)
103
104                if p.poll() is not None:
105                    raise error.TestError('Audio playback stopped prematurely')
106            finally:
107                cmd_utils.kill_or_log_returncode(p)
108
109            rms_value = audio_helper.reduce_noise_and_get_rms(
110                recorded_file, noise_file)[0]
111
112            logging.info('muted (%s): %s' % (muted, rms_value))
113            recorded_rms.append(rms_value)
114
115        rms_diff = recorded_rms[0] - recorded_rms[1]
116        self.write_perf_keyval({'rms_diff': rms_diff})
117
118        if audio_allowed:
119            if rms_diff < 0.4:
120                raise error.TestFail('RMS difference not large enough between '
121                                     'mute and ummute: %s' % rms_diff)
122        else:
123            if abs(rms_diff) > 0.05:
124                raise error.TestFail('RMS difference too wide while audio '
125                                     'disabled: %s' % rms_diff)
126
127
128    def _test_unmute_disabled(self, policy_value):
129        """
130        Verify AudioOutputAllowed does not allow unmuting when disabled.
131
132        Attempt to unmute the system with CRAS and check the system state
133        after.
134
135        @param policy_value: policy value for this case.
136
137        @raises error.TestFail: In the case where the audio behavior
138            does not match the policy value.
139
140        """
141        audio_allowed = policy_value or policy_value is None
142
143        cras_utils.set_system_mute(False)
144
145        if not audio_allowed and not self.is_muted():
146            raise error.TestFail('System should be muted, but is not')
147        elif audio_allowed and self.is_muted():
148            raise error.TestFail('System is muted but should not be')
149
150
151    def run_once(self, case):
152        """
153        Setup and run the test configured for the specified test case.
154
155        @param case: Name of the test case to run.
156
157        """
158        case_value = self.TEST_CASES[case]
159        self.setup_case(user_policies={self.POLICY_NAME: case_value})
160        self._test_audio_disabled(case_value)
161        self._test_unmute_disabled(case_value)
162