Various test improvement, especially for testing minimal templates

qubes/storage/lvm.py

@@ -266,7 +266,7 @@ def _parse_lvm_cache(lvm_output):
             usage = int(size / 100 * float(usage_percent))
         else:
             usage = 0
-        if metadata_size:
+        if metadata_size and metadata_percent:
             metadata_size = int(metadata_size[:-1])
             metadata_usage = int(metadata_size / 100 * float(metadata_percent))
         else:

qubes/tests/__init__.py

@@ -154,6 +154,22 @@ def skipUnlessEnv(varname):
     return unittest.skipUnless(os.getenv(varname), "no {} set".format(varname))
 
 
+def skipIfTemplate(*templates):
+    """Decorator generator for skipping on specific templates.
+
+    Some tests are supported only on some templates. This decorator allows
+    excluding test for some of them, especially useful for excluding tests on
+    minimal templates or Whonix.
+    Multiple templates can be given.
+    """
+
+    def decorator(func):
+        func.__qubestest_skip_templates__ = templates
+        return func
+
+    return decorator
+
+
 class TestEmitter(qubes.events.Emitter):
     """Dummy event emitter which records events fired on it.
 
@@ -253,6 +269,8 @@ def wait_on_fail(func):
     def wrapper(self, *args, **kwargs):
         try:
             func(self, *args, **kwargs)
+        except unittest.case.SkipTest:
+            raise
         except:
             print("FAIL\n")
             traceback.print_exc()
@@ -439,9 +457,17 @@ class QubesTestCase(unittest.TestCase):
     def __init__(self, methodName="runTest"):
         try:
             test_method = getattr(self, methodName)
-            setattr(
-                self, methodName, _clear_ex_info(self.set_result)(test_method)
-            )
+            skip_templates = getattr(
+                self, "__qubestest_skip_templates__", []
+            ) or getattr(test_method, "__qubestest_skip_templates__", [])
+            template = getattr(self, "template", "")
+            if any(skip in template for skip in skip_templates):
+                test_method = unittest.skip(
+                    f"Test skipped on template {template}"
+                )(test_method)
+            else:
+                test_method = _clear_ex_info(self.set_result)(test_method)
+            setattr(self, methodName, test_method)
         except AttributeError:
             pass
         super(QubesTestCase, self).__init__(methodName)

qubes/tests/integ/audio.py

@@ -34,6 +34,7 @@
 from qubes.tests.integ.vm_qrexec_gui import TC_00_AppVMMixin, in_qemu
 
 
+@qubes.tests.skipIfTemplate("whonix-g")
 class TC_00_AudioMixin(TC_00_AppVMMixin):
     def wait_for_pulseaudio_startup(self, vm):
         self.loop.run_until_complete(self.wait_for_session(self.testvm1))
@@ -64,8 +65,6 @@ def wait_for_pulseaudio_startup(self, vm):
         self.loop.run_until_complete(asyncio.sleep(1))
 
     def prepare_audio_test(self, backend):
-        if "whonix-g" in self.template:
-            self.skipTest("whonix gateway have no audio")
         self.loop.run_until_complete(self.testvm1.start())
         pulseaudio_units = "pulseaudio.socket pulseaudio.service"
         pipewire_units = "pipewire.socket wireplumber.service pipewire.service"
@@ -253,47 +252,52 @@ def common_audio_playback(self):
             p.wait()
             self.check_audio_sample(recorded_audio.file.read(), sfreq)
 
-    def _configure_audio_recording(self, vm):
-        """Connect VM's source-output to sink monitor instead of mic"""
+    def _call_in_audiovm(self, audiovm, command):
         local_user = grp.getgrnam("qubes").gr_mem[0]
-        audiovm = vm.audiovm
-
         sudo = ["sudo", "-E", "-u", local_user]
-
-        source_outputs_cmd = ["pactl", "-f", "json", "list", "source-outputs"]
         if audiovm.name != "dom0":
             stdout, _ = self.loop.run_until_complete(
-                audiovm.run_for_stdio(" ".join(source_outputs_cmd))
+                audiovm.run_for_stdio(" ".join(command))
             )
-            source_outputs = json.loads(stdout)
+            return stdout
         else:
-            source_outputs = json.loads(
-                subprocess.check_output(sudo + source_outputs_cmd)
-            )
-
-        if not source_outputs:
-            self.fail("no source-output found in {}".format(audiovm.name))
-            assert False
+            return subprocess.check_output(sudo + command)
 
+    def _find_pactl_entry_for_vm(self, pactl_data, vm_name):
         try:
