diff --git a/qubes/devices.py b/qubes/devices.py index a0838ff0..91f38fc0 100644 --- a/qubes/devices.py +++ b/qubes/devices.py @@ -270,6 +270,27 @@ class DeviceCollection(object): device_assignment.bus = self._bus self._set.add(device_assignment) + def update_persistent(self, device: DeviceInfo, persistent: bool): + '''Update `persistent` flag of already attached device. + ''' + + if self._vm.is_halted(): + raise qubes.exc.QubesVMNotStartedError(self._vm, + 'VM must be running to modify device persistence flag') + assignments = [a for a in self.assignments() if a.device == device] + if not assignments: + raise qubes.exc.QubesValueError('Device not assigned') + assert len(assignments) == 1 + assignment = assignments[0] + + # be careful to use already present assignment, not the provided one + # - to not change options as a side effect + if persistent and device not in self._set: + assignment.persistent = True + self._set.add(assignment) + elif not persistent and device in self._set: + self._set.discard(assignment) + @asyncio.coroutine def detach(self, device_assignment: DeviceAssignment): '''Detach (remove) device from domain. diff --git a/qubes/tests/devices.py b/qubes/tests/devices.py index 3433eba2..c436ea3e 100644 --- a/qubes/tests/devices.py +++ b/qubes/tests/devices.py @@ -71,6 +71,9 @@ class TestVM(qubes.tests.TestEmitter): def is_halted(self): return not self.running + def is_running(self): + return self.running + class TC_00_DeviceCollection(qubes.tests.QubesTestCase): @@ -130,8 +133,6 @@ class TC_00_DeviceCollection(qubes.tests.QubesTestCase): self.loop.run_until_complete(self.collection.attach(self.assignment)) self.assertEventFired(self.emitter, 'device-list-attached:testclass') self.assertEqual({self.device}, set(self.collection.persistent())) - self.assertEqual({self.device}, - set(self.collection.persistent())) self.assertEqual(set([]), set(self.collection.attached())) @@ -153,6 +154,52 @@ class TC_00_DeviceCollection(qubes.tests.QubesTestCase): self.assertEqual({self.device}, set(self.collection)) self.assertEventFired(self.emitter, 'device-list:testclass') + def test_020_update_persistent_to_false(self): + self.emitter.running = True + self.assertEqual(set([]), set(self.collection.persistent())) + self.loop.run_until_complete(self.collection.attach(self.assignment)) + # device-attach event not implemented, so manipulate object manually + self.device.frontend_domain = self.emitter + self.assertEqual({self.device}, set(self.collection.persistent())) + self.assertEqual({self.device}, set(self.collection.attached())) + self.assertEqual({self.device}, set(self.collection.persistent())) + self.assertEqual({self.device}, set(self.collection.attached())) + self.collection.update_persistent(self.device, False) + self.assertEqual(set(), set(self.collection.persistent())) + self.assertEqual({self.device}, set(self.collection.attached())) + + def test_021_update_persistent_to_true(self): + self.assignment.persistent = False + self.emitter.running = True + self.assertEqual(set([]), set(self.collection.persistent())) + self.loop.run_until_complete(self.collection.attach(self.assignment)) + # device-attach event not implemented, so manipulate object manually + self.device.frontend_domain = self.emitter + self.assertEqual(set(), set(self.collection.persistent())) + self.assertEqual({self.device}, set(self.collection.attached())) + self.assertEqual(set(), set(self.collection.persistent())) + self.assertEqual({self.device}, set(self.collection.attached())) + self.collection.update_persistent(self.device, True) + self.assertEqual({self.device}, set(self.collection.persistent())) + self.assertEqual({self.device}, set(self.collection.attached())) + + def test_022_update_persistent_reject_not_running(self): + self.assertEqual(set([]), set(self.collection.persistent())) + self.loop.run_until_complete(self.collection.attach(self.assignment)) + self.assertEqual({self.device}, set(self.collection.persistent())) + self.assertEqual(set(), set(self.collection.attached())) + with self.assertRaises(qubes.exc.QubesVMNotStartedError): + self.collection.update_persistent(self.device, False) + + def test_023_update_persistent_reject_not_attached(self): + self.assertEqual(set(), set(self.collection.persistent())) + self.assertEqual(set(), set(self.collection.attached())) + self.emitter.running = True + with self.assertRaises(qubes.exc.QubesValueError): + self.collection.update_persistent(self.device, True) + with self.assertRaises(qubes.exc.QubesValueError): + self.collection.update_persistent(self.device, False) + class TC_01_DeviceManager(qubes.tests.QubesTestCase): def setUp(self):