diff --git a/qubes/tests/integ/devices_pci.py b/qubes/tests/integ/devices_pci.py index 484202a1..d0344134 100644 --- a/qubes/tests/integ/devices_pci.py +++ b/qubes/tests/integ/devices_pci.py @@ -37,7 +37,10 @@ class TC_00_Devices_PCI(qubes.tests.SystemTestCase): if self._testMethodName not in ['test_000_list']: pcidev = os.environ['QUBES_TEST_PCIDEV'] self.dev = self.app.domains[0].devices['pci'][pcidev] - self.assignment = qubes.devices.DeviceAssignment(backend_domain=self.dev.backend_domain, ident=self.dev.ident, persistent=True) + self.assignment = qubes.devices.DeviceAssignment( + backend_domain=self.dev.backend_domain, + ident=self.dev.ident, + persistent=True) if isinstance(self.dev, qubes.devices.UnknownDevice): self.skipTest('Specified device {} does not exists'.format(pcidev)) self.init_default_template() @@ -45,7 +48,8 @@ class TC_00_Devices_PCI(qubes.tests.SystemTestCase): name=self.make_vm_name('vm'), label='red', ) - self.vm.create_on_disk() + self.loop.run_until_complete( + self.vm.create_on_disk()) self.vm.features['pci-no-strict-reset/' + pcidev] = True self.app.save() @@ -57,11 +61,12 @@ class TC_00_Devices_PCI(qubes.tests.SystemTestCase): l.split(' (')[0].split(' ', 1) for l in p.communicate()[0].decode().splitlines()) for dev in self.app.domains[0].devices['pci']: + lspci_ident = dev.ident.replace('_', ':') self.assertIsInstance(dev, qubes.ext.pci.PCIDevice) self.assertEqual(dev.backend_domain, self.app.domains[0]) - self.assertIn(dev.ident, actual_devices) - self.assertEqual(dev.description, actual_devices[dev.ident]) - actual_devices.pop(dev.ident) + self.assertIn(lspci_ident, actual_devices) + self.assertEqual(dev.description, actual_devices[lspci_ident]) + actual_devices.pop(lspci_ident) if actual_devices: self.fail('Not all devices listed, missing: {}'.format( @@ -76,7 +81,8 @@ class TC_00_Devices_PCI(qubes.tests.SystemTestCase): def test_010_attach_offline_persistent(self): dev_col = self.vm.devices['pci'] self.assertDeviceNotInCollection(self.dev, dev_col) - dev_col.attach(self.assignment) + self.loop.run_until_complete( + dev_col.attach(self.assignment)) self.app.save() self.assertNotIn(self.dev, dev_col.attached()) self.assertIn(self.dev, dev_col.persistent()) @@ -84,12 +90,11 @@ class TC_00_Devices_PCI(qubes.tests.SystemTestCase): self.assertIn(self.dev, dev_col.assignments(persistent=True)) self.assertNotIn(self.dev, dev_col.assignments(persistent=False)) - - self.vm.start() + self.loop.run_until_complete(self.vm.start()) self.assertIn(self.dev, dev_col.attached()) - p = self.vm.run('lspci', passio_popen=True) - (stdout, _) = p.communicate() + (stdout, _) = self.loop.run_until_complete( + self.vm.run_for_stdio('lspci')) self.assertIn(self.dev.description, stdout.decode()) @@ -98,14 +103,17 @@ class TC_00_Devices_PCI(qubes.tests.SystemTestCase): self.assertDeviceNotInCollection(self.dev, dev_col) self.assignment.persistent = False with self.assertRaises(qubes.exc.QubesVMNotRunningError): - dev_col.attach(self.assignment) + self.loop.run_until_complete( + dev_col.attach(self.assignment)) def test_020_attach_online_persistent(self): - self.vm.start() + self.loop.run_until_complete( + self.vm.start()) dev_col = self.vm.devices['pci'] self.assertDeviceNotInCollection(self.dev, dev_col) - dev_col.attach(self.assignment) + self.loop.run_until_complete( + dev_col.attach(self.assignment)) self.assertIn(self.dev, dev_col.attached()) self.assertIn(self.dev, dev_col.persistent()) @@ -115,39 +123,46 @@ class TC_00_Devices_PCI(qubes.tests.SystemTestCase): # give VM kernel some time to discover new device time.sleep(1) - p = self.vm.run('lspci', passio_popen=True) - (stdout, _) = p.communicate() + (stdout, _) = self.loop.run_until_complete( + self.vm.run_for_stdio('lspci')) self.assertIn(self.dev.description, stdout.decode()) def test_021_persist_detach_online_fail(self): dev_col = self.vm.devices['pci'] self.assertDeviceNotInCollection(self.dev, dev_col) - dev_col.attach(self.assignment) + self.loop.run_until_complete( + dev_col.attach(self.assignment)) self.app.save() - self.vm.start() + self.loop.run_until_complete( + self.vm.start()) with self.assertRaises(qubes.exc.QubesVMNotHaltedError): - self.vm.devices['pci'].detach(self.assignment) + self.loop.run_until_complete( + self.vm.devices['pci'].detach(self.assignment)) def test_030_persist_attach_detach_offline(self): dev_col = self.vm.devices['pci'] self.assertDeviceNotInCollection(self.dev, dev_col) - dev_col.attach(self.assignment) + self.loop.run_until_complete( + dev_col.attach(self.assignment)) self.app.save() self.assertNotIn(self.dev, dev_col.attached()) self.assertIn(self.dev, dev_col.persistent()) self.assertIn(self.dev, dev_col.assignments()) self.assertIn(self.dev, dev_col.assignments(persistent=True)) self.assertNotIn(self.dev, dev_col.assignments(persistent=False)) - dev_col.detach(self.assignment) + self.loop.run_until_complete( + dev_col.detach(self.assignment)) self.assertDeviceNotInCollection(self.dev, dev_col) def test_031_attach_detach_online_temp(self): dev_col = self.vm.devices['pci'] - self.vm.start() + self.loop.run_until_complete( + self.vm.start()) self.assignment.persistent = False self.assertDeviceNotInCollection(self.dev, dev_col) - dev_col.attach(self.assignment) + self.loop.run_until_complete( + dev_col.attach(self.assignment)) self.assertIn(self.dev, dev_col.attached()) self.assertNotIn(self.dev, dev_col.persistent()) @@ -159,13 +174,14 @@ class TC_00_Devices_PCI(qubes.tests.SystemTestCase): # give VM kernel some time to discover new device time.sleep(1) - p = self.vm.run('lspci', passio_popen=True) - (stdout, _) = p.communicate() + (stdout, _) = self.loop.run_until_complete( + self.vm.run_for_stdio('lspci')) self.assertIn(self.dev.description, stdout.decode()) - dev_col.detach(self.assignment) + self.loop.run_until_complete( + dev_col.detach(self.assignment)) self.assertDeviceNotInCollection(self.dev, dev_col) - p = self.vm.run('lspci', passio_popen=True) - (stdout, _) = p.communicate() + (stdout, _) = self.loop.run_until_complete( + self.vm.run_for_stdio('lspci')) self.assertNotIn(self.dev.description, stdout.decode())