q-dev: add tests

Author: piotrbartman

qubes/api/admin.py

@@ -1281,7 +1281,7 @@ async def vm_device_attached(self, endpoint):
                 raise qubes.exc.QubesException("qubesd shutdown in progress")
             raise
         if self.arg:
-            select_backend, select_ident = self.arg.split('+', 1)  # TODO
+            select_backend, select_ident = self.arg.split('+', 1)
             device_assignments = [dev for dev in device_assignments
                                   if (str(dev.backend_domain), dev.port_id)
                                   == (select_backend, select_ident)]
@@ -1421,7 +1421,7 @@ async def vm_device_set_required(self, endpoint, untrusted_payload):
 
         dev = VirtualDevice.from_qarg(self.arg, devclass, self.app.domains)
 
-        self.fire_event_for_permission(device=dev, assignment=assignment)
+        self.fire_event_for_permission(device=dev, mode=assignment)
 
         await self.dest.devices[devclass].update_required(dev, assignment)
         self.app.save()

qubes/devices.py

@@ -238,6 +238,9 @@ async def assign(self, assignment: DeviceAssignment):
                 f'{self._bus} device {device!s} '
                 f'already assigned to {self._vm!s}')
 
+        if not assignment.attach_automatically:
+            raise ValueError('Only auto-attachable devices can be assigned.')
+
         self._set.add(assignment)
 
         await self._vm.fire_event_async(
@@ -281,8 +284,10 @@ async def update_required(self, device: VirtualDevice, required: bool):
         if assignment.required == required:
             return
 
-        assignments[0] = assignment.clone(
+        new_assignment = assignment.clone(
             mode='required' if required else 'auto-attach')
+        self._set.discard(assignment)
+        self._set.add(new_assignment)
         await self._vm.fire_event_async(
             'device-assignment-changed:' + self._bus, device=device)
 

qubes/ext/admin.py

@@ -168,7 +168,7 @@ def on_tag_add(self, vm, event, tag, **kwargs):
 
     @qubes.ext.handler(*(f'admin-permission:admin.vm.device.{ep.name}.Attach'
         for ep in importlib.metadata.entry_points(group='qubes.devices')))
-    def on_usb_device_attacg(
+    def on_device_attach(
             self, vm, event, dest, arg, device, mode, options, **kwargs
     ):
         # ignore auto-attachment