q-dev: devices improvements

allow to update assignment of halted vm (no reason to forbid it in new API) do not double-check number of matched devices during attaching assignment
QubesOS · Oct 15, 2024 · 1109708 · 1109708
1 parent e6b70ef
commit 1109708
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 56 deletions.
diff --git a/qubes/devices.py b/qubes/devices.py
@@ -206,12 +206,13 @@ async def attach(self, assignment: DeviceAssignment):
                 self._vm,"VM not running, cannot attach device,"
                 " do you mean `assign`?")
 
-        if len(assignment.devices) != 1:
+        try:
+            device = assignment.device
+        except ProtocolError:
+            # assignment matches no or top many devices
             raise ProtocolError(
                 f'Cannot attach ambiguous {assignment.devclass} device.')
 
-        device = assignment.device
-
         if isinstance(device, UnknownDevice):
             raise ProtocolError(f"{device.devclass} device not recognized "
                                 f"in {device.port_id} port.")
@@ -271,16 +272,14 @@ async def update_assignment(
             self, device: VirtualDevice, mode: AssignmentMode
     ):
         """
-        Update `required` flag of an already attached device.
+        Update assignment mode of an already assigned device.
 
         :param VirtualDevice device: device for which change required flag
         :param AssignmentMode mode: new assignment mode
         """
-        if self._vm.is_halted():
-            raise qubes.exc.QubesVMNotStartedError(
-                self._vm,
-                'VM must be running to modify device assignment'
-            )
+        if mode == AssignmentMode.MANUAL:
+            raise qubes.exc.QubesValueError(
+                "Cannot change assignment mode to 'manual'")
         assignments = [a for a in self.get_assigned_devices()
                        if a.virtual_device == device]
         if not assignments:
@@ -289,11 +288,11 @@ async def update_assignment(
         assert len(assignments) == 1
         assignment = assignments[0]
 
-        # be careful to use already present assignment, not the provided one
-        # - to not change options as a side effect
         if assignment.mode == mode:
             return
 
+        # be careful to use already present assignment, not the provided one
+        # - to not change options as a side effect
         new_assignment = assignment.clone(mode=mode)
 
         await self._vm.fire_event_async(

diff --git a/qubes/tests/devices.py b/qubes/tests/devices.py
@@ -208,7 +208,6 @@ def test_018_list_available(self):
     def test_020_update_mode_to_auto(self):
         self.assertEqual(set([]), set(self.collection.get_assigned_devices()))
         self.loop.run_until_complete(self.collection.assign(self.assignment))
-        self.attach()
         self.assertEqual(
             {self.assignment},
             set(self.collection.get_assigned_devices(required_only=True)))
@@ -221,13 +220,10 @@ def test_020_update_mode_to_auto(self):
             set(self.collection.get_assigned_devices(required_only=True)))
         self.assertEqual(
             {self.assignment}, set(self.collection.get_assigned_devices()))
-        self.assertEqual(
-            {self.assignment}, set(self.collection.get_attached_devices()))
 
     def test_021_update_mode_to_ask(self):
         self.assertEqual(set([]), set(self.collection.get_assigned_devices()))
         self.loop.run_until_complete(self.collection.assign(self.assignment))
-        self.attach()
         self.assertEqual(
             {self.assignment},
             set(self.collection.get_assigned_devices(required_only=True)))
@@ -240,59 +236,29 @@ def test_021_update_mode_to_ask(self):
             set(self.collection.get_assigned_devices(required_only=True)))
         self.assertEqual(
             {self.assignment}, set(self.collection.get_assigned_devices()))
-        self.assertEqual(
-            {self.assignment}, set(self.collection.get_attached_devices()))
 
     def test_022_update_mode_to_required(self):
         self.assignment = self.assignment.clone(mode='auto-attach')
-        self.attach()
         self.assertEqual(set(), set(self.collection.get_assigned_devices()))
         self.loop.run_until_complete(self.collection.assign(self.assignment))
+
         self.assertEqual(
             set(),
             set(self.collection.get_assigned_devices(required_only=True)))
-        self.assertEqual({self.assignment},
-                         set(self.collection.get_attached_devices()))
-        self.assertEqual({self.assignment}
-                         , set(self.collection.get_assigned_devices()))
-        self.assertEqual({self.assignment},
-                         set(self.collection.get_attached_devices()))
+        self.assertEqual(
+            {self.assignment},
+            set(self.collection.get_assigned_devices()))
+
         self.loop.run_until_complete(
             self.collection.update_assignment(
                 self.device, AssignmentMode.REQUIRED))
-        self.assertEqual({self.assignment},
-                         set(self.collection.get_assigned_devices(
-                             required_only=True)))
-        self.assertEqual(
-            {self.assignment}, set(self.collection.get_attached_devices()))
 
-    def test_023_update_mode_reject_not_running(self):
-        self.assertEqual(set([]), set(self.collection.get_assigned_devices()))
-        self.loop.run_until_complete(self.collection.assign(self.assignment))
-        self.assertEqual({self.assignment},
-                         set(self.collection.get_assigned_devices()))
-        self.assertEqual(set(), set(self.collection.get_attached_devices()))
-        with self.assertRaises(qubes.exc.QubesVMNotStartedError):
-            self.loop.run_until_complete(
-                self.collection.update_assignment(
-                    self.device, AssignmentMode.ASK))
-
-    def test_024_update_required_reject_not_attached(self):
-        self.assertEqual(set(), set(self.collection.get_assigned_devices()))
-        self.assertEqual(set(), set(self.collection.get_attached_devices()))
-        self.emitter.running = True
-        with self.assertRaises(qubes.exc.QubesValueError):
-            self.loop.run_until_complete(
-                self.collection.update_assignment(
-                    self.device, AssignmentMode.REQUIRED))
-        with self.assertRaises(qubes.exc.QubesValueError):
-            self.loop.run_until_complete(
-                self.collection.update_assignment(
-                    self.device, AssignmentMode.ASK))
-        with self.assertRaises(qubes.exc.QubesValueError):
-            self.loop.run_until_complete(
-                self.collection.update_assignment(
-                    self.device, AssignmentMode.AUTO))
+        self.assertEqual(
+            {self.assignment},
+            set(self.collection.get_assigned_devices(required_only=True)))
+        self.assertEqual(
+            {self.assignment},
+            set(self.collection.get_assigned_devices()))
 
     def test_030_assign(self):
         self.emitter.running = True