Skip to content

Commit

Permalink
q-dev: fix qvm-device
Browse files Browse the repository at this point in the history
  • Loading branch information
@piotrbartman
piotrbartman committed Oct 14, 2024
1 parent 250afc8 commit 922bce8
Show file tree
Hide file tree
Showing 2 changed files with 63 additions and 21 deletions.
72 changes: 54 additions & 18 deletions qubesadmin/tests/tools/qvm_device.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ def test_001_list_assigned_required(self):
b"devclass='testclass' backend_domain='test-vm2'\n")
self.expected_device_call(
'test-vm3', 'Available',
b"0\0dev3 port_id='dev3' device_id='0000:0000::?******' "
b"0\0dev3 port_id='dev3' device_id='0000:0000::p000000' "
b"devclass='testclass' backend_domain='test-vm3' "
b"vendor='evil inc.' product='test-device-3'\n"
)
Expand All @@ -103,9 +103,9 @@ def test_001_list_assigned_required(self):
self.expected_device_call(
'test-vm2', 'Assigned',
b"0\0test-vm1+dev1 port_id='dev1' devclass='testclass' "
b"backend_domain='test-vm1' "
b"mode='required' _option='other option' _extra_opt='yes'\n"
b"test-vm3+dev3 device_id='0000:0000::?******' port_id='dev3' "
b"backend_domain='test-vm1' mode='required' _option='other option' "
b"_extra_opt='yes'\n"
b"test-vm3+dev3 device_id='0000:0000::p000000' port_id='dev3' "
b"devclass='testclass' backend_domain='test-vm3' mode='required'\n"
)
self.expected_device_call(
Expand All @@ -116,14 +116,14 @@ def test_001_list_assigned_required(self):

with qubesadmin.tests.tools.StdoutBuffer() as buf:
qubesadmin.tools.qvm_device.main(
['testclass', 'list', 'test-vm3'], app=self.app)
['testclass', 'list', '-s', 'test-vm3'], app=self.app)
self.assertEqual(
buf.getvalue(),
'test-vm1:dev1 any device '
'test-vm2 (required: option=other option, extra_opt=yes), '
'test-vm3 (required: option=test option)\n'
'test-vm3:dev3 ?******: evil inc. test-device-3 '
'test-vm2 (required)\n'
'test-vm1:dev1 any device '
'*test-vm2 (required: option=other option, extra_opt=yes), '
'*test-vm3 (required: option=test option)\n'
'test-vm3:dev3 0000:0000::p000000 *test-vm2 (required)\n'
'test-vm3:dev3 ?******: evil inc. test-device-3 \n'
)

def test_002_list_attach(self):
Expand Down Expand Up @@ -152,7 +152,7 @@ def test_002_list_attach(self):
self.assertEqual(
buf.getvalue(),
'test-vm1:dev1 Multimedia: itl test-device '
'test-vm3 (required)\n'
'test-vm3 (attached)\n'
)

