Improve preloaded dispvm tests

Author: ben-grande

ci/requirements.txt

@@ -8,3 +8,4 @@ pylint
 sphinx
 PyYAML
 pyinotify
+psutil

linux/aux-tools/Makefile

@@ -3,6 +3,7 @@ all:
 
 install:
 	mkdir -p $(DESTDIR)/usr/lib/qubes
-	cp cleanup-dispvms $(DESTDIR)/usr/lib/qubes
-	cp startup-misc.sh $(DESTDIR)/usr/lib/qubes
+	cp preload-dispvm $(DESTDIR)/usr/lib/qubes/
+	cp cleanup-dispvms $(DESTDIR)/usr/lib/qubes/
+	cp startup-misc.sh $(DESTDIR)/usr/lib/qubes/
 	cp fix-dir-perms.sh $(DESTDIR)/usr/lib/qubes/

linux/aux-tools/preload-dispvm

@@ -1,32 +1,21 @@
 #!/usr/bin/env python3
 
-#import asyncio
 import qubesadmin
 
-def get_apps():
-    domains = qubesadmin.Qubes().domains
-    return [qube for qube in domains
-    if int(qube.features.get("dispvm-preload-max", 0)) > 0
-    and qube.klass == "AppVM"
-    and getattr(qube, "template_for_dispvms", False)
-    ]
-
-#async def main():
-#    appvms = get_apps()
-#    ## TODO: How to fire events outside of qubesd? How to load DispVM by hand.
-#    #o = MyClass()
-#    #o.events_enabled = True
-#    #effect = o.fire_event('event1')
-#    event = "domain-preloaded-dispvm-autostart"
-#    tasks = [qube.fire_event_async(event) for qube in appvms]
-#    await asyncio.gather(*tasks)
 
 def main():
-    appvms = get_apps()
+    domains = qubesadmin.Qubes().domains
+    appvms = [
+        qube
+        for qube in domains
+        if int(qube.features.get("dispvm-preload-max", 0)) > 0
+        and qube.klass == "AppVM"
+        and getattr(qube, "template_for_dispvms", False)
+    ]
     method = "admin.vm.CreateDisposable"
     for qube in appvms:
         qube.qubesd_call("dom0", method, "preload-autostart")
 
+
 if __name__ == "__main__":
-    #asyncio.run(main())
     main()

linux/systemd/qubes-preload-dispvm.service

@@ -1,14 +1,9 @@
 [Unit]
 Description=Preload Qubes DispVMs
-After=qubes-vm@.service
-# TODO: or 'Requires='? Should a failure to autostart sys-(usb|net) make this
-# unit be skipped?
-Wants=qubes-vm@.service
 ConditionKernelCommandLine=!qubes.skip_autostart
 
 [Service]
 Type=oneshot
-Environment=DISPLAY=:0
 ExecStart=/usr/lib/qubes/preload-dispvm
 Group=qubes
 RemainAfterExit=yes

linux/systemd/qubes-vm@.service

@@ -1,12 +1,12 @@
 [Unit]
 Description=Start Qubes VM %i
 After=qubesd.service qubes-meminfo-writer-dom0.service
+Before=qubes-vm@.service
 ConditionKernelCommandLine=!qubes.skip_autostart
 
 [Service]
 Type=oneshot
-Environment=DISPLAY=:0
-ExecStart=/usr/bin/qvm-start --skip-if-running %i
+ExecStart=/usr/bin/qvm-start --skip-if-running -- %i
 Group=qubes
 RemainAfterExit=yes
 

qubes/api/admin.py

@@ -298,7 +298,9 @@ async def vm_property_set(self, untrusted_payload):
     async def property_set(self, untrusted_payload):
         """Set property value"""
         self.enforce(self.dest.name == "dom0")
-        return self._property_set(self.app, untrusted_payload=untrusted_payload)
+        return self._property_set(
+            self.app, untrusted_payload=untrusted_payload
+        )
 
     def _property_set(self, dest, untrusted_payload):
         if self.arg not in dest.property_list():
@@ -484,7 +486,9 @@ async def vm_volume_clone_to(self, untrusted_payload):
 
         # make sure the volume still exists, but invalidate token anyway
         self.enforce(str(src_volume.pool) in self.app.pools)
-        self.enforce(src_volume in self.app.pools[str(src_volume.pool)].volumes)
+        self.enforce(
+            src_volume in self.app.pools[str(src_volume.pool)].volumes
+        )
 
         dst_volume = self.dest.volumes[self.arg]
 
