q-dev: device protocol minor fixes

qubesadmin/app.py

@@ -39,7 +39,7 @@
 import qubesadmin.utils
 import qubesadmin.vm
 import qubesadmin.config
-import qubesadmin.devices
+import qubesadmin.device_protocol
 
 try:
     import qubesdb

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.devices import DeviceAssignment
+from qubesadmin.device_protocol import DeviceAssignment
 from qubesadmin.exc import QubesException
 from qubesadmin.utils import size_to_human
 

qubesadmin/devices.py

@@ -158,9 +158,7 @@ def get_attached_devices(self) -> Iterable[DeviceAssignment]:
 
             yield DeviceAssignment.deserialize(
                 untrusted_rest.encode('ascii'),
-                expected_backend_domain=backend_domain,
-                expected_ident=ident,
-                expected_devclass=None
+                expected_device=Device(backend_domain, ident)
             )
 
     def get_assigned_devices(
@@ -180,9 +178,7 @@ def get_assigned_devices(
 
             assignment = DeviceAssignment.deserialize(
                 untrusted_rest.encode('ascii'),
-                expected_backend_domain=backend_domain,
-                expected_ident=ident,
-                expected_devclass=None
+                expected_device=Device(backend_domain, ident)
             )
             if not required_only or assignment.required:
                 yield assignment

qubesadmin/tests/devices.py

@@ -19,7 +19,7 @@
 # with this program; if not, see <http://www.gnu.org/licenses/>.
 
 import qubesadmin.tests
-import qubesadmin.devices
+import qubesadmin.device_protocol
 
 
 serialized_test_device = (
@@ -43,7 +43,7 @@ def test_000_available(self):
         devices = list(self.vm.devices['test'].get_exposed_devices())
         self.assertEqual(len(devices), 1)
         dev = devices[0]
-        self.assertIsInstance(dev, qubesadmin.devices.DeviceInfo)
+        self.assertIsInstance(dev, qubesadmin.device_protocol.DeviceInfo)
         self.assertEqual(dev.backend_domain, self.vm)
         self.assertEqual(dev.ident, 'dev1')
         self.assertEqual(
@@ -59,7 +59,7 @@ def test_001_available_desc(self):
         devices = list(self.vm.devices['test'].get_exposed_devices())
         self.assertEqual(len(devices), 1)
         dev = devices[0]
-        self.assertIsInstance(dev, qubesadmin.devices.DeviceInfo)
+        self.assertIsInstance(dev, qubesadmin.device_protocol.DeviceInfo)
         self.assertEqual(dev.backend_domain, self.vm)
         self.assertEqual(dev.ident, 'dev1')
         self.assertEqual(dev.description, 'test-device (itl)')
@@ -73,7 +73,7 @@ def test_002_available_options(self):
         devices = list(self.vm.devices['test'].get_exposed_devices())
         self.assertEqual(len(devices), 1)
         dev = devices[0]
-        self.assertIsInstance(dev, qubesadmin.devices.DeviceInfo)
+        self.assertIsInstance(dev, qubesadmin.device_protocol.DeviceInfo)
         self.assertEqual(dev.backend_domain, self.vm)
         self.assertEqual(dev.ident, 'dev1')
         self.assertEqual(dev.description, 'test-device (itl)')
@@ -86,7 +86,7 @@ def test_010_getitem(self):
             ('test-vm', 'admin.vm.device.test.Available', None, None)] = \
             serialized_test_device + b"\n"
         dev = self.vm.devices['test']['dev1']
-        self.assertIsInstance(dev, qubesadmin.devices.DeviceInfo)
+        self.assertIsInstance(dev, qubesadmin.device_protocol.DeviceInfo)
         self.assertEqual(dev.backend_domain, self.vm)
         self.assertEqual(dev.ident, 'dev1')
         self.assertEqual(dev.description, 'test-device (itl)')
@@ -99,7 +99,7 @@ def test_011_getitem_missing(self):
             ('test-vm', 'admin.vm.device.test.Available', None, None)] = \
             serialized_test_device + b"\n"
         dev = self.vm.devices['test']['dev2']
-        self.assertIsInstance(dev, qubesadmin.devices.UnknownDevice)
+        self.assertIsInstance(dev, qubesadmin.device_protocol.UnknownDevice)
         self.assertEqual(dev.backend_domain, self.vm)
         self.assertEqual(dev.ident, 'dev2')
         self.assertEqual(dev.description,
@@ -115,7 +115,7 @@ def test_020_attach(self):
              b"devclass='test' backend_domain='test-vm2' "
              b"frontend_domain='test-vm'")] = \
             b'0\0'
-        assign = qubesadmin.devices.DeviceAssignment(
+        assign = qubesadmin.device_protocol.DeviceAssignment(
             self.app.domains['test-vm2'], 'dev1')
         self.vm.devices['test'].attach(assign)
         self.assertAllCalled()
@@ -127,7 +127,7 @@ def test_021_attach_options(self):
              b"devclass='test' backend_domain='test-vm2' "
              b"frontend_domain='test-vm' _ro='True' "
              b"_something='value'")] = b'0\0'