def test_003_list_device_classes(self):
Expand Down Expand Up @@ -233,7 +233,7 @@ def test_020_detach(self):
""" Test detach action """
self.app.expected_calls[
('test-vm2', 'admin.vm.device.testclass.Detach',
'test-vm1+dev1:*', None)] = b'0\0'
'test-vm1+dev1:dead:beef:babe:u012345', None)] = b'0\0'
qubesadmin.tools.qvm_device.main(
['testclass', 'detach', 'test-vm2', 'test-vm1:dev1'], app=self.app)
self.assertAllCalled()
Expand All @@ -242,7 +242,7 @@ def test_021_detach_unknown(self):
""" Test detach action """
self.app.expected_calls[
('test-vm2', 'admin.vm.device.testclass.Detach',
'test-vm1+dev7:*', None)] = b'0\0'
'test-vm1+dev7:0000:0000::?******', None)] = b'0\0'
qubesadmin.tools.qvm_device.main(
['testclass', 'detach', 'test-vm2', 'test-vm1:dev7'], app=self.app)
self.assertAllCalled()
Expand Down Expand Up @@ -272,6 +272,9 @@ def test_030_assign(self):
b"devclass='testclass' backend_domain='test-vm1' "
b"mode='auto-attach' frontend_domain='test-vm2'"
)] = b'0\0'
self.app.expected_calls[(
'test-vm2', 'admin.vm.device.testclass.Attached', None, None
)] = b'0\0'
qubesadmin.tools.qvm_device.main(
['testclass', 'assign', 'test-vm2', 'test-vm1:dev1'], app=self.app)
self.assertAllCalled()
Expand All @@ -286,6 +289,9 @@ def test_031_assign_required(self):
b"devclass='testclass' backend_domain='test-vm1' mode='required' "
b"frontend_domain='test-vm2'"
)] = b'0\0'
self.app.expected_calls[(
'test-vm2', 'admin.vm.device.testclass.Attached', None, None
)] = b'0\0'
qubesadmin.tools.qvm_device.main(
['testclass', 'assign', '--required', 'test-vm2', 'test-vm1:dev1'], app=self.app)
self.assertAllCalled()
Expand All @@ -300,6 +306,9 @@ def test_032_assign_ask_and_options(self):
b"devclass='testclass' backend_domain='test-vm1' "
b"mode='ask-to-attach' frontend_domain='test-vm2' _read-only='yes'"
)] = b'0\0'
self.app.expected_calls[(
'test-vm2', 'admin.vm.device.testclass.Attached', None, None
)] = b'0\0'
with qubesadmin.tests.tools.StdoutBuffer() as buf:
qubesadmin.tools.qvm_device.main(
['testclass', 'assign', '--ro', '--ask', 'test-vm2',
Expand All @@ -324,11 +333,11 @@ def test_033_assign_invalid(self):
def test_034_assign_invalid_device(self):
""" Test attach action """
with qubesadmin.tests.tools.StderrBuffer() as stderr:
with self.assertRaises(SystemExit):
qubesadmin.tools.qvm_device.main(
retcode = qubesadmin.tools.qvm_device.main(
['testclass', 'assign', 'test-vm2', 'test-vm1:invalid'],
app=self.app)
self.assertIn('doesn\'t expose testclass device', stderr.getvalue())
self.assertEqual(retcode, 1)
self.assertIn("doesn't expose testclass device", stderr.getvalue())
self.assertAllCalled()

def test_035_assign_invalid_backend(self):
Expand All @@ -351,6 +360,9 @@ def test_036_assign_port(self):
b"devclass='testclass' backend_domain='test-vm1' "
b"mode='auto-attach' frontend_domain='test-vm2'"
)] = b'0\0'
self.app.expected_calls[(
'test-vm2', 'admin.vm.device.testclass.Attached', None, None
)] = b'0\0'
qubesadmin.tools.qvm_device.main(
['testclass', 'assign', 'test-vm2', 'test-vm1:dev1', '--port'],
app=self.app)
Expand All @@ -366,6 +378,9 @@ def test_037_assign_port_asterisk(self):
b"devclass='testclass' backend_domain='test-vm1' "
b"mode='auto-attach' frontend_domain='test-vm2'"
)] = b'0\0'
self.app.expected_calls[(
'test-vm2', 'admin.vm.device.testclass.Attached', None, None
)] = b'0\0'
qubesadmin.tools.qvm_device.main(
['testclass', 'assign', 'test-vm2', 'test-vm1:dev1:*'],
app=self.app)
Expand All @@ -381,6 +396,9 @@ def test_038_assign_device_from_port(self):
b"devclass='testclass' backend_domain='test-vm1' "
b"mode='auto-attach' frontend_domain='test-vm2'"
)] = b'0\0'
self.app.expected_calls[(
'test-vm2', 'admin.vm.device.testclass.Attached', None, None
)] = b'0\0'
qubesadmin.tools.qvm_device.main(
['testclass', 'assign', 'test-vm2', 'test-vm1:dev1', '--device'],
app=self.app)
Expand Down Expand Up @@ -433,14 +451,17 @@ def test_041_assign_denied_device(self, mock_deny_list):
qubesadmin.tools.qvm_device.main(
['testclass', 'assign', '--ask', 'test-vm2', 'test-vm1:dev1'],
app=self.app)
self.assertIn('Attention:', buf.getvalue())
self.assertIn('Warning:', buf.getvalue())
self.assertAllCalled()

