diff --git a/qubesusbproxy/core3ext.py b/qubesusbproxy/core3ext.py index b3d5fff..e469751 100644 --- a/qubesusbproxy/core3ext.py +++ b/qubesusbproxy/core3ext.py @@ -787,7 +787,11 @@ async def on_domain_start(self, vm, _event, **_kwargs): async def on_domain_shutdown(self, vm, _event, **_kwargs): # pylint: disable=unused-argument vm.fire_event("device-list-change:usb") + # devices exposed by the shutting-down backend vm: + # notify that they are gone and detach them from frontend vms utils.device_list_change(self, {}, vm, None, USBDevice) + # devices attached to shutting-down frontend vm: notify detach + utils.detach_attached_devices_on_shutdown(self, vm, USBDevice) if vm.uuid in self.autoattach_locks: del self.autoattach_locks[vm.uuid] diff --git a/qubesusbproxy/tests.py b/qubesusbproxy/tests.py index b6027a5..712e283 100644 --- a/qubesusbproxy/tests.py +++ b/qubesusbproxy/tests.py @@ -1010,6 +1010,27 @@ def test_023_on_startup_already_attached(self): loop.run_until_complete(self.ext.on_domain_start(front, None)) attach_and_notify.assert_not_called() + def test_030_on_domain_shutdown_frontend(self): + # a frontend that has a usb device attached is shutting down; + # the detach event is expected + back, front = self.added_assign_setup() + + exp_dev = qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb")) + self.ext.devices_cache = {"sys-usb": {"1-1": front}, "front-vm": {}} + + loop = asyncio.get_event_loop() + with mock.patch("asyncio.ensure_future"): + loop.run_until_complete(self.ext.on_domain_shutdown(front, None)) + + front.fire_event_async.assert_called_with( + "device-detach:usb", port=exp_dev.port + ) + self.assertEqual( + front.fire_event_async.call_args.kwargs["port"].backend_domain, + back, + ) + self.assertIsNone(self.ext.devices_cache["sys-usb"]["1-1"]) + def list_tests(): tests = [TC_00_USBProxy]