Qubes/core-admin-client
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

devices: adjust API for 'devices: add assignment.device property' change

Browse Source 
Make it easy to retrieve DeviceInfo object out of DeviceAssignment
object. The only missing piece of information for that is device class,
so add it. Make it optional, as it can be filled on demand when passing
the object through DeviceCollection (either by listing devices, or
attaching/detaching).

This also makes DeviceCollection._device method not needed anymore.
This commit is contained in:
Marek Marczykowski-Górecki 2017-05-22 01:17:18 +02:00
parent 3edbc85282
commit c826378579
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
2 changed files with 46 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
qubesadmin/devices.py
View File

@ -34,10 +34,11 @@ Devices are identified by pair of (backend domain, `ident`), where `ident` is
class DeviceAssignment(object): # pylint: disable=too-few-public-methods
''' Maps a device to a frontend_domain. '''
def __init__(self, backend_domain, ident, options=None, persistent=False,
frontend_domain=None):
def __init__(self, backend_domain, ident, options=None,
persistent=False, frontend_domain=None, devclass=None):
self.backend_domain = backend_domain
self.ident = ident
self.devclass = devclass
self.options = options or {}
self.persistent = persistent
self.frontend_domain = frontend_domain
@ -55,6 +56,22 @@ class DeviceAssignment(object): # pylint: disable=too-few-public-methods
return self.backend_domain == other.backend_domain \
and self.ident == other.ident
def clone(self):
'''Clone object instance'''
return self.__class__(
self.backend_domain,
self.ident,
self.options,
self.persistent,
self.frontend_domain,
self.devclass,
)
@property
def device(self):
'''Get DeviceInfo object corresponding to this DeviceAssignment'''
return self.backend_domain.devices[self.devclass][self.ident]
class DeviceInfo(object):
''' Holds all information about a device '''
@ -120,6 +137,10 @@ class DeviceCollection(object):
else:
assert device_assignment.frontend_domain == self._vm, \
"Trying to attach DeviceAssignment belonging to other domain"
if device_assignment.devclass is None:
device_assignment.devclass = self._class
else:
assert device_assignment.devclass == self._class
options = device_assignment.options.copy()
if device_assignment.persistent:
@ -143,6 +164,10 @@ class DeviceCollection(object):
else:
assert device_assignment.frontend_domain == self._vm, \
"Trying to detach DeviceAssignment belonging to other domain"
if device_assignment.devclass is None:
device_assignment.devclass = self._class
else:
assert device_assignment.devclass == self._class
self._vm.qubesd_call(None,
'admin.vm.device.{}.Detach'.format(self._class),
@ -174,13 +199,14 @@ class DeviceCollection(object):
continue
backend_domain = self._vm.app.domains[backend_domain]
yield DeviceAssignment(backend_domain, ident, options,
persistent=dev_persistent, frontend_domain=self._vm)
persistent=dev_persistent, frontend_domain=self._vm,
devclass=self._class)
def attached(self):
'''List devices which are (or may be) attached to this vm '''
for assignment in self.assignments():
yield self._device(assignment)
yield assignment.device
def persistent(self):
''' Devices persistently attached and safe to access before libvirt
@ -188,13 +214,7 @@ class DeviceCollection(object):
'''
for assignment in self.assignments(True):
yield self._device(assignment)
def _device(self, assignment):
''' Helper method for geting a `qubes.devices.DeviceInfo` object from
`qubes.devices.DeviceAssignment`. '''
return assignment.backend_domain.devices[self._class][assignment.ident]
yield assignment.device
def available(self):
'''List devices exposed by this vm'''

15
qubesadmin/tests/devices.py
View File

@ -161,6 +161,12 @@ class TC_00_DeviceCollection(qubesadmin.tests.QubesTestCase):
('test-vm', 'admin.vm.device.test.List', None, None)] = \
b'0\0test-vm2+dev1\n' \
b'test-vm3+dev2\n'
self.app.expected_calls[
('test-vm2', 'admin.vm.device.test.Available', None, None)] = \
b'0\0dev1 description=desc\n'
self.app.expected_calls[
('test-vm3', 'admin.vm.device.test.Available', None, None)] = \
b'0\0dev2 description=desc\n'
assigns = list(self.vm.devices['test'].assignments())
self.assertEqual(len(assigns), 2)
self.assertIsInstance(assigns[0], qubesadmin.devices.DeviceAssignment)
@ -170,6 +176,9 @@ class TC_00_DeviceCollection(qubesadmin.tests.QubesTestCase):
self.assertEqual(assigns[0].frontend_domain,
self.app.domains['test-vm'])
self.assertEqual(assigns[0].options, {})
self.assertEqual(assigns[0].devclass, 'test')
self.assertEqual(assigns[0].device,
self.app.domains['test-vm2'].devices['test']['dev1'])
self.assertIsInstance(assigns[1], qubesadmin.devices.DeviceAssignment)
self.assertEqual(assigns[1].backend_domain,
@ -178,6 +187,9 @@ class TC_00_DeviceCollection(qubesadmin.tests.QubesTestCase):
self.assertEqual(assigns[1].frontend_domain,
self.app.domains['test-vm'])
self.assertEqual(assigns[1].options, {})
self.assertEqual(assigns[1].devclass, 'test')
self.assertEqual(assigns[1].device,
self.app.domains['test-vm3'].devices['test']['dev2'])
self.assertAllCalled()
@ -196,6 +208,7 @@ class TC_00_DeviceCollection(qubesadmin.tests.QubesTestCase):
self.app.domains['test-vm'])
self.assertEqual(assigns[0].options, {'ro': 'True'})
self.assertEqual(assigns[0].persistent, False)
self.assertEqual(assigns[0].devclass, 'test')
self.assertIsInstance(assigns[1], qubesadmin.devices.DeviceAssignment)
self.assertEqual(assigns[1].backend_domain,
@ -205,6 +218,7 @@ class TC_00_DeviceCollection(qubesadmin.tests.QubesTestCase):
self.app.domains['test-vm'])
self.assertEqual(assigns[1].options, {'ro': 'False'})
self.assertEqual(assigns[1].persistent, True)
self.assertEqual(assigns[1].devclass, 'test')
self.assertAllCalled()
@ -223,6 +237,7 @@ class TC_00_DeviceCollection(qubesadmin.tests.QubesTestCase):
self.app.domains['test-vm'])
self.assertEqual(assigns[0].options, {})
self.assertEqual(assigns[0].persistent, True)
self.assertEqual(assigns[0].devclass, 'test')
self.assertAllCalled()
def test_042_assignments_non_persistent(self):