-            output_index = [
-                s["index"]
-                for s in source_outputs
-                if s["properties"].get("application.name") == vm.name
+            return [
+                s
+                for s in pactl_data
+                if s["properties"].get("application.name") == vm_name
             ][0]
         except IndexError:
             self.fail("source-output for VM {} not found".format(vm.name))
             # self.fail never returns
             assert False
 
-        sources_cmd = ["pactl", "-f", "json", "list", "sources"]
-        if audiovm.name != "dom0":
-            res, _ = self.loop.run_until_complete(
-                audiovm.run_for_stdio(" ".join(sources_cmd))
+    def _configure_audio_recording(self, vm):
+        """Connect VM's source-output to sink monitor instead of mic"""
+        audiovm = vm.audiovm
+
+        source_outputs = json.loads(
+            self._call_in_audiovm(
+                audiovm, ["pactl", "-f", "json", "list", "source-outputs"]
             )
-            sources = json.loads(res)
-        else:
-            sources = json.loads(subprocess.check_output(sudo + sources_cmd))
+        )
+
+        if not source_outputs:
+            self.fail("no source-output found in {}".format(audiovm.name))
+            assert False
+
+        output_info = self._find_pactl_entry_for_vm(source_outputs, vm.name)
+        output_index = output_info["index"]
+        current_source = output_info["source"]
+
+        sources = json.loads(
+            self._call_in_audiovm(
+                audiovm, ["pactl", "-f", "json", "list", "sources"]
+            )
+        )
 
         if not sources:
             self.fail("no sources found in {}".format(audiovm.name))
@@ -308,16 +312,31 @@ def _configure_audio_recording(self, vm):
             # self.fail never returns
             assert False
 
-        cmd = [
-            "pactl",
-            "move-source-output",
-            str(output_index),
-            str(source_index),
-        ]
-        if audiovm.name != "dom0":
-            self.loop.run_until_complete(audiovm.run(" ".join(cmd)))
-        else:
-            subprocess.check_call(sudo + cmd)
+        attempts_left = 5
+        # pactl seems to fail sometimes, still with exit code 0...
+        while current_source != source_index and attempts_left:
+            assert isinstance(output_index, int)
+            assert isinstance(source_index, int)
+            cmd = [
+                "pactl",
+                "move-source-output",
+                str(output_index),
+                str(source_index),
+            ]
+            self._call_in_audiovm(audiovm, cmd)
+
+            source_outputs = json.loads(
+                self._call_in_audiovm(
+                    audiovm, ["pactl", "-f", "json", "list", "source-outputs"]
+                )
+            )
+
+            output_info = self._find_pactl_entry_for_vm(source_outputs, vm.name)
+            output_index = output_info["index"]
+            current_source = output_info["source"]
+            attempts_left -= 1
+
+        self.assertGreater(attempts_left, 0, "Failed to move-source-output")
 
     async def retrieve_audio_input(self, vm, status):
         try:

qubes/tests/integ/basic.py

@@ -636,7 +636,7 @@ async def _test_clipboard(
         # correct timestamp (so gui-daemon would not drop the copy request)
         subprocess.check_call(["xdotool", "key", "ctrl+a", "ctrl+c"])
         # wait a bit to let the zenity actually copy
-        await asyncio.sleep(1)
+        await asyncio.sleep(5)
         subprocess.check_call(["xdotool", "key", "ctrl+shift+c", "Escape"])
 
         await self.wait_for_window_coro(window_title, show=False)

qubes/tests/integ/grub.py

@@ -46,6 +46,13 @@ def setUp(self):
             )
 
     def install_packages(self, vm):
+        if os.environ.get("QUBES_TEST_SKIP_KERNEL_INSTALL") == "1":
+            return
+        else:
+            print(
+                "Installing kernel packages, you can skip by setting "
+                "QUBES_TEST_SKIP_KERNEL_INSTALL=1 in environment"
+            )
         if self.template.startswith("fedora-"):
             cmd_install1 = (
                 "dnf clean expire-cache && "
@@ -122,6 +129,11 @@ def test_000_standalone_vm(self):
         self.loop.run_until_complete(self.testvm1.shutdown(wait=True))
 
         self.testvm1.kernel = self.kernel
+        if self.virt_mode == "hvm":
+            # HVM has disabled memory-hotplug, which means VM is started with
+            # full maxmem and need extra memory for page structures for full
+            # maxmem
+            self.testvm1.memory = 450
         self.loop.run_until_complete(self.testvm1.start())
         (actual_kver, _) = self.loop.run_until_complete(
             self.testvm1.run_for_stdio("uname -r")
@@ -149,15 +161,18 @@ def test_010_template_based_vm(self):
             name=self.make_vm_name("vm1"),
             label="red",
         )
-        self.testvm1.virt_mode = self.virt_mode
         self.loop.run_until_complete(self.testvm1.create_on_disk())
         self.loop.run_until_complete(self.test_template.start())
         self.install_packages(self.test_template)
         kver = self.get_kernel_version(self.test_template)
         self.loop.run_until_complete(self.test_template.shutdown(wait=True))
 
         self.test_template.kernel = self.kernel
-        self.testvm1.kernel = self.kernel
+        if self.virt_mode == "hvm":
+            # HVM has disabled memory-hotplug, which means VM is started with
+            # full maxmem and need extra memory for page structures for full
+            # maxmem
+            self.test_template.memory = 450
 
         # Check if TemplateBasedVM boots and has the right kernel
         self.loop.run_until_complete(self.testvm1.start())

qubes/tests/integ/network.py

@@ -34,6 +34,7 @@
 
 
 # noinspection PyAttributeOutsideInit,PyPep8Naming
+@qubes.tests.skipIfTemplate("whonix")
 class VmNetworkingMixin(object):
     test_ip = "192.168.123.45"
     test_name = "test.example.com"
@@ -71,11 +72,6 @@ def setUp(self):
                 "Test not supported here - Whonix uses its own "
                 "firewall settings"
             )
-        if self.template.endswith("-minimal"):
-            self.skipTest(
-                "Test not supported here - minimal template don't have "
-                "networking packages by default"
-            )
         self.init_default_template(self.template)
         self.testnetvm = self.app.add_new_vm(
             qubes.vm.appvm.AppVM, name=self.make_vm_name("netvm1"), label="red"

qubes/tests/integ/salt.py

@@ -311,12 +311,28 @@ def setUp(self):
         super(SaltVMTestMixin, self).setUp()
         self.init_default_template(self.template)
 
+        mgmt_tpl = self.app.domains[self.template]
+        if "minimal" in self.template:
+            # minimal template doesn't support being mgmt vm, but still test
+            # it being a target
+            mgmt_tpl = os.environ.get("QUBES_TEST_MGMT_TPL")
+            if not mgmt_tpl:
+                mgmt_tpl = str(self.host_app.default_template)
+            print(
+                f"Using {mgmt_tpl} template for mgmt vm when testing "
+                f"minimal template as target. You can set "
+                f"QUBES_TEST_MGMT_TPL env variable to use "
+                f"different template for mgmt vm"
+            )
+            mgmt_tpl = self.app.domains[mgmt_tpl]
+
         dispvm_tpl_name = self.make_vm_name("disp-tpl")
         dispvm_tpl = self.app.add_new_vm(
             "AppVM",
             label="red",
             template_for_dispvms=True,
             name=dispvm_tpl_name,
+            template=mgmt_tpl,
         )
         self.loop.run_until_complete(dispvm_tpl.create_on_disk())
         self.app.default_dispvm = dispvm_tpl
@@ -611,8 +627,10 @@ def test_002_grains_id(self):
 
     def test_003_update(self):
         vmname = self.make_vm_name("target")
-        self.vm = self.app.add_new_vm("AppVM", name=vmname, label="red")
-        self.loop.run_until_complete(self.vm.create_on_disk())
+        self.vm = self.app.add_new_vm("TemplateVM", name=vmname, label="red")
+        self.loop.run_until_complete(
+            self.vm.clone_disk_files(self.app.default_template)
+        )
         # start the VM manually, so it stays running after applying salt state
         self.loop.run_until_complete(self.vm.start())
         state_output = self.salt_call(

qubes/tests/integ/vm_qrexec_gui.py

@@ -123,7 +123,9 @@ def test_011_run_gnome_terminal(self):
         self.loop.run_until_complete(self.testvm1.start())
         self.assertEqual(self.testvm1.get_power_state(), "Running")
         self.loop.run_until_complete(self.wait_for_session(self.testvm1))
-        p = self.loop.run_until_complete(self.testvm1.run("gnome-terminal"))
+        p = self.loop.run_until_complete(
+            self.testvm1.run("gnome-terminal || " "ptyxis")
+        )
         try:
             title = "user@{}".format(self.testvm1.name)
             if self.template.count("whonix"):
@@ -156,7 +158,7 @@ def test_011_run_gnome_terminal(self):
                 wait_count += 1
                 if wait_count > 100:
                     self.fail(
-                        "Timeout while waiting for gnome-terminal "
+                        "Timeout while waiting for gnome-terminal/ptyxis "
                         "termination"
                     )
                 self.loop.run_until_complete(asyncio.sleep(0.1))