@@ -766,7 +770,9 @@ async def pool_add(self, untrusted_payload):
 
         driver_parameters = qubes.storage.driver_parameters(self.arg)
         unexpected_parameters = [
-            key for key in untrusted_pool_config if key not in driver_parameters
+            key
+            for key in untrusted_pool_config
+            if key not in driver_parameters
         ]
         if unexpected_parameters:
             raise qubes.exc.QubesException(
@@ -1114,7 +1120,9 @@ async def vm_feature_checkwithtpladminvm(self):
 
         self.fire_event_for_permission()
         try:
-            value = self.dest.features.check_with_template_and_adminvm(self.arg)
+            value = self.dest.features.check_with_template_and_adminvm(
+                self.arg
+            )
         except KeyError:
             raise qubes.exc.QubesFeatureNotFoundError(self.dest, self.arg)
         return value
@@ -1286,20 +1294,18 @@ async def create_disposable(self, untrusted_payload):
             )
 
         if self.dest.name == "dom0":
-            dispvm_template = self.src.default_dispvm
+            appvm = self.src.default_dispvm
         else:
-            dispvm_template = self.dest
+            appvm = self.dest
 
-        self.fire_event_for_permission(dispvm_template=dispvm_template)
+        self.fire_event_for_permission(dispvm_template=appvm)
 
         if preload_autostart:
-            await (
-                self.dest.fire_event_async("domain-preloaded-dispvm-autostart")
-            )
+            await appvm.fire_event_async("domain-preloaded-dispvm-autostart")
             return
 
         dispvm = await qubes.vm.dispvm.DispVM.from_appvm(
-            dispvm_template, preload=preload
+            appvm, preload=preload
         )
         # TODO: move this to extension (in race-free fashion, better than here)
         dispvm.tags.add("created-by-" + str(self.src))