def test_050_unassign(self):
""" Test unassign action """
self.app.expected_calls[
('test-vm2', 'admin.vm.device.testclass.Unassign',
'test-vm1+dev1:dead:beef:babe:u012345', None)] = b'0\0'
self.app.expected_calls[(
'test-vm2', 'admin.vm.device.testclass.Attached', None, None
)] = b'0\0'
qubesadmin.tools.qvm_device.main(
['testclass', 'unassign', 'test-vm2', 'test-vm1:dev1'],
app=self.app)
Expand All @@ -451,6 +472,9 @@ def test_051_unassign_unknown(self):
self.app.expected_calls[
('test-vm2', 'admin.vm.device.testclass.Unassign',
'test-vm1+dev7:0000:0000::?******', None)] = b'0\0'
self.app.expected_calls[(
'test-vm2', 'admin.vm.device.testclass.Attached', None, None
)] = b'0\0'
qubesadmin.tools.qvm_device.main(
['testclass', 'unassign', 'test-vm2', 'test-vm1:dev7'],
app=self.app)
Expand All @@ -461,6 +485,9 @@ def test_052_unassign_port(self):
self.app.expected_calls[
('test-vm2', 'admin.vm.device.testclass.Unassign',
'test-vm1+dev1:*', None)] = b'0\0'
self.app.expected_calls[(
'test-vm2', 'admin.vm.device.testclass.Attached', None, None
)] = b'0\0'
qubesadmin.tools.qvm_device.main(
['testclass', 'unassign', 'test-vm2', 'test-vm1:dev1', '--port'],
app=self.app)
Expand All @@ -471,6 +498,9 @@ def test_053_unassign_device_from_port(self):
self.app.expected_calls[
('test-vm2', 'admin.vm.device.testclass.Unassign',
'test-vm1+*:dead:beef:babe:u012345', None)] = b'0\0'
self.app.expected_calls[(
'test-vm2', 'admin.vm.device.testclass.Attached', None, None
)] = b'0\0'
qubesadmin.tools.qvm_device.main(
['testclass', 'unassign', 'test-vm2', 'test-vm1:dev1', '--device'],
app=self.app)
Expand All @@ -491,6 +521,9 @@ def test_055_unassign_explicit_device_port(self):
self.app.expected_calls[
('test-vm2', 'admin.vm.device.testclass.Unassign',
'test-vm1+dev1:*', None)] = b'0\0'
self.app.expected_calls[(
'test-vm2', 'admin.vm.device.testclass.Attached', None, None
)] = b'0\0'
qubesadmin.tools.qvm_device.main(
['testclass', 'unassign', 'test-vm2',
'test-vm1:dev1:dead:beef:babe:u0123456', '--port'], app=self.app)
Expand Down Expand Up @@ -518,6 +551,9 @@ def test_057_unassign_all(self):
self.app.expected_calls[
('test-vm2', 'admin.vm.device.testclass.Unassign',
'test-vm1+dev2:*', None)] = b'0\0'
self.app.expected_calls[(
'test-vm2', 'admin.vm.device.testclass.Attached', None, None
)] = b'0\0'
qubesadmin.tools.qvm_device.main(
['testclass', 'unassign', 'test-vm2'], app=self.app)
self.assertAllCalled()
Expand All @@ -528,6 +564,6 @@ def test_060_device_info(self):
qubesadmin.tools.qvm_device.main(
['testclass', 'info', 'test-vm1:dev1'],
app=self.app)
self.assertIn('dead:beef:babe:u012345: Multimedia: itl test-device',
self.assertIn('Multimedia: itl test-device\ndevice ID: dead:beef:babe:u012345',
buf.getvalue())
self.assertAllCalled()
12 changes: 9 additions & 3 deletions qubesadmin/tools/qvm_device.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -91,8 +91,10 @@ def list_devices(args):
lines = _load_lines(args.app, domains, args.devclass, actual_devices=True)
lines = list(lines.values())
if args.assignments:
# we need to check assignments for all domains since
# selected vm can be mentioned there as backend
extra_lines = _load_lines(
args.app, domains, args.devclass, actual_devices=False)
args.app, [], args.devclass, actual_devices=False)
lines += list(extra_lines.values())
qubesadmin.tools.print_table(prepare_table(lines))

Expand Down Expand Up @@ -157,7 +159,7 @@ def _load_frontends_info(vm, dev, devclass, actual_devices):
yield _frontend_desc(vm, assignment)
else:
for assignment in vm.devices[devclass].get_assigned_devices():
if dev == assignment.virtual_device:
if assignment.matches(dev):
yield _frontend_desc(vm, assignment, virtual=True)
except qubesadmin.exc.QubesVMNotFoundError:
pass
Expand Down Expand Up @@ -281,6 +283,10 @@ def assign_device(args):
device = args.device
if args.only_port:
device = device.clone(device_id="*")
elif device.device_id == UnknownDevice(device.port).device_id:
raise qubesadmin.exc.QubesException(
f"backend vm {device.backend_name} doesn't expose "
f"{device.devclass} device {device.port_id!r}")
if args.only_device:
device = device.clone(
port=Port(device.backend_domain, "*", device.devclass))
Expand Down Expand Up @@ -314,7 +320,7 @@ def _print_attach_hint(assignment, vm):
if dev not in attached and not isinstance(dev, UnknownDevice)]

if ports:
print("Assigned. To attach you can now restart domain or run: \n"
print("Assigned. To attach you can now restart domain or run: \n" +
"\n".join(ports))


Expand Down

0 comments on commit 922bce8

Please sign in to comment.