diff --git a/qubesadmin/tests/tools/qvm_device.py b/qubesadmin/tests/tools/qvm_device.py index 0c5656df..ade142b3 100644 --- a/qubesadmin/tests/tools/qvm_device.py +++ b/qubesadmin/tests/tools/qvm_device.py @@ -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" ) @@ -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( @@ -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): @@ -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): @@ -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() @@ -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() @@ -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() @@ -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() @@ -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', @@ -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): @@ -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) @@ -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) @@ -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) @@ -433,7 +451,7 @@ 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): @@ -441,6 +459,9 @@ def test_050_unassign(self): 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) @@ -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) @@ -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) @@ -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) @@ -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) @@ -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() @@ -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() diff --git a/qubesadmin/tools/qvm_device.py b/qubesadmin/tools/qvm_device.py index f35e114d..5517eed0 100644 --- a/qubesadmin/tools/qvm_device.py +++ b/qubesadmin/tools/qvm_device.py @@ -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)) @@ -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 @@ -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)) @@ -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))