@@ -1646,7 +1652,10 @@ async def vm_device_set_required(self, endpoint, untrusted_payload):
         self.app.save()
 
     @qubes.api.method(
-        "admin.vm.device.denied.List", no_payload=True, scope="local", read=True
+        "admin.vm.device.denied.List",
+        no_payload=True,
+        scope="local",
+        read=True,
     )
     async def vm_device_denied_list(self):
         """
@@ -1754,7 +1763,10 @@ async def vm_firewall_set(self, untrusted_payload):
         self.dest.firewall.save()
 
     @qubes.api.method(
-        "admin.vm.firewall.Reload", no_payload=True, scope="local", execute=True
+        "admin.vm.firewall.Reload",
+        no_payload=True,
+        scope="local",
+        execute=True,
     )
     async def vm_firewall_reload(self):
         self.enforce(not self.arg)
@@ -1955,7 +1967,9 @@ async def backup_info(self):
                 "Backup profile {} does not exist".format(self.arg)
             )
 
-        backup = await self._load_backup_profile(self.arg, skip_passphrase=True)
+        backup = await self._load_backup_profile(
+            self.arg, skip_passphrase=True
+        )
         return backup.get_backup_summary()
 
     def _send_stats_single(

qubes/tests/api_admin.py

@@ -3720,31 +3720,41 @@ def test_642_vm_create_disposable_not_allowed(self, storage_mock):
     def test_643_vm_create_disposable_preload(self, mock_storage):
         mock_storage.side_effect = self.dummy_coro
         self.vm.template_for_dispvms = True
-        self.vm.features["preload-dispvm-max"] = 1
+        self.vm.features["preload-dispvm-max"] = "1"
         self.app.default_dispvm = self.vm
         retval = self.call_mgmt_func(
-            b"admin.vm.CreateDisposable", b"dom0", arg="preload"
+            b"admin.vm.CreateDisposable", b"dom0", arg=b"preload"
         )
-        dispvm_preload = self.vm.features.get("dispvm-preload", "").split(" ")
-        self.assertIn(retval, dispvm_preload)
+        dispvm_preload = self.vm.get_feat_preload()
+        self.assertEqual(dispvm_preload, [retval])
+        dispvm = self.app.domains["".join(dispvm_preload)]
+        self.assertIn("created-by-dom0", dispvm.tags)
+        self.assertIn("disp-created-by-dom0", dispvm.tags)
         mock_storage.assert_called_once_with()
         self.assertTrue(self.app.save.called)
 
     @unittest.mock.patch("qubes.storage.Storage.create")
     def test_643_vm_create_disposable_preload_autostart(self, mock_storage):
         mock_storage.side_effect = self.dummy_coro
         self.vm.template_for_dispvms = True
-        self.vm.features["preload-dispvm-max"] = 1
+        self.app.domains[self.vm].fire_event = self.emitter.fire_event
+        self.vm.features["preload-dispvm-max"] = "1"
         self.app.default_dispvm = self.vm
+        # TODO: how to mock start of a qube that we don't have its object?
+        # Looked as test_220_start(), but 'self.vm.start = corountine_mock'
+        # won't help if we don't have the dispvm object.
         retval = self.call_mgmt_func(
-            b"admin.vm.CreateDisposable", b"dom0", arg="preload-autostart"
+            b"admin.vm.CreateDisposable", b"dom0", arg=b"preload-autostart"
         )
-        # TODO: doesn't return any value, so how to check if it was preloaded?
-        #dispvm_preload = self.vm.features.get("preload-dispvm", "").split(" ")
+        self.assertEventFired(self.emitter, "domain-preloaded-dispvm-autostart")
+        dispvm_preload = self.vm.get_feat_preload()
+        self.assertEqual(len(dispvm_preload), 1)
         self.assertIsNone(retval)
         mock_storage.assert_called_once_with()
         self.assertTrue(self.app.save.called)
 
+    ## TODO: test return of the preloaded dispvm
+
     def test_650_vm_device_set_mode_required(self):
         assignment = DeviceAssignment(
             VirtualDevice(Port(self.vm, "1234", "testclass"), device_id="bee"),

qubes/tests/integ/dispvm.py

@@ -37,7 +37,6 @@
 
 
 class TC_04_DispVM(qubes.tests.SystemTestCase):
-
     def setUp(self):
         super(TC_04_DispVM, self).setUp()
         self.init_default_template()
@@ -151,7 +150,9 @@ def test_010_failed_start(self):
             )
         )
         timeout = 120
-        self.loop.run_until_complete(asyncio.wait_for(p.communicate(), timeout))
+        self.loop.run_until_complete(
+            asyncio.wait_for(p.communicate(), timeout)
+        )
         self.assertEqual(p.returncode, 126)
         self.assertEqual(self.startup_counter, 1)
 
@@ -178,13 +179,14 @@ def test_011_failed_start_timeout(self):
             )
         )
         timeout = 120
-        self.loop.run_until_complete(asyncio.wait_for(p.communicate(), timeout))
+        self.loop.run_until_complete(
+            asyncio.wait_for(p.communicate(), timeout)
+        )
         self.assertEqual(p.returncode, 126)
         self.assertEqual(self.startup_counter, 1)
 
 
 class TC_20_DispVMMixin(object):
-
     def setUp(self):
         super(TC_20_DispVMMixin, self).setUp()
         if "whonix-g" in self.template:
@@ -206,8 +208,7 @@ def tearDown(self):
         self.app.default_dispvm = None
         super(TC_20_DispVMMixin, self).tearDown()
 
-    # TODO: Test if run_service() marks the prelaoded DispVM as used.
-    def test_010_simple_dvm_run(self):
+    def test_010_dvm_run_simple(self):
         dispvm = self.loop.run_until_complete(
             qubes.vm.dispvm.DispVM.from_appvm(self.disp_base)
         )
@@ -222,6 +223,45 @@ def test_010_simple_dvm_run(self):
         finally:
             self.loop.run_until_complete(dispvm.cleanup())
 
+    def test_011_dvm_run_preload(self):
+        dispvm = self.loop.run_until_complete(
+            qubes.vm.dispvm.DispVM.from_appvm(self.disp_base, preload=True)
+        )
+        try:
+            self.assertEqual(dispvm.get_feat_preload(), [dispvm.name])
+            self.assertTrue(dispvm.features.get("internal", False))
+            self.assertTrue(dispvm.is_paused)
+            self.loop.run_until_complete(
+                dispvm.run_service_for_stdio(
+                    "qubes.VMShell", input=b"echo test"
+                )
+            )
+            self.assertFalse(self.disp_base.get_feat_preload())
+            self.assertFalse(dispvm.features.get("internal", False))
+            self.assertFalse(dispvm.is_paused)
+        finally:
+            self.loop.run_until_complete(dispvm.cleanup())
+
+    def test_011_dvm_run_preload_nogui(self):
+        self.disp_base.features["gui"] = False
+        dispvm = self.loop.run_until_complete(
+            qubes.vm.dispvm.DispVM.from_appvm(self.disp_base, preload=True)
+        )
+        try:
+            self.assertEqual(dispvm.get_feat_preload(), [dispvm.name])
+            self.assertTrue(dispvm.features.get("internal", False))
+            self.assertTrue(dispvm.is_paused)
+            self.loop.run_until_complete(
+                dispvm.run_service(
+                    "qubes.VMShell", input=b"echo test", wait=False
+                )
+            )
+            self.assertFalse(self.disp_base.get_feat_preload())
+            self.assertFalse(dispvm.features.get("internal", False))
+            self.assertFalse(dispvm.is_paused)
+        finally:
+            self.loop.run_until_complete(dispvm.cleanup())
+
     @unittest.skipUnless(
         spawn.find_executable("xdotool"), "xdotool not installed"
     )

qubes/tests/run.py

@@ -1,3 +1,4 @@
+#!/usr/bin/env python3
 #
 # The Qubes OS Project, https://www.qubes-os.org/
 #
@@ -305,8 +306,8 @@ def test_6_unexpected_success(self):
 
 parser = argparse.ArgumentParser(
     epilog="""When running only specific tests, write their names like in log,
-        in format: MODULE+"/"+CLASS+"/"+FUNCTION. MODULE should omit initial
-        "qubes.tests.". Example: basic/TC_00_Basic/test_000_create"""
+        in format: MODULE+"/"+CLASS+"/"+FUNCTION.
+        Example: qubes.tests.basic/TC_00_Basic/test_000_create"""
 )
 
 parser.add_argument(