-        assign = qubesadmin.devices.DeviceAssignment(
+        assign = qubesadmin.device_protocol.DeviceAssignment(
             self.app.domains['test-vm2'], 'dev1')
         assign.options['ro'] = True
         assign.options['something'] = 'value'
@@ -140,7 +140,7 @@ def test_022_attach_required(self):
              b"required='yes' attach_automatically='yes' ident='dev1' "
              b"devclass='test' backend_domain='test-vm2' "
              b"frontend_domain='test-vm'")] = b'0\0'
-        assign = qubesadmin.devices.DeviceAssignment(
+        assign = qubesadmin.device_protocol.DeviceAssignment(
             self.app.domains['test-vm2'], 'dev1',
             attach_automatically=True, required=True)
         self.vm.devices['test'].attach(assign)
@@ -152,7 +152,7 @@ def test_023_attach_required_options(self):
              b"required='yes' attach_automatically='yes' ident='dev1' "
              b"devclass='test' backend_domain='test-vm2' "
              b"frontend_domain='test-vm' _ro='True'")] = b'0\0'
-        assign = qubesadmin.devices.DeviceAssignment(
+        assign = qubesadmin.device_protocol.DeviceAssignment(
             self.app.domains['test-vm2'], 'dev1',
             attach_automatically=True, required=True)
         assign.options['ro'] = True
@@ -163,7 +163,7 @@ def test_030_detach(self):
         self.app.expected_calls[
             ('test-vm', 'admin.vm.device.test.Detach', 'test-vm2+dev1',
              None)] = b'0\0'
-        assign = qubesadmin.devices.DeviceAssignment(
+        assign = qubesadmin.device_protocol.DeviceAssignment(
             self.app.domains['test-vm2'], 'dev1')
         self.vm.devices['test'].detach(assign)
         self.assertAllCalled()
@@ -188,7 +188,7 @@ def test_040_dedicated(self):
         dedicated = sorted(list(
             self.vm.devices['test'].get_dedicated_devices()))
         self.assertEqual(len(dedicated), 2)
-        self.assertIsInstance(dedicated[0], qubesadmin.devices.DeviceAssignment)
+        self.assertIsInstance(dedicated[0], qubesadmin.device_protocol.DeviceAssignment)
         self.assertEqual(dedicated[0].backend_domain,
                          self.app.domains['test-vm2'])
         self.assertEqual(dedicated[0].ident, 'dev1')
@@ -199,7 +199,7 @@ def test_040_dedicated(self):
         self.assertEqual(dedicated[0].device,
                          self.app.domains['test-vm2'].devices['test']['dev1'])
 
-        self.assertIsInstance(dedicated[1], qubesadmin.devices.DeviceAssignment)
+        self.assertIsInstance(dedicated[1], qubesadmin.device_protocol.DeviceAssignment)
         self.assertEqual(dedicated[1].backend_domain,
                          self.app.domains['test-vm3'])
         self.assertEqual(dedicated[1].ident, 'dev2')
@@ -226,7 +226,7 @@ def test_041_assignments_options(self):
         assigns = sorted(list(
             self.vm.devices['test'].get_dedicated_devices()))
         self.assertEqual(len(assigns), 2)
-        self.assertIsInstance(assigns[0], qubesadmin.devices.DeviceAssignment)
+        self.assertIsInstance(assigns[0], qubesadmin.device_protocol.DeviceAssignment)
         self.assertEqual(assigns[0].backend_domain,
                          self.app.domains['test-vm2'])
         self.assertEqual(assigns[0].ident, 'dev1')
@@ -236,7 +236,7 @@ def test_041_assignments_options(self):
         self.assertEqual(assigns[0].required, False)
         self.assertEqual(assigns[0].devclass, 'test')
 
-        self.assertIsInstance(assigns[1], qubesadmin.devices.DeviceAssignment)
+        self.assertIsInstance(assigns[1], qubesadmin.device_protocol.DeviceAssignment)
         self.assertEqual(assigns[1].backend_domain,
                          self.app.domains['test-vm3'])
         self.assertEqual(assigns[1].ident, 'dev2')
@@ -258,7 +258,7 @@ def test_050_required(self):
         devs = list(self.vm.devices['test'].get_assigned_devices(
             required_only=True))
         self.assertEqual(len(devs), 1)
-        self.assertIsInstance(devs[0], qubesadmin.devices.DeviceAssignment)
+        self.assertIsInstance(devs[0], qubesadmin.device_protocol.DeviceAssignment)
         self.assertEqual(devs[0].backend_domain, self.app.domains['test-vm3'])
         self.assertEqual(devs[0].ident, 'dev2')
         self.assertAllCalled()
@@ -272,10 +272,10 @@ def test_060_attached(self):
              b"attach_automatically='yes' required='no'\n")
         devs = list(self.vm.devices['test'].get_attached_devices())
         self.assertEqual(len(devs), 2)
-        self.assertIsInstance(devs[0], qubesadmin.devices.DeviceAssignment)
+        self.assertIsInstance(devs[0], qubesadmin.device_protocol.DeviceAssignment)
         self.assertEqual(devs[0].backend_domain, self.app.domains['test-vm2'])
         self.assertEqual(devs[0].ident, 'dev1')
-        self.assertIsInstance(devs[1], qubesadmin.devices.DeviceAssignment)
+        self.assertIsInstance(devs[1], qubesadmin.device_protocol.DeviceAssignment)
         self.assertEqual(devs[1].backend_domain, self.app.domains['test-vm3'])
         self.assertEqual(devs[1].ident, 'dev2')
         self.assertAllCalled()
@@ -284,7 +284,7 @@ def test_070_update_assignment(self):
         self.app.expected_calls[
             ('test-vm', 'admin.vm.device.test.Set.assignment', 'test-vm2+dev1',
              b'True')] = b'0\0'
-        dev = qubesadmin.devices.DeviceAssignment(
+        dev = qubesadmin.device_protocol.DeviceAssignment(
             self.app.domains['test-vm2'], devclass='test', ident='dev1')
         self.vm.devices['test'].update_assignment(dev, True)
         self.assertAllCalled()
@@ -293,7 +293,7 @@ def test_071_update_assignment_false(self):
         self.app.expected_calls[
             ('test-vm', 'admin.vm.device.test.Set.assignment', 'test-vm2+dev1',
              b'False')] = b'0\0'
-        dev = qubesadmin.devices.DeviceAssignment(
+        dev = qubesadmin.device_protocol.DeviceAssignment(
             self.app.domains['test-vm2'], devclass='test', ident='dev1')
         self.vm.devices['test'].update_assignment(dev, False)
         self.assertAllCalled()
@@ -302,7 +302,7 @@ def test_072_update_assignment_none(self):
         self.app.expected_calls[
             ('test-vm', 'admin.vm.device.test.Set.assignment', 'test-vm2+dev1',
              b'None')] = b'0\0'
-        dev = qubesadmin.devices.DeviceAssignment(
+        dev = qubesadmin.device_protocol.DeviceAssignment(
             self.app.domains['test-vm2'], devclass='test', ident='dev1')
         self.vm.devices['test'].update_assignment(dev, None)
         self.assertAllCalled()

qubesadmin/tests/tools/qvm_device.py

@@ -26,7 +26,7 @@
 import unittest.mock as mock
 import qubesadmin.tests
 import qubesadmin.tests.tools
-import qubesadmin.devices
+import qubesadmin.device_protocol
 import qubesadmin.tools.qvm_device
 
 

qubesadmin/tools/qvm_device.py

@@ -32,7 +32,7 @@
 import qubesadmin
 import qubesadmin.exc
 import qubesadmin.tools
-import qubesadmin.devices
+import qubesadmin.device_protocol
 
 
 def prepare_table(dev_list):
@@ -65,7 +65,7 @@ class Line(object):
     """Helper class to hold single device info for listing"""
 
     # pylint: disable=too-few-public-methods
-    def __init__(self, device: qubesadmin.devices.DeviceInfo, attached_to=None):
+    def __init__(self, device: qubesadmin.device_protocol.DeviceInfo, attached_to=None):
         self.ident = "{!s}:{!s}".format(device.backend_domain, device.ident)
         self.description = device.description
         self.attached_to = attached_to if attached_to else ""
@@ -136,7 +136,7 @@ def attach_device(args):
     """
     vm = args.domains[0]
     device = args.device
-    assignment = qubesadmin.devices.DeviceAssignment.from_device(
+    assignment = qubesadmin.device_protocol.DeviceAssignment.from_device(
         device,
         # backward compatibility
         attach_automatically=args.required, required=args.required)
@@ -157,7 +157,7 @@ def detach_device(args):
     vm = args.domains[0]
     if args.device:
         device = args.device
-        assignment = qubesadmin.devices.DeviceAssignment.from_device(device)
+        assignment = qubesadmin.device_protocol.DeviceAssignment.from_device(device)
         vm.devices[args.devclass].detach(assignment)
     else:
         for device_assignment in (
@@ -171,7 +171,7 @@ def assign_device(args):
     """
     vm = args.domains[0]
     device = args.device
-    assignment = qubesadmin.devices.DeviceAssignment.from_device(
+    assignment = qubesadmin.device_protocol.DeviceAssignment.from_device(
         device, required=args.required, attach_automatically=True)
     options = dict(opt.split('=', 1) for opt in args.option or [])
     if args.ro:
@@ -194,7 +194,7 @@ def unassign_device(args):
     vm = args.domains[0]
     if args.device:
         device = args.device
-        assignment = qubesadmin.devices.DeviceAssignment.from_device(
+        assignment = qubesadmin.device_protocol.DeviceAssignment.from_device(
             device, frontend_domain=vm)
         _unassign_and_show_message(assignment, vm, args)
     else:
@@ -217,7 +217,7 @@ def info_device(args):
     vm = args.domains[0]
     if args.device:
         device = args.device
-        assignment = qubesadmin.devices.DeviceAssignment.from_device(device)
+        assignment = qubesadmin.device_protocol.DeviceAssignment.from_device(device)
         print("description:", assignment.device.description)
         print("data:", assignment.device.data)
     else:
@@ -275,13 +275,13 @@ def parse_qubes_app(self, parser, namespace):
             try:
                 dev = vm.devices[devclass][device_id]
                 if not self.allow_unknown and \
-                        isinstance(dev, qubesadmin.devices.UnknownDevice):
+                        isinstance(dev, qubesadmin.device_protocol.UnknownDevice):
                     raise KeyError(device_id)
             except KeyError:
                 parser.error_runtime(
                     f"backend vm {vmname!r} doesn't expose "
                     f"{devclass} device {device_id!r}")
-                dev = qubesadmin.devices.Device(vm, device_id, devclass)
+                dev = qubesadmin.device_protocol.Device(vm, device_id, devclass)
             setattr(namespace, self.dest, dev)
         except ValueError:
             parser.error(

qubesadmin/tools/qvm_start.py

@@ -27,7 +27,7 @@
 
 import time
 
-import qubesadmin.devices
+import qubesadmin.device_protocol
 import qubesadmin.exc
 import qubesadmin.tools
 
@@ -79,7 +79,7 @@ def __call__(self, parser, namespace, values, option_string=None):
 
 
 def get_drive_assignment(app, drive_str):
-    ''' Prepare :py:class:`qubesadmin.devices.DeviceAssignment` object for a
+    ''' Prepare :py:class:`qubesadmin.device_protocol.DeviceAssignment` object for a
     given drive.
 
     If running in dom0, it will also take care about creating appropriate
@@ -143,7 +143,7 @@ def get_drive_assignment(app, drive_str):
         # FIXME: convert this to waiting for event
         timeout = 10
         while isinstance(backend_domain.devices['block'][ident],
-                qubesadmin.devices.UnknownDevice):
+                qubesadmin.device_protocol.UnknownDevice):
             if timeout == 0:
                 raise qubesadmin.exc.QubesException(
                     'Timeout waiting for {}:{} device to appear'.format(
@@ -155,7 +155,7 @@ def get_drive_assignment(app, drive_str):
         'devtype': devtype,
         'read-only': devtype == 'cdrom'
     }
-    assignment = qubesadmin.devices.DeviceAssignment(
+    assignment = qubesadmin.device_protocol.DeviceAssignment(
         backend_domain,
         ident,
         options=options,

qubesadmin/vm/__init__.py

@@ -31,6 +31,7 @@
 import qubesadmin.storage
 import qubesadmin.features
 import qubesadmin.devices
+import qubesadmin.device_protocol
 import qubesadmin.firewall
 import qubesadmin.tags