Skip to content

Commit

Permalink
q-dev: add flag to device listing
Browse files Browse the repository at this point in the history
Listing all assignments by default could be overwhelming. Now to see all assignment a flag `-s` must be used. Assignments are indicated by '*' before qube name.
  • Loading branch information
piotrbartman committed Oct 14, 2024
1 parent 469ee24 commit e09ab84
Show file tree
Hide file tree
Showing 2 changed files with 50 additions and 27 deletions.
6 changes: 5 additions & 1 deletion doc/manpages/qvm-device.rst
Original file line number Diff line number Diff line change
Expand Up @@ -41,10 +41,14 @@ Commands
list
^^^^

| :command:`qvm-device` *DEVICE_CLASS* list [-h] [--verbose] [--quiet] [*VMNAME* [*VMNAME* ...]]
| :command:`qvm-device` *DEVICE_CLASS* list [-h] [--verbose] [--quiet] [-s] [*VMNAME* [*VMNAME* ...]]
List devices.

.. option:: --assignments, -s

Include info about device assignments, indicated by '*' before qube name.

.. option:: --all

List devices from all qubes. You can use :option:`--exclude` to limit the
Expand Down
71 changes: 45 additions & 26 deletions qubesadmin/tools/qvm_device.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,42 +69,51 @@ class Line:
"""Helper class to hold single device info for listing"""

# pylint: disable=too-few-public-methods
def __init__(self, device: DeviceInfo, attached_to=None):
def __init__(self, device: DeviceInfo, assignment=False):
self.ident = "{!s}:{!s}".format(
device.backend_domain, device.port_id)
self.description = device.description
self.attached_to = attached_to if attached_to else ""
self.assignment = assignment
self.frontends = []

@property
def assignments(self):
"""list of frontends the device is assigned to"""
return ', '.join(self.frontends)
fronts = (f'{"*" if self.assignment else ""}' + front
for front in self.frontends)
return ', '.join(fronts)


def list_devices(args):
"""
Called by the parser to execute the qubes-devices list subcommand. """
app = args.app

domains = args.domains if hasattr(args, 'domains') else None
devices = _load_devices(app, domains, args.devclass)

result = {dev: Line(dev) for dev in devices}

lines = _load_lines(args.app, domains, args.devclass, actual_devices=True)
lines = list(lines.values())
if args.assignments:
extra_lines = _load_lines(
args.app, domains, args.devclass, actual_devices=False)
lines += list(extra_lines.values())
qubesadmin.tools.print_table(prepare_table(lines))


def _load_lines(app, domains, devclass, actual_devices: bool):
devices = _load_devices(app, domains, devclass, actual_devices)
result = {dev: Line(dev, not actual_devices) for dev in devices}
for dev in result:
for vm in app.domains:
frontends = _load_frontends_info(vm, dev, args.devclass)
frontends = _load_frontends_info(vm, dev, devclass, actual_devices)
result[dev].frontends.extend(frontends)
return result

qubesadmin.tools.print_table(prepare_table(result.values()))


def _load_devices(app, domains, devclass):
def _load_devices(app, domains, devclass, actual_devices):
"""
Loads device exposed or connected to given domains.
If `domains` is empty/`None` load all devices.
If `actual_devices` is True only devices currently present will be included,
otherwise only device assignments
"""
devices = set()
if domains:
Expand All @@ -115,12 +124,14 @@ def _load_devices(app, domains, devclass):
try:
for vm in domains:
try:
for ass in vm.devices[devclass].get_attached_devices():
devices.add(ass.device)
for ass in vm.devices[devclass].get_assigned_devices():
devices.add(ass.virtual_device)
for dev in vm.devices[devclass].get_exposed_devices():
devices.add(dev)
if actual_devices:
for ass in vm.devices[devclass].get_attached_devices():
devices.add(ass.device)
for dev in vm.devices[devclass].get_exposed_devices():
devices.add(dev)
else:
for ass in vm.devices[devclass].get_assigned_devices():
devices.add(ass.virtual_device)
except qubesadmin.exc.QubesVMNotFoundError:
if ignore_errors:
continue
Expand All @@ -132,20 +143,22 @@ def _load_devices(app, domains, devclass):
return devices


def _load_frontends_info(vm, dev, devclass):
def _load_frontends_info(vm, dev, devclass, actual_devices):
"""
Returns string of vms to which a device is connected or `None`.
"""
if vm == dev.backend_domain:
return

try:
for assignment in vm.devices[devclass].get_attached_devices():
if dev in assignment.devices:
yield _frontend_desc(vm, assignment)
for assignment in vm.devices[devclass].get_assigned_devices():
if dev == assignment.virtual_device:
yield _frontend_desc(vm, assignment)
if actual_devices:
for assignment in vm.devices[devclass].get_attached_devices():
if dev in assignment.devices:
yield _frontend_desc(vm, assignment)
else:
for assignment in vm.devices[devclass].get_assigned_devices():
if dev == assignment.virtual_device:
yield _frontend_desc(vm, assignment)
except qubesadmin.exc.QubesVMNotFoundError:
pass

Expand Down Expand Up @@ -402,6 +415,12 @@ def init_list_parser(sub_parsers):
list_parser = sub_parsers.add_parser('list', aliases=('ls', 'l'),
help='list devices')

list_parser.add_argument('--assignments', '-s',
action='store_true',
default=False,
help="Include info about device assignments, "
"indicated by '*' before qube name.")

vm_name_group = qubesadmin.tools.VmNameGroup(
list_parser, required=False, vm_action=qubesadmin.tools.VmNameAction,
help='list devices assigned to specific domain(s)')
Expand Down

0 comments on commit e09ab84

Please sign in to comment.