q-dev: assignment

Author: piotrbartman

qubes/device_protocol.py

@@ -792,6 +792,13 @@ def __init__(self, backend_domain, ident, *, devclass, **kwargs):
         super().__init__(port, **kwargs)
 
 
+class AssignmentMode(Enum):
+    MANUAL = "manual"
+    ASK = "ask-to-attach"
+    AUTO = "auto-attach"
+    REQUIRED = "required"
+
+
 class DeviceAssignment(Port):
     """ Maps a device to a frontend_domain.
 
@@ -812,29 +819,20 @@ class DeviceAssignment(Port):
                                and required to start domain.
     """
 
-    class AssignmentType(Enum):
-        MANUAL = 0
-        ASK = 1
-        AUTO = 2
-        REQUIRED = 3
-
     def __init__(
             self,
             port: Port,
+            device_id=None,
             frontend_domain=None,
             options=None,
-            required=False,
-            attach_automatically=False
+            mode: Union[str, AssignmentMode] = "manual",
     ):
         super().__init__(port.backend_domain, port.ident, port.devclass)
         self.__options = options or {}
-        if required:
-            assert attach_automatically
-            self.type = DeviceAssignment.AssignmentType.REQUIRED
-        elif attach_automatically:
-            self.type = DeviceAssignment.AssignmentType.AUTO
+        if isinstance(mode, AssignmentMode):
+            self.mode = mode
         else:
-            self.type = DeviceAssignment.AssignmentType.MANUAL
+            self.mode = AssignmentMode(mode)
         self.frontend_domain = frontend_domain
 
     def clone(self, **kwargs):
@@ -885,26 +883,26 @@ def required(self) -> bool:
         Is the presence of this device required for the domain to start? If yes,
         it will be attached automatically.
         """
-        return self.type == DeviceAssignment.AssignmentType.REQUIRED
+        return self.mode == AssignmentMode.REQUIRED
 
     @required.setter
     def required(self, required: bool):
-        self.type = DeviceAssignment.AssignmentType.REQUIRED
+        self.mode = AssignmentMode.REQUIRED
 
     @property
     def attach_automatically(self) -> bool:
         """
         Should this device automatically connect to the frontend domain when
         available and not connected to other qubes?
         """
-        return self.type in (
-            DeviceAssignment.AssignmentType.AUTO,
-            DeviceAssignment.AssignmentType.REQUIRED
+        return self.mode in (
+            AssignmentMode.AUTO,
+            AssignmentMode.REQUIRED
         )
 
     @attach_automatically.setter
     def attach_automatically(self, attach_automatically: bool):
-        self.type = DeviceAssignment.AssignmentType.AUTO
+        self.mode = AssignmentMode.AUTO
 
     @property
     def options(self) -> Dict[str, Any]:
@@ -923,9 +921,7 @@ def serialize(self) -> bytes:
         properties = b' '.join(
             self.pack_property(key, value)
             for key, value in (
-                ('required', 'yes' if self.required else 'no'),
-                ('attach_automatically',
-                 'yes' if self.attach_automatically else 'no'),
+                ('mode', self.mode.value),
                 ('ident', self.ident),
                 ('devclass', self.devclass)))
 
@@ -973,8 +969,4 @@ def _deserialize(
         del properties['ident']
         del properties['devclass']
 
-        properties['attach_automatically'] = qbool(
-            properties.get('attach_automatically', 'no'))
-        properties['required'] = qbool(properties.get('required', 'no'))
-
         return cls(expected_port, **properties)

qubes/devices.py

@@ -229,8 +229,8 @@ async def assign(self, assignment: DeviceAssignment):
         device = assignment.device
         if device in self.get_assigned_devices():
             raise DeviceAlreadyAssigned(
-                'device {!s} of class {} already assigned to {!s}'.format(
-                    device, self._bus, self._vm))
+                '{} device {!s} already assigned to {!s}'.format(
+                    self._bus, device, self._vm))
 
         self._set.add(assignment)
 
@@ -355,8 +355,7 @@ def get_attached_devices(self) -> Iterable[DeviceAssignment]:
                     ),
                     frontend_domain=self._vm,
                     options=options,
-                    attach_automatically=False,
-                    required=False,
+                    mode='manual',
                 )
 
     def get_assigned_devices(

qubes/tests/api_admin.py

@@ -1787,7 +1787,7 @@ def test_462_vm_device_available_invalid(self):
     def test_470_vm_device_list_assigned(self):
         assignment = qubes.device_protocol.DeviceAssignment(
             qubes.device_protocol.Port(self.vm, '1234', 'test'),
-            attach_automatically=True, required=True)
+            mode='required')
         self.loop.run_until_complete(
             self.vm.devices['testclass'].assign(assignment))
         value = self.call_mgmt_func(b'admin.vm.device.testclass.Assigned',
@@ -1800,12 +1800,11 @@ def test_470_vm_device_list_assigned(self):
     def test_471_vm_device_list_assigned_options(self):
         assignment = qubes.device_protocol.DeviceAssignment(
             qubes.device_protocol.Port(self.vm, '1234', 'test'),
-            attach_automatically=True, required=True, options={'opt1': 'value'})
+            mode='required', options={'opt1': 'value'})
         self.loop.run_until_complete(
             self.vm.devices['testclass'].assign(assignment))
         assignment = qubes.device_protocol.DeviceAssignment(
-            self.vm, '4321', 'test',
-            attach_automatically=True, required=True)
+            self.vm, '4321', 'test', mode='required')
         self.loop.run_until_complete(
             self.vm.devices['testclass'].assign(assignment))
         value = self.call_mgmt_func(b'admin.vm.device.testclass.Assigned',
@@ -1838,12 +1837,12 @@ def test_472_vm_device_list_attached(self):
     def test_473_vm_device_list_assigned_specific(self):
         assignment = qubes.device_protocol.DeviceAssignment(
             qubes.device_protocol.Port(self.vm, '1234', 'test'),
-            attach_automatically=True, required=True)
+            mode='required')
         self.loop.run_until_complete(
             self.vm.devices['testclass'].assign(assignment))
         assignment = qubes.device_protocol.DeviceAssignment(
             qubes.device_protocol.Port(self.vm, '4321', 'test'),
-            attach_automatically=True, required=True)
+            mode='required')
         self.loop.run_until_complete(
             self.vm.devices['testclass'].assign(assignment))
         value = self.call_mgmt_func(b'admin.vm.device.testclass.Assigned',
@@ -2027,8 +2026,7 @@ def test_488_vm_device_assign_options(self):
     def test_490_vm_device_unassign_from_running(self):
         assignment = qubes.device_protocol.DeviceAssignment(
             qubes.device_protocol.Port(self.vm, '1234', 'test'),
-            attach_automatically=True, required=False,
-            options={'opt1': 'value'})
+            mode='auto-attach', options={'opt1': 'value'})
         self.loop.run_until_complete(
             self.vm.devices['testclass'].assign(assignment))
         mock_action = unittest.mock.Mock()
@@ -2048,8 +2046,7 @@ def test_490_vm_device_unassign_from_running(self):
     def test_491_vm_device_unassign_required_from_running(self):
         assignment = qubes.device_protocol.DeviceAssignment(
             qubes.device_protocol.Port(self.vm, '1234', 'test'),
-            attach_automatically=True, required=True,
-            options={'opt1': 'value'})
+            mode='required', options={'opt1': 'value'})
         self.loop.run_until_complete(
             self.vm.devices['testclass'].assign(assignment))
         mock_action = unittest.mock.Mock()
@@ -2070,8 +2067,7 @@ def test_491_vm_device_unassign_required_from_running(self):
     def test_492_vm_device_unassign_from_halted(self):
         assignment = qubes.device_protocol.DeviceAssignment(
             qubes.device_protocol.Port(self.vm, '1234', 'test'),
-            attach_automatically=True, required=False,
-            options={'opt1': 'value'})
+            mode='required', options={'opt1': 'value'})
         self.loop.run_until_complete(
             self.vm.devices['testclass'].assign(assignment))
         mock_action = unittest.mock.Mock()
@@ -2089,8 +2085,7 @@ def test_492_vm_device_unassign_from_halted(self):
     def test_493_vm_device_unassign_required_from_halted(self):
         assignment = qubes.device_protocol.DeviceAssignment(
             qubes.device_protocol.Port(self.vm, '1234', 'test'),
-            attach_automatically=True, required=True,
-            options={'opt1': 'value'})
+            mode='required', options={'opt1': 'value'})
         self.loop.run_until_complete(
             self.vm.devices['testclass'].assign(assignment))
         mock_action = unittest.mock.Mock()
@@ -2111,8 +2106,7 @@ def test_494_vm_device_unassign_attached(self):
                             self.device_list_single_attached_testclass)
         assignment = qubes.device_protocol.DeviceAssignment(
             qubes.device_protocol.Port(self.vm, '1234', 'test'),
-            attach_automatically=True, required=False,
-            options={'opt1': 'value'})
+            mode='auto-attach', options={'opt1': 'value'})
         self.loop.run_until_complete(
             self.vm.devices['testclass'].assign(assignment))
         mock_action = unittest.mock.Mock()
@@ -2199,7 +2193,7 @@ def test_502_vm_remove_attached(self, mock_rmtree, mock_remove):
         self.setup_for_clone()
         assignment = qubes.device_protocol.DeviceAssignment(
             qubes.device_protocol.Port(self.vm, '1234', 'test'),
-            attach_automatically=True, required=True)
+            mode='required')
         self.loop.run_until_complete(
             self.vm2.devices['testclass'].assign(assignment))
 
@@ -2919,8 +2913,7 @@ def test_642_vm_create_disposable_not_allowed(self, storage_mock):
     def test_650_vm_device_set_required_true(self):
         assignment = qubes.device_protocol.DeviceAssignment(
             qubes.device_protocol.Port(self.vm, '1234', 'test'),
-            attach_automatically=True, required=False,
-            options={'opt1': 'value'})
+            mode='auto-attach', options={'opt1': 'value'})
         self.loop.run_until_complete(
             self.vm.devices['testclass'].assign(assignment))
         mock_action = unittest.mock.Mock()
@@ -2950,8 +2943,7 @@ def test_650_vm_device_set_required_true(self):
     def test_651_vm_device_set_required_false(self):
         assignment = qubes.device_protocol.DeviceAssignment(
             qubes.device_protocol.Port(self.vm, '1234', 'test'),
-            attach_automatically=True, required=True,
-            options={'opt1': 'value'})
+            mode='required', options={'opt1': 'value'})
         self.loop.run_until_complete(
             self.vm.devices['testclass'].assign(assignment))
         mock_action = unittest.mock.Mock()
@@ -2981,8 +2973,7 @@ def test_651_vm_device_set_required_false(self):
     def test_652_vm_device_set_required_true_unchanged(self):
         assignment = qubes.device_protocol.DeviceAssignment(
             qubes.device_protocol.Port(self.vm, '1234', 'test'),
-            attach_automatically=True, required=True,
-            options={'opt1': 'value'})
+            mode='required', options={'opt1': 'value'})
         self.loop.run_until_complete(
             self.vm.devices['testclass'].assign(assignment))
         with unittest.mock.patch.object(qubes.vm.qubesvm.QubesVM,
@@ -3000,8 +2991,7 @@ def test_652_vm_device_set_required_true_unchanged(self):
     def test_653_vm_device_set_required_false_unchanged(self):
         assignment = qubes.device_protocol.DeviceAssignment(
             qubes.device_protocol.Port(self.vm, '1234', 'test'),
-            attach_automatically=True, required=False,
-            options={'opt1': 'value'})
+            mode='auto-attach', options={'opt1': 'value'})
         self.loop.run_until_complete(
             self.vm.devices['testclass'].assign(assignment))
         with unittest.mock.patch.object(qubes.vm.qubesvm.QubesVM,

qubes/tests/devices.py

@@ -90,11 +90,7 @@ def setUp(self):
         self.app.domains['vm'] = self.emitter
         self.device = self.emitter.device
         self.collection = self.emitter.devices['testclass']
-        self.assignment = DeviceAssignment(
-            self.device,
-            attach_automatically=True,
-            required=True,
-        )
+        self.assignment = DeviceAssignment(self.device, mode='required')
 
     def attach(self):
         self.emitter.running = True
@@ -339,8 +335,7 @@ def test_000_init(self):
 
     def test_001_missing(self):
         device = TestDevice(self.emitter.app.domains['vm'], 'testdev')
-        assignment = DeviceAssignment(
-            device, attach_automatically=True, required=True)
+        assignment = DeviceAssignment(device, mode='required')
         self.loop.run_until_complete(
             self.manager['testclass'].assign(assignment))
         self.assertEqual(
@@ -532,8 +527,7 @@ def test_011_serialize_required(self):
                 ident="1-1.1.1",
                 devclass="bus",
             ),
-            attach_automatically=True,
-            required=True,
+            mode='required',
         )
         actual = assignment.serialize()
         expected = (
@@ -603,8 +597,7 @@ def test_020_deserialize(self):
                 devclass="bus",
             ),
             frontend_domain=self.vm,
-            attach_automatically=True,
-            required=False,
+            mode='auto-attach',
             options={'read-only': 'yes'},
         )
 
@@ -642,8 +635,7 @@ def test_030_serialize_and_deserialize(self):
                 devclass="bus",
             ),
             frontend_domain=self.vm,
-            attach_automatically=True,
-            required=False,
+            mode='auto-attach',
             options={'read-only': 'yes'},
         )
         serialized = expected.serialize()

qubes/tests/integ/devices_pci.py

@@ -39,12 +39,10 @@ def setUp(self):
             pcidev = os.environ['QUBES_TEST_PCIDEV']
             self.dev = self.app.domains[0].devices['pci'][pcidev]
             self.assignment = DeviceAssignment(
-                self.dev, attach_automatically=True
+                self.dev, mode='auto-attach'
             )
             self.required_assignment = DeviceAssignment(
-                self.dev,
-                attach_automatically=True,
-                required=True,
+                self.dev, mode='required',
             )
             if isinstance(self.dev, qubes.device_protocol.UnknownDevice):
                 self.skipTest('Specified device {} does not exists'.format(pcidev))

qubes/tests/vm/qubesvm.py

@@ -1317,8 +1317,7 @@ def test_600_libvirt_xml_hvm_pcidev(self):
                 ident='00_00.0',
                 devclass="pci",
             ),
-            attach_automatically=True,
-            required=True,
+            mode='required',
         )
         vm.devices['pci']._set.add(
             assignment)
@@ -1405,7 +1404,7 @@ def test_600_libvirt_xml_hvm_pcidev_s0ix(self):
                 ident='00_00.0',
                 devclass="pci",
             ),
-            attach_automatically=True, required=True)
+            mode='required')
         vm.devices['pci']._set.add(
             assignment)
         libvirt_xml = vm.create_config_file()
@@ -1493,7 +1492,7 @@ def test_600_libvirt_xml_hvm_cdrom_boot(self):
                 devclass="block",
             ),
             options={'devtype': 'cdrom', 'read-only': 'yes'},
-            attach_automatically=True, required=True)
+            mode='required')
         self.loop.run_until_complete(vm.devices['block'].assign(dev))
         libvirt_xml = vm.create_config_file()
         self.assertXMLEqual(lxml.etree.XML(libvirt_xml),
@@ -1602,7 +1601,7 @@ def test_600_libvirt_xml_hvm_cdrom_dom0_kernel_boot(self):
                     devclass="block",
                 ),
                 options={'devtype': 'cdrom', 'read-only': 'yes'},
-                 attach_automatically=True, required=True)
+                mode='required')
             self.loop.run_until_complete(vm.devices['block'].assign(dev))
             libvirt_xml = vm.create_config_file()
         self.assertXMLEqual(lxml.etree.XML(libvirt_xml),