Integration tests for audio-initial-volume feature

qubes/tests/integ/audio.py

@@ -125,6 +125,8 @@ def create_audio_vm(self, backend, start=True):
 admin.vm.property.Get               +is_preload     {vm}     @tag:audiovm-{vm}  allow   target=dom0
 admin.vm.feature.CheckWithTemplate  +audio   {vm}     @tag:audiovm-{vm}  allow   target=dom0
 admin.vm.feature.CheckWithTemplate  +audio-model   {vm}     @tag:audiovm-{vm}  allow   target=dom0
+admin.vm.feature.CheckWithTemplate  +audio-initial-volume   {vm}     @tag:audiovm-{vm}  allow   target=dom0
+admin.vm.feature.CheckWithTemplate  +audio-initial-volume   {vm}     {vm}  allow   target=dom0
 """.format(
                     vm=self.audiovm.name
                 )
@@ -368,6 +370,59 @@ def _configure_audio_recording(self, vm, expect_stream=True):
 
         self.assertGreater(attempts_left, 0, "Failed to move-source-output")
 
+    def _get_sink_volume(self, vm) -> str:
+        """Return VM's sink-input volume percent or `mute` if mute"""
+        audiovm = vm.audiovm
+
+        sinks = json.loads(
+            self._call_in_audiovm(
+                audiovm, ["pactl", "-f", "json", "list", "sink-inputs"]
+            )
+        )
+
+        if not sinks:
+            self.fail("no sink-inputs found in {}".format(audiovm.name))
+            assert False
+
+        for sink in sinks:
+            if sink["properties"]["application.name"] == vm.name:
+                if sink["mute"]:
+                    return "mute"
+                vol_l = sink["volume"]["front-left"]["value_percent"]
+                vol_r = sink["volume"]["front-right"]["value_percent"]
+                assert vol_l == vol_r
+                return vol_l
+
+        self.fail("{} sink-input not found in {}".format(vm.name, audiovm.name))
+        assert False
+
+    def _reset_sink_volume(self, vm) -> None:
+        """Reset VM's sink-input volume to 100% and unmuted"""
+        audiovm = vm.audiovm
+
+        sinks = json.loads(
+            self._call_in_audiovm(
+                audiovm, ["pactl", "-f", "json", "list", "sink-inputs"]
+            )
+        )
+
+        if not sinks:
+            self.fail("no sink-inputs found in {}".format(audiovm.name))
+            assert False
+
+        for sink in sinks:
+            if sink["properties"]["application.name"] == vm.name:
+                index = str(sink["index"])
+                self._call_in_audiovm(
+                    audiovm,
+                    ["pactl", "set-sink-input-volume", index, "100%"],
+                )
+                self._call_in_audiovm(
+                    audiovm,
+                    ["pactl", "set-sink-input-mute", index, "0"],
+                )
+                break
+
     async def retrieve_audio_input(self, vm, status):
         try:
             await asyncio.wait_for(
@@ -748,6 +803,36 @@ def test_260_audio_mic_enabled_switch_audiovm(self):
         self.assert_pacat_running(self.audiovm, self.testvm1, True)
         self.common_audio_record_muted()
 
+    @unittest.skipUnless(
+        spawn.find_executable("pactl"),
+        "pulseaudio-utils not installed in dom0",
+    )
+    def test_261_audio_initial_volume_42_percent(self):
+        self.testvm1.features["audio-initial-volume"] = "42"
+        self.loop.run_until_complete(self.testvm1.start())
+        self.wait_for_pulseaudio_startup(self.testvm1)
+        try:
+            assert self._get_sink_volume(self.testvm1) == "42%"
+        except AssertionError:
+            raise
+        finally:
+            self._reset_sink_volume(self.testvm1)
+
+    @unittest.skipUnless(
+        spawn.find_executable("pactl"),
+        "pulseaudio-utils not installed in dom0",
+    )
+    def test_262_audio_initial_volume_mute(self):
+        self.testvm1.features["audio-initial-volume"] = "mute"
+        self.loop.run_until_complete(self.testvm1.start())
+        self.wait_for_pulseaudio_startup(self.testvm1)
+        try:
+            assert self._get_sink_volume(self.testvm1) == "mute"
+        except AssertionError:
+            raise
+        finally:
+            self._reset_sink_volume(self.testvm1)
+
 
 def create_testcases_for_templates():
     yield from qubes.tests.create_testcases_for_templates(