q-dev: virtual device

qubesadmin/backup/restore.py

@@ -50,7 +50,7 @@
 from qubesadmin.backup import BackupVM
 from qubesadmin.backup.core2 import Core2Qubes
 from qubesadmin.backup.core3 import Core3Qubes
-from qubesadmin.device_protocol import DeviceAssignment, Port, Device
+from qubesadmin.device_protocol import DeviceAssignment, Port, VirtualDevice
 from qubesadmin.exc import QubesException
 from qubesadmin.utils import size_to_human
 
@@ -2094,7 +2094,7 @@ def _restore_vms_metadata(self, restore_info):
                     else:
                         required = False
                     assignment = DeviceAssignment(
-                        Device(Port(
+                        VirtualDevice(Port(
                             backend_domain=self.app.domains[backend_domain],
                             port_id=port_id,
                             devclass=bus,

qubesadmin/device_protocol.py

@@ -154,7 +154,7 @@ def pack_property(cls, key: str, value: Optional[str]):
 
     @staticmethod
     def parse_basic_device_properties(
-            expected_device: 'Device', properties: Dict[str, Any]):
+            expected_device: 'VirtualDevice', properties: Dict[str, Any]):
         """
         Validates properties against an expected port configuration.
 
@@ -338,7 +338,7 @@ def devclass(self) -> str:
         return "peripheral"
 
 
-class Device:
+class VirtualDevice:
     def __init__(
             self,
             port: Optional[Port] = None,
@@ -396,7 +396,7 @@ def __hash__(self):
         return hash((self.port, self.device_id))
 
     def __eq__(self, other):
-        if isinstance(other, (Device, DeviceAssignment)):
+        if isinstance(other, (VirtualDevice, DeviceAssignment)):
             result = (
                     self.port == other.port
                     and self.device_id == other.device_id
@@ -418,7 +418,7 @@ def __lt__(self, other):
         3. *:<devid>
         4. *:*
         """
-        if isinstance(other, (Device, DeviceAssignment)):
+        if isinstance(other, (VirtualDevice, DeviceAssignment)):
             if self.port == '*' and other.port != '*':
                 return True
             if self.port != '*' and other.port == '*':
@@ -429,7 +429,7 @@ def __lt__(self, other):
                     reprs[obj].append(obj.device_id)
             return reprs[self] < reprs[other]
         elif isinstance(other, Port):
-            _other = Device(other, '*')
+            _other = VirtualDevice(other, '*')
             return self < _other
         else:
             raise TypeError(
@@ -450,7 +450,7 @@ def from_qarg(
             domains,
             blind=False,
             backend=None,
-    ) -> 'Device':
+    ) -> 'VirtualDevice':
         if backend is None:
             if blind:
                 get_domain = domains.get_blind
@@ -464,7 +464,7 @@ def from_qarg(
     def from_str(
             cls, representation: str, devclass: Optional[str], domains,
             blind=False, backend=None
-    ) -> 'Device':
+    ) -> 'VirtualDevice':
         if backend is None:
             if blind:
                 get_domain = domains.get_blind
@@ -482,7 +482,7 @@ def _parse(
             get_domain: Callable,
             backend,
             sep: str
-    ) -> 'Device':
+    ) -> 'VirtualDevice':
         if backend is None:
             backend_name, identity = representation.split(sep, 1)
             backend = get_domain(backend_name)
@@ -693,7 +693,7 @@ def _load_classes(bus: str):
         return result
 
 
-class DeviceInfo(Device):
+class DeviceInfo(VirtualDevice):
     """ Holds all information about a device """
 
     def __init__(
@@ -833,7 +833,7 @@ def interfaces(self) -> List[DeviceInterface]:
         return self._interfaces
 
     @property
-    def parent_device(self) -> Optional[Device]:
+    def parent_device(self) -> Optional[VirtualDevice]:
         """
         The parent device, if any.
 
@@ -843,7 +843,7 @@ def parent_device(self) -> Optional[Device]:
         return self._parent
 
     @property
-    def subdevices(self) -> List[Device]:
+    def subdevices(self) -> List[VirtualDevice]:
         """
         The list of children devices if any.
 
@@ -864,7 +864,7 @@ def serialize(self) -> bytes:
         """
         Serialize an object to be transmitted via Qubes API.
         """
-        properties = Device.serialize(self)
+        properties = VirtualDevice.serialize(self)
         # 'attachment', 'interfaces', 'data', 'parent_device'
         # are not string, so they need special treatment
         default = DeviceInfo(self.port)
@@ -904,7 +904,7 @@ def deserialize(
         Recovers a serialized object, see: :py:meth:`serialize`.
         """
         head, _, rest = serialization.partition(b' ')
-        device = Device.from_str(
+        device = VirtualDevice.from_str(
             head.decode('ascii', errors='ignore'), expected_devclass,
             domains=None, backend=expected_backend_domain)
 
@@ -920,7 +920,7 @@ def deserialize(
     def _deserialize(
             cls,
             untrusted_serialization: bytes,
-            expected_device: Device
+            expected_device: VirtualDevice
     ) -> 'DeviceInfo':
         """
         Actually deserializes the object.
@@ -1020,13 +1020,13 @@ class DeviceAssignment:
 
     def __init__(
             self,
-            device: Device,
+            device: VirtualDevice,
             frontend_domain=None,
             options=None,
             mode: Union[str, AssignmentMode] = "manual",
     ):
         if isinstance(device, DeviceInfo):
-            device = Device(device.port, device.device_id)
+            device = VirtualDevice(device.port, device.device_id)
         self._device_ident = device
         self.__options = options or {}
         if isinstance(mode, AssignmentMode):
@@ -1040,7 +1040,7 @@ def clone(self, **kwargs):
         Clone object and substitute attributes with explicitly given.
         """
         kwargs["device"] = kwargs.get(
-            "device", Device(
+            "device", VirtualDevice(
                 Port(self.backend_domain, self.port_id, self.devclass),
                 self.device_id
             ))
@@ -1062,7 +1062,7 @@ def __hash__(self):
         return hash(self._device_ident)
 
     def __eq__(self, other):
-        if isinstance(other, (Device, DeviceAssignment)):
+        if isinstance(other, (VirtualDevice, DeviceAssignment)):
             result = (
                     self.port == other.port
                     and self.device_id == other.device_id
@@ -1073,7 +1073,7 @@ def __eq__(self, other):
     def __lt__(self, other):
         if isinstance(other, DeviceAssignment):
             return self._device_ident < other._device_ident
-        if isinstance(other, Device):
+        if isinstance(other, VirtualDevice):
             return self._device_ident < other
         raise TypeError(
             f"Comparing instances of {type(self)} and '{type(other)}' "
@@ -1182,7 +1182,7 @@ def serialize(self) -> bytes:
     def deserialize(
             cls,
             serialization: bytes,
-            expected_device: Device,
+            expected_device: VirtualDevice,
     ) -> 'DeviceAssignment':
         """
         Recovers a serialized object, see: :py:meth:`serialize`.
@@ -1197,7 +1197,7 @@ def deserialize(
     def _deserialize(
             cls,
             untrusted_serialization: bytes,
-            expected_device: Device,
+            expected_device: VirtualDevice,
     ) -> 'DeviceAssignment':
         """
         Actually deserializes the object.

qubesadmin/devices.py

@@ -36,7 +36,7 @@ class is implemented by an extension.
 
 import qubesadmin.exc
 from qubesadmin.device_protocol import (Port, DeviceInfo, UnknownDevice,
-                                        DeviceAssignment, Device)
+                                        DeviceAssignment, VirtualDevice)
 
 
 class DeviceCollection:
@@ -153,7 +153,7 @@ def get_attached_devices(self) -> Iterable[DeviceAssignment]:
             None, 'admin.vm.device.{}.Attached'.format(self._class)).decode()
         for assignment_str in assignments_str.splitlines():
             head, _, untrusted_rest = assignment_str.partition(' ')
-            device = Device.from_qarg(
+            device = VirtualDevice.from_qarg(
                 head, self._class, self._vm.app.domains, blind=True)
 
             yield DeviceAssignment.deserialize(
@@ -171,7 +171,7 @@ def get_assigned_devices(
             None, 'admin.vm.device.{}.Assigned'.format(self._class)).decode()
         for assignment_str in assignments_str.splitlines():
             head, _, untrusted_rest = assignment_str.partition(' ')
-            device = Device.from_qarg(
+            device = VirtualDevice.from_qarg(
                 head, self._class, self._vm.app.domains, blind=True)
 
             assignment = DeviceAssignment.deserialize(
@@ -196,7 +196,7 @@ def update_assignment(self, device: Port, required: Optional[bool]):
         """
         Update assignment of already attached device.
 
-        :param Device device: device for which change required flag
+        :param VirtualDevice device: device for which change required flag
         :param bool required: new assignment:
                               `None` -> unassign device from qube
                               `False` -> device will be auto-attached to qube

qubesadmin/events/__init__.py

@@ -26,7 +26,7 @@
 
 import qubesadmin.config
 import qubesadmin.exc
-from qubesadmin.device_protocol import Device, Port
+from qubesadmin.device_protocol import VirtualDevice, Port
 
 
 class EventsDispatcher(object):
@@ -234,7 +234,7 @@ def handle(self, subject, event, **kwargs):
             try:
                 if 'device' in kwargs:
                     devclass = event.split(':', 1)[1]
-                    device = Device.from_str(
+                    device = VirtualDevice.from_str(
                         kwargs['device'],
                         devclass,
                         self.app.domains,

qubesadmin/tests/devices.py

@@ -21,7 +21,7 @@
 import qubesadmin.tests
 import qubesadmin.device_protocol
 
-from qubesadmin.device_protocol import (DeviceAssignment, Port, Device,
+from qubesadmin.device_protocol import (DeviceAssignment, Port, VirtualDevice,
                                         DeviceInfo, UnknownDevice)
 
 
@@ -119,7 +119,7 @@ def test_020_attach(self):
              b"frontend_domain='test-vm'")] = \
             b'0\0'
         assign = DeviceAssignment(
-            Device(Port(
+            VirtualDevice(Port(
                 self.app.domains['test-vm2'], 'dev1', devclass='test',)))
         self.vm.devices['test'].attach(assign)
         self.assertAllCalled()
@@ -132,7 +132,7 @@ def test_021_attach_options(self):
              b"frontend_domain='test-vm' _ro='True' "
              b"_something='value'")] = b'0\0'
         assign = DeviceAssignment(
-            Device(Port(
+            VirtualDevice(Port(
                 self.app.domains['test-vm2'], 'dev1', devclass='test')))
         assign.options['ro'] = True
         assign.options['something'] = 'value'
@@ -146,7 +146,7 @@ def test_022_attach_required(self):
              b"backend_domain='test-vm2' mode='required' "
              b"frontend_domain='test-vm'")] = b'0\0'
         assign = DeviceAssignment(
-            Device(Port(
+            VirtualDevice(Port(
                 self.app.domains['test-vm2'], 'dev1', devclass='test')),
             mode='required')
         self.vm.devices['test'].attach(assign)
@@ -159,7 +159,7 @@ def test_023_attach_required_options(self):
              b"backend_domain='test-vm2' mode='required' "
              b"frontend_domain='test-vm' _ro='True'")] = b'0\0'
         assign = DeviceAssignment(
-            Device(Port(
+            VirtualDevice(Port(
                 self.app.domains['test-vm2'], 'dev1', devclass='test')),
             mode='required')
         assign.options['ro'] = True
@@ -171,7 +171,7 @@ def test_030_detach(self):
             ('test-vm', 'admin.vm.device.test.Detach', 'test-vm2+dev1:*',
              None)] = b'0\0'
         assign = DeviceAssignment(
-            Device(Port(
+            VirtualDevice(Port(
                 self.app.domains['test-vm2'], 'dev1', devclass='test')))
         self.vm.devices['test'].detach(assign)
         self.assertAllCalled()
@@ -293,7 +293,7 @@ def test_070_update_assignment(self):
             ('test-vm', 'admin.vm.device.test.Set.assignment',
              'test-vm2+dev1:*', b'True')] = b'0\0'
         dev = DeviceAssignment(
-            Device(
+            VirtualDevice(
                 Port(
                     self.app.domains['test-vm2'],
                     devclass='test',
@@ -307,7 +307,7 @@ def test_071_update_assignment_false(self):
             ('test-vm', 'admin.vm.device.test.Set.assignment',
              'test-vm2+dev1:*', b'False')] = b'0\0'
         dev = DeviceAssignment(
-            Device(
+            VirtualDevice(
                 Port(
                     self.app.domains['test-vm2'],
                     devclass='test',
@@ -321,7 +321,7 @@ def test_072_update_assignment_none(self):
             ('test-vm', 'admin.vm.device.test.Set.assignment',
              'test-vm2+dev1:*', b'None')] = b'0\0'
         dev = DeviceAssignment(
-            Device(
+            VirtualDevice(
                 Port(
                     self.app.domains['test-vm2'],
                     devclass='test',

qubesadmin/tools/qvm_device.py

@@ -263,9 +263,8 @@ def info_device(args):
     vm = args.domains[0]
     if args.device:
         device = args.device
-        assignment = DeviceAssignment(device)
-        print("description:", assignment.device.description)
-        print("data:", assignment.device.data)
+        print("description:", device.description)
+        print("data:", device.data)
     else:
         for device_assignment in (
                 vm.devices[args.devclass].get_dedicated_devices()):

qubesadmin/tools/qvm_start.py

@@ -158,7 +158,7 @@ def get_drive_assignment(app, drive_str):
         'read-only': devtype == 'cdrom'
     }
     assignment = qubesadmin.device_protocol.DeviceAssignment(
-        qubesadmin.device_protocol.Device(qubesadmin.device_protocol.Port(
+        qubesadmin.device_protocol.VirtualDevice(qubesadmin.device_protocol.Port(
             backend_domain=backend_domain,
             port_id=port_id,
             devclass='block',