123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175 |
- # pylint: disable=protected-access,pointless-statement
- #
- # The Qubes OS Project, https://www.qubes-os.org/
- #
- # Copyright (C) 2015-2016 Joanna Rutkowska <joanna@invisiblethingslab.com>
- # Copyright (C) 2015-2016 Wojtek Porczyk <woju@invisiblethingslab.com>
- #
- # This program is free software; you can redistribute it and/or modify
- # it under the terms of the GNU General Public License as published by
- # the Free Software Foundation; either version 2 of the License, or
- # (at your option) any later version.
- #
- # This program is distributed in the hope that it will be useful,
- # but WITHOUT ANY WARRANTY; without even the implied warranty of
- # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- # GNU General Public License for more details.
- #
- # You should have received a copy of the GNU General Public License along
- # with this program; if not, write to the Free Software Foundation, Inc.,
- # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
- #
- import os
- import subprocess
- import time
- import unittest
- import qubes.devices
- import qubes.ext.pci
- import qubes.tests
- class TC_00_Devices_PCI(qubes.tests.SystemTestsMixin,
- qubes.tests.QubesTestCase):
- def setUp(self):
- super(TC_00_Devices_PCI, self).setUp()
- if self._testMethodName not in ['test_000_list']:
- pcidev = os.environ.get('QUBES_TEST_PCIDEV', None)
- if pcidev is None:
- self.skipTest('Specify PCI device with QUBES_TEST_PCIDEV '
- 'environment variable')
- 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):
- self.skipTest('Specified device {} does not exists'.format(pcidev))
- self.init_default_template()
- self.vm = self.app.add_new_vm(qubes.vm.appvm.AppVM,
- name=self.make_vm_name('vm'),
- label='red',
- )
- self.vm.create_on_disk()
- self.vm.features['pci-no-strict-reset/' + pcidev] = True
- self.app.save()
- @unittest.expectedFailure
- def test_000_list(self):
- p = subprocess.Popen(['lspci'], stdout=subprocess.PIPE)
- # get a dict: BDF -> description
- actual_devices = dict(
- l.split(' (')[0].split(' ', 1)
- for l in p.communicate()[0].decode().splitlines())
- for dev in self.app.domains[0].devices['pci']:
- 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)
- if actual_devices:
- self.fail('Not all devices listed, missing: {}'.format(
- actual_devices))
- def assertDeviceNotInCollection(self, dev, dev_col):
- self.assertNotIn(dev, dev_col.attached())
- self.assertNotIn(dev, dev_col.persistent())
- self.assertNotIn(dev, dev_col.assignments())
- self.assertNotIn(dev, dev_col.assignments(persistent=True))
- 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.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))
- self.vm.start()
- self.assertIn(self.dev, dev_col.attached())
- p = self.vm.run('lspci', passio_popen=True)
- (stdout, _) = p.communicate()
- self.assertIn(self.dev.description, stdout.decode())
- 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.exc.QubesVMNotRunningError):
- dev_col.attach(self.assignment)
- def test_020_attach_online_persistent(self):
- self.vm.start()
- dev_col = self.vm.devices['pci']
- self.assertDeviceNotInCollection(self.dev, dev_col)
- dev_col.attach(self.assignment)
- self.assertIn(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))
- # give VM kernel some time to discover new device
- time.sleep(1)
- p = self.vm.run('lspci', passio_popen=True)
- (stdout, _) = p.communicate()
- 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.app.save()
- self.vm.start()
- with self.assertRaises(qubes.exc.QubesVMNotHaltedError):
- 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))
- # give VM kernel some time to discover new device
- time.sleep(1)
- p = self.vm.run('lspci', passio_popen=True)
- (stdout, _) = p.communicate()
- self.assertIn(self.dev.description, stdout.decode())
- dev_col.detach(self.assignment)
- self.assertDeviceNotInCollection(self.dev, dev_col)
- p = self.vm.run('lspci', passio_popen=True)
- (stdout, _) = p.communicate()
- self.assertNotIn(self.dev.description, stdout.decode())
|