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

tests: update PCI devices tests for core3 API

Browse Source 
- use asyncio where needed
- attach now takes DeviceAssignment, not DeviceInfo
- PCI ident have ':' replaced with '_'
This commit is contained in:
Marek Marczykowski-Górecki 2018-01-18 16:52:01 +01:00
parent edbfd3843e
commit b245bbca6f
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
1 changed files with 43 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
qubes/tests/integ/devices_pci.py
View File

@ -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())