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

devices: make PEP8 happy

Browse Source
This commit is contained in:
Frédéric Pierret (fepitre) 2019-08-06 11:30:32 +02:00
parent 7bca004532
commit 1052217973
No known key found for this signature in database
GPG Key ID: 484010B5CDC576E2
1 changed files with 68 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

126
qubesadmin/devices.py
View File

@ -20,7 +20,7 @@
# You should have received a copy of the GNU Lesser General Public License along # You should have received a copy of the GNU Lesser General Public License along
# with this program; if not, see <http://www.gnu.org/licenses/>. # with this program; if not, see <http://www.gnu.org/licenses/>.
'''API for various types of devices. """API for various types of devices.
Main concept is that some domain main Main concept is that some domain main
expose (potentially multiple) devices, which can be attached to other domains. expose (potentially multiple) devices, which can be attached to other domains.
@ -29,13 +29,14 @@ class is implemented by an extension.
Devices are identified by pair of (backend domain, `ident`), where `ident` is Devices are identified by pair of (backend domain, `ident`), where `ident` is
:py:class:`str`. :py:class:`str`.
''' """
class DeviceAssignment(object): # pylint: disable=too-few-public-methods class DeviceAssignment(object): # pylint: disable=too-few-public-methods
''' Maps a device to a frontend_domain. ''' """ Maps a device to a frontend_domain. """
def __init__(self, backend_domain, ident, options=None, def __init__(self, backend_domain, ident, options=None, persistent=False,
persistent=False, frontend_domain=None, devclass=None): frontend_domain=None, devclass=None):
self.backend_domain = backend_domain self.backend_domain = backend_domain
self.ident = ident self.ident = ident
self.devclass = devclass self.devclass = devclass
@ -54,10 +55,10 @@ class DeviceAssignment(object): # pylint: disable=too-few-public-methods
return NotImplemented return NotImplemented
return self.backend_domain == other.backend_domain \ return self.backend_domain == other.backend_domain \
and self.ident == other.ident and self.ident == other.ident
def clone(self): def clone(self):
'''Clone object instance''' """Clone object instance"""
return self.__class__( return self.__class__(
self.backend_domain, self.backend_domain,
self.ident, self.ident,
@ -69,12 +70,13 @@ class DeviceAssignment(object): # pylint: disable=too-few-public-methods
@property @property
def device(self): def device(self):
'''Get DeviceInfo object corresponding to this DeviceAssignment''' """Get DeviceInfo object corresponding to this DeviceAssignment"""
return self.backend_domain.devices[self.devclass][self.ident] return self.backend_domain.devices[self.devclass][self.ident]
class DeviceInfo(object): class DeviceInfo(object):
''' Holds all information about a device ''' """ Holds all information about a device """
# pylint: disable=too-few-public-methods # pylint: disable=too-few-public-methods
def __init__(self, backend_domain, devclass, ident, description=None, def __init__(self, backend_domain, devclass, ident, description=None,
**kwargs): **kwargs):
@ -94,9 +96,9 @@ class DeviceInfo(object):
def __eq__(self, other): def __eq__(self, other):
try: try:
return ( return (
self.devclass == other.devclass and self.devclass == other.devclass and
self.backend_domain == other.backend_domain and self.backend_domain == other.backend_domain and
self.ident == other.ident self.ident == other.ident
) )
except AttributeError: except AttributeError:
return False return False
@ -107,35 +109,36 @@ class DeviceInfo(object):
class UnknownDevice(DeviceInfo): class UnknownDevice(DeviceInfo):
# pylint: disable=too-few-public-methods # pylint: disable=too-few-public-methods
'''Unknown device - for example exposed by domain not running currently''' """Unknown device - for example exposed by domain not running currently"""
def __init__(self, backend_domain, devclass, ident, description=None, def __init__(self, backend_domain, devclass, ident, description=None,
**kwargs): **kwargs):
if description is None: if description is None:
description = "Unknown device" description = "Unknown device"
super(UnknownDevice, self).__init__(backend_domain, devclass, ident, super(UnknownDevice, self).__init__(backend_domain, devclass, ident,
description, **kwargs) description, **kwargs)
class DeviceCollection(object): class DeviceCollection(object):
'''Bag for devices. """Bag for devices.
Used as default value for :py:meth:`DeviceManager.__missing__` factory. Used as default value for :py:meth:`DeviceManager.__missing__` factory.
:param vm: VM for which we manage devices :param vm: VM for which we manage devices
:param class_: device class :param class_: device class
''' """
def __init__(self, vm, class_): def __init__(self, vm, class_):
self._vm = vm self._vm = vm
self._class = class_ self._class = class_
self._dev_cache = {} self._dev_cache = {}
def attach(self, device_assignment): def attach(self, device_assignment):
'''Attach (add) device to domain. """Attach (add) device to domain.
:param DeviceAssignment device_assignment: device object :param DeviceAssignment device_assignment: device object
''' """
if not device_assignment.frontend_domain: if not device_assignment.frontend_domain:
device_assignment.frontend_domain = self._vm device_assignment.frontend_domain = self._vm
@ -150,20 +153,21 @@ class DeviceCollection(object):
options = device_assignment.options.copy() options = device_assignment.options.copy()
if device_assignment.persistent: if device_assignment.persistent:
options['persistent'] = 'True' options['persistent'] = 'True'
options_str = ' '.join('{}={}'.format(opt, options_str = ' '.join('{}={}'.format(opt, val)
val) for opt, val in sorted(options.items())) for opt, val in sorted(options.items()))
self._vm.qubesd_call(None, self._vm.qubesd_call(None,
'admin.vm.device.{}.Attach'.format(self._class), 'admin.vm.device.{}.Attach'.format(self._class),
'{!s}+{!s}'.format(device_assignment.backend_domain, '{!s}+{!s}'.format(
device_assignment.ident), device_assignment.backend_domain,
options_str.encode('utf-8')) device_assignment.ident),
options_str.encode('utf-8'))
def detach(self, device_assignment): def detach(self, device_assignment):
'''Detach (remove) device from domain. """Detach (remove) device from domain.
:param DeviceAssignment device_assignment: device to detach :param DeviceAssignment device_assignment: device to detach
(obtained from :py:meth:`assignments`) (obtained from :py:meth:`assignments`)
''' """
if not device_assignment.frontend_domain: if not device_assignment.frontend_domain:
device_assignment.frontend_domain = self._vm device_assignment.frontend_domain = self._vm
else: else:
@ -175,12 +179,13 @@ class DeviceCollection(object):
assert device_assignment.devclass == self._class assert device_assignment.devclass == self._class
self._vm.qubesd_call(None, self._vm.qubesd_call(None,
'admin.vm.device.{}.Detach'.format(self._class), 'admin.vm.device.{}.Detach'.format(self._class),
'{!s}+{!s}'.format(device_assignment.backend_domain, '{!s}+{!s}'.format(
device_assignment.ident)) device_assignment.backend_domain,
device_assignment.ident))
def assignments(self, persistent=None): def assignments(self, persistent=None):
'''List assignments for devices which are (or may be) attached to the """List assignments for devices which are (or may be) attached to the
vm. vm.
Devices may be attached persistently (so they are included in Devices may be attached persistently (so they are included in
@ -189,73 +194,79 @@ class DeviceCollection(object):
:param bool persistent: only include devices which are or are not :param bool persistent: only include devices which are or are not
attached persistently. attached persistently.
''' """
assignments_str = self._vm.qubesd_call(None, assignments_str = self._vm.qubesd_call(None,
'admin.vm.device.{}.List'.format(self._class)).decode() 'admin.vm.device.{}.List'.format(
self._class)).decode()
for assignment_str in assignments_str.splitlines(): for assignment_str in assignments_str.splitlines():
device, _, options_all = assignment_str.partition(' ') device, _, options_all = assignment_str.partition(' ')
backend_domain, ident = device.split('+', 1) backend_domain, ident = device.split('+', 1)
options = dict(opt_single.split('=', 1) options = dict(opt_single.split('=', 1)
for opt_single in options_all.split(' ') if opt_single) for opt_single in options_all.split(' ') if
opt_single)
dev_persistent = (options.pop('persistent', False) in dev_persistent = (options.pop('persistent', False) in
['True', 'yes', True]) ['True', 'yes', True])
if persistent is not None and dev_persistent != persistent: if persistent is not None and dev_persistent != persistent:
continue continue
backend_domain = self._vm.app.domains[backend_domain] backend_domain = self._vm.app.domains[backend_domain]
yield DeviceAssignment(backend_domain, ident, options, yield DeviceAssignment(backend_domain, ident, options,
persistent=dev_persistent, frontend_domain=self._vm, persistent=dev_persistent,
devclass=self._class) frontend_domain=self._vm,
devclass=self._class)
def attached(self): def attached(self):
'''List devices which are (or may be) attached to this vm ''' """List devices which are (or may be) attached to this vm """
for assignment in self.assignments(): for assignment in self.assignments():
yield assignment.device yield assignment.device
def persistent(self): def persistent(self):
''' Devices persistently attached and safe to access before libvirt """ Devices persistently attached and safe to access before libvirt
bootstrap. bootstrap.
''' """
for assignment in self.assignments(True): for assignment in self.assignments(True):
yield assignment.device yield assignment.device
def available(self): def available(self):
'''List devices exposed by this vm''' """List devices exposed by this vm"""
devices_str = self._vm.qubesd_call(None, devices_str = \
'admin.vm.device.{}.Available'.format(self._class)).decode() self._vm.qubesd_call(None,
'admin.vm.device.{}.Available'.format(
self._class)).decode()
for dev_str in devices_str.splitlines(): for dev_str in devices_str.splitlines():
ident, _, info = dev_str.partition(' ') ident, _, info = dev_str.partition(' ')
# description is special that it can contain spaces # description is special that it can contain spaces
info, _, description = info.partition('description=') info, _, description = info.partition('description=')
info_dict = dict(info_single.split('=', 1) info_dict = dict(info_single.split('=', 1)
for info_single in info.split(' ') if info_single) for info_single in info.split(' ') if info_single)
yield DeviceInfo(self._vm, self._class, ident, yield DeviceInfo(self._vm, self._class, ident,
description=description, description=description,
**info_dict) **info_dict)
def update_persistent(self, device, persistent): def update_persistent(self, device, persistent):
'''Update `persistent` flag of already attached device. """Update `persistent` flag of already attached device.
:param DeviceInfo device: device for which change persistent flag :param DeviceInfo device: device for which change persistent flag
:param bool persistent: new persistent flag :param bool persistent: new persistent flag
''' """
self._vm.qubesd_call(None, self._vm.qubesd_call(None,
'admin.vm.device.{}.Set.persistent'.format(self._class), 'admin.vm.device.{}.Set.persistent'.format(
'{!s}+{!s}'.format(device.backend_domain, self._class),
device.ident), '{!s}+{!s}'.format(device.backend_domain,
str(persistent).encode('utf-8')) device.ident),
str(persistent).encode('utf-8'))
__iter__ = available __iter__ = available
def clear_cache(self): def clear_cache(self):
'''Clear cache of available devices''' """Clear cache of available devices"""
self._dev_cache.clear() self._dev_cache.clear()
def __getitem__(self, item): def __getitem__(self, item):
'''Get device object with given ident. """Get device object with given ident.
:returns: py:class:`DeviceInfo` :returns: py:class:`DeviceInfo`
@ -263,7 +274,7 @@ class DeviceCollection(object):
so return UnknownDevice object. Also do the same for non-existing so return UnknownDevice object. Also do the same for non-existing
devices - otherwise it will be impossible to detach already devices - otherwise it will be impossible to detach already
disconnected device. disconnected device.
''' """
# fist, check if we have cached device info # fist, check if we have cached device info
if item in self._dev_cache: if item in self._dev_cache:
return self._dev_cache[item] return self._dev_cache[item]
@ -277,12 +288,11 @@ class DeviceCollection(object):
return UnknownDevice(self._vm, self._class, item) return UnknownDevice(self._vm, self._class, item)
class DeviceManager(dict): class DeviceManager(dict):
'''Device manager that hold all devices by their classess. """Device manager that hold all devices by their classes.
:param vm: VM for which we manage devices :param vm: VM for which we manage devices
''' """
def __init__(self, vm): def __init__(self, vm):
super(DeviceManager, self).__init__() super(DeviceManager, self).__init__()