Fix pci device integration tests

Signed-off-by: Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>
This commit is contained in:
Bahtiar `kalkin-` Gadimov 2017-04-07 22:17:09 +02:00
parent 0f78d257d4
commit e84114d3a9
No known key found for this signature in database
GPG Key ID: 07799AE179ED4FD4

View File

@ -41,6 +41,7 @@ class TC_00_Devices_PCI(qubes.tests.SystemTestsMixin,
self.skipTest('Specify PCI device with QUBES_TEST_PCIDEV ' self.skipTest('Specify PCI device with QUBES_TEST_PCIDEV '
'environment variable') 'environment variable')
self.dev = self.app.domains[0].devices['pci'][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)
if isinstance(self.dev, qubes.devices.UnknownDevice): if isinstance(self.dev, qubes.devices.UnknownDevice):
self.skipTest('Specified device {} does not exists'.format(pcidev)) self.skipTest('Specified device {} does not exists'.format(pcidev))
self.init_default_template() self.init_default_template()
@ -64,110 +65,111 @@ class TC_00_Devices_PCI(qubes.tests.SystemTestsMixin,
self.assertEqual(dev.backend_domain, self.app.domains[0]) self.assertEqual(dev.backend_domain, self.app.domains[0])
self.assertIn(dev.ident, actual_devices) self.assertIn(dev.ident, actual_devices)
self.assertEqual(dev.description, actual_devices[dev.ident]) self.assertEqual(dev.description, actual_devices[dev.ident])
self.assertIsInstance(dev.frontend_domain,
(qubes.vm.BaseVM, None.__class__))
actual_devices.pop(dev.ident) actual_devices.pop(dev.ident)
if actual_devices: if actual_devices:
self.fail('Not all devices listed, missing: {}'.format( self.fail('Not all devices listed, missing: {}'.format(
actual_devices)) actual_devices))
def test_010_attach_offline(self): def assertDeviceNotInCollection(self, dev, dev_col):
self.assertIsNone(self.dev.frontend_domain) self.assertNotIn(dev, dev_col.attached())
self.assertNotIn(self.dev, self.vm.devices['pci'].attached()) self.assertNotIn(dev, dev_col.persistent())
self.assertNotIn(self.dev, self.vm.devices['pci'].attached( self.assertNotIn(dev, dev_col.assignments())
persistent=True)) self.assertNotIn(dev, dev_col.assignments(persistent=True))
self.assertNotIn(self.dev, self.vm.devices['pci'].attached(
persistent=False))
self.vm.devices['pci'].attach(self.dev) 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.app.save() 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))
# still should be None, as domain is not started yet
self.assertIsNone(self.dev.frontend_domain)
self.assertIn(self.dev, self.vm.devices['pci'].attached())
self.assertIn(self.dev, self.vm.devices['pci'].attached(
persistent=True))
self.assertNotIn(self.dev, self.vm.devices['pci'].attached(
persistent=False))
self.vm.start() self.vm.start()
self.assertEqual(self.dev.frontend_domain, self.vm) self.assertIn(self.dev, dev_col.attached())
self.assertIn(self.dev, self.vm.devices['pci'].attached())
self.assertIn(self.dev, self.vm.devices['pci'].attached(
persistent=True))
self.assertNotIn(self.dev, self.vm.devices['pci'].attached(
persistent=False))
p = self.vm.run('lspci', passio_popen=True) p = self.vm.run('lspci', passio_popen=True)
(stdout, _) = p.communicate() (stdout, _) = p.communicate()
self.assertIn(self.dev.description, stdout) self.assertIn(self.dev.description, stdout.decode())
def test_011_attach_online(self):
def test_011_attach_offline_temp_fail(self):
dev_col = self.vm.devices['pci']
self.assertDeviceNotInCollection(self.dev, dev_col)
self.assignment.persistent = False
with self.assertRaises(qubes.devices.WrongAssignment):
dev_col.attach(self.assignment)
def test_020_attach_online_persistent(self):
self.vm.start() self.vm.start()
self.vm.devices['pci'].attach(self.dev) dev_col = self.vm.devices['pci']
self.assertDeviceNotInCollection(self.dev, dev_col)
dev_col.attach(self.assignment)
self.assertEqual(self.dev.frontend_domain, self.vm) self.assertIn(self.dev, dev_col.attached())
self.assertIn(self.dev, self.vm.devices['pci'].attached()) self.assertIn(self.dev, dev_col.persistent())
self.assertIn(self.dev, self.vm.devices['pci'].attached( self.assertIn(self.dev, dev_col.assignments())
persistent=True)) self.assertIn(self.dev, dev_col.assignments(persistent=True))
self.assertNotIn(self.dev, self.vm.devices['pci'].attached( self.assertNotIn(self.dev, dev_col.assignments(persistent=False))
persistent=False))
# give VM kernel some time to discover new device # give VM kernel some time to discover new device
time.sleep(1) time.sleep(1)
p = self.vm.run('lspci', passio_popen=True) p = self.vm.run('lspci', passio_popen=True)
(stdout, _) = p.communicate() (stdout, _) = p.communicate()
self.assertIn(self.dev.description, stdout) self.assertIn(self.dev.description, stdout.decode())
def test_012_attach_online_temp(self):
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.app.save()
self.vm.start() self.vm.start()
self.vm.devices['pci'].attach(self.dev, persistent=False) with self.assertRaises(qubes.devices.WrongAssignment):
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.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.assertDeviceNotInCollection(self.dev, dev_col)
def test_031_attach_detach_online_temp(self):
dev_col = self.vm.devices['pci']
self.vm.start()
self.assignment.persistent = False
self.assertDeviceNotInCollection(self.dev, dev_col)
dev_col.attach(self.assignment)
self.assertIn(self.dev, dev_col.attached())
self.assertNotIn(self.dev, dev_col.persistent())
self.assertIn(self.dev, dev_col.assignments())
self.assertIn(self.dev, dev_col.assignments(persistent=False))
self.assertNotIn(self.dev, dev_col.assignments(persistent=True))
self.assertIn(self.dev, dev_col.assignments(persistent=False))
self.assertEqual(self.dev.frontend_domain, self.vm)
self.assertIn(self.dev, self.vm.devices['pci'].attached())
self.assertNotIn(self.dev, self.vm.devices['pci'].attached(
persistent=True))
self.assertIn(self.dev, self.vm.devices['pci'].attached(
persistent=False))
# give VM kernel some time to discover new device # give VM kernel some time to discover new device
time.sleep(1) time.sleep(1)
p = self.vm.run('lspci', passio_popen=True) p = self.vm.run('lspci', passio_popen=True)
(stdout, _) = p.communicate() (stdout, _) = p.communicate()
self.assertIn(self.dev.description, stdout)
def test_020_detach_online(self): self.assertIn(self.dev.description, stdout.decode())
self.vm.devices['pci'].attach(self.dev) dev_col.detach(self.assignment)
self.app.save() self.assertDeviceNotInCollection(self.dev, dev_col)
self.vm.start()
self.assertIn(self.dev, self.vm.devices['pci'].attached())
self.assertIn(self.dev, self.vm.devices['pci'].attached(
persistent=True))
self.assertNotIn(self.dev, self.vm.devices['pci'].attached(
persistent=False))
self.assertEqual(self.dev.frontend_domain, self.vm)
self.vm.devices['pci'].detach(self.dev)
self.assertIsNone(self.dev.frontend_domain)
self.assertNotIn(self.dev, self.vm.devices['pci'].attached())
self.assertNotIn(self.dev, self.vm.devices['pci'].attached(
persistent=True))
self.assertNotIn(self.dev, self.vm.devices['pci'].attached(
persistent=False))
p = self.vm.run('lspci', passio_popen=True) p = self.vm.run('lspci', passio_popen=True)
(stdout, _) = p.communicate() (stdout, _) = p.communicate()
self.assertNotIn(self.dev.description, stdout) self.assertNotIn(self.dev.description, stdout.decode())
# can't do this right now because of kernel bug - it cause the whole
# PCI bus being deregistered, which emit some warning in sysfs
# handling code (removing non-existing "0000:00" group)
#
# p = self.vm.run('dmesg', passio_popen=True)
# (stdout, _) = p.communicate()
# # check for potential oops
# self.assertNotIn('end trace', stdout)