diff --git a/qubes/storage/lvm.py b/qubes/storage/lvm.py index 1012bb3fb..dcd9e8a54 100644 --- a/qubes/storage/lvm.py +++ b/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: diff --git a/qubes/tests/__init__.py b/qubes/tests/__init__.py index 4189ed06c..1fd5894df 100644 --- a/qubes/tests/__init__.py +++ b/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) diff --git a/qubes/tests/integ/audio.py b/qubes/tests/integ/audio.py index 52311f41b..4c1fcbf39 100644 --- a/qubes/tests/integ/audio.py +++ b/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: diff --git a/qubes/tests/integ/basic.py b/qubes/tests/integ/basic.py index 19deccbf1..54be7b0b3 100644 --- a/qubes/tests/integ/basic.py +++ b/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) diff --git a/qubes/tests/integ/grub.py b/qubes/tests/integ/grub.py index 9acde4060..d0f3e3012 100644 --- a/qubes/tests/integ/grub.py +++ b/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,7 +161,6 @@ 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) @@ -157,7 +168,11 @@ def test_010_template_based_vm(self): 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()) diff --git a/qubes/tests/integ/network.py b/qubes/tests/integ/network.py index 76eae9674..b7e22b4f9 100644 --- a/qubes/tests/integ/network.py +++ b/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" diff --git a/qubes/tests/integ/salt.py b/qubes/tests/integ/salt.py index 19bfde313..3c33ae6d9 100644 --- a/qubes/tests/integ/salt.py +++ b/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( diff --git a/qubes/tests/integ/vm_qrexec_gui.py b/qubes/tests/integ/vm_qrexec_gui.py index 1f9c0e02f..021c688e5 100644 --- a/qubes/tests/integ/vm_qrexec_gui.py +++ b/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)) diff --git a/qubes/tests/integ/vm_update.py b/qubes/tests/integ/vm_update.py index 819c45fed..3d661b2db 100644 --- a/qubes/tests/integ/vm_update.py +++ b/qubes/tests/integ/vm_update.py @@ -244,6 +244,15 @@ def setUp(self): self.loop.run_until_complete(self.testvm1.create_on_disk()) self.repo_proc = None + # template used for repo-hosting vm + self.repo_template = self.app.default_template + if self.template.count("minimal"): + self.repo_template = self.host_app.default_template + print( + f"Using {self.repo_template!s} for repo hosting vm when " + f"testing minimal template" + ) + def tearDown(self): if self.repo_proc: self.repo_proc.terminate() @@ -256,6 +265,10 @@ def test_000_simple_update(self): :type self: qubes.tests.SystemTestCase | VmUpdatesMixin """ + if self.template.count("minimal"): + self.skipTest( + "Template {} not supported by this test".format(self.template) + ) self.app.save() self.testvm1 = self.app.domains[self.testvm1.qid] self.loop.run_until_complete(self.testvm1.start()) @@ -396,6 +409,13 @@ def create_repo_and_serve(self): "Template {} not supported by this test".format(self.template) ) + # wait for the repo to become reachable + self.loop.run_until_complete( + self.netvm_repo.run_for_stdio( + "while ! curl http://localhost:8080/ >/dev/null; do sleep 0.5; done" + ) + ) + def add_update_to_repo(self): """ :type self: qubes.tests.SystemTestCase | VmUpdatesMixin @@ -449,7 +469,10 @@ def start_vm_with_proxy_repo(self): :type self: qubes.tests.SystemTestCase | VmUpdatesMixin """ self.netvm_repo = self.app.add_new_vm( - qubes.vm.appvm.AppVM, name=self.make_vm_name("net"), label="red" + qubes.vm.appvm.AppVM, + name=self.make_vm_name("net"), + label="red", + template=self.repo_template, ) self.netvm_repo.provides_network = True self.loop.run_until_complete(self.netvm_repo.create_on_disk()) @@ -574,11 +597,6 @@ def update_via_proxy_qubes_vm_update_impl( :type break_repo: bool :type: expect_updated: bool """ - if self.template.count("minimal"): - self.skipTest( - "Template {} not supported by this test".format(self.template) - ) - if expected_ret_codes is None: expected_ret_codes = self.ret_code_ok @@ -643,7 +661,7 @@ def upgrade_status_notify(self): """ self.loop.run_until_complete( self.testvm1.run_for_stdio( - "/usr/lib/qubes/upgrades-status-notify", + "/usr/lib/qubes/upgrades-status-notify 2>/dev/console", user="root", ) )