From d780960381177a347360abb5ebb33e2937d22d2b Mon Sep 17 00:00:00 2001 From: Vitali Graf Date: Sat, 11 Apr 2026 21:21:57 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20EvdevHotkeyListener.stop()=20cancelt=20T?= =?UTF-8?q?asks=20und=20schlie=C3=9Ft=20Devices?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- tests/test_hotkey.py | 33 +++++++++++++++++++++++++++++++++ whisper_local/hotkey/_evdev.py | 22 ++++++++++++++++------ 2 files changed, 49 insertions(+), 6 deletions(-) diff --git a/tests/test_hotkey.py b/tests/test_hotkey.py index b89391c..dbbf214 100644 --- a/tests/test_hotkey.py +++ b/tests/test_hotkey.py @@ -127,3 +127,36 @@ class TestPynputHotkeyListenerStop: await asyncio.sleep(0) # Loop einen Schritt weiter listener.stop() await asyncio.wait_for(listen_task, timeout=1.0) + + +@pytest.mark.skipif(sys.platform != "linux", reason="evdev nur auf Linux") +class TestEvdevListenerStop: + @pytest.mark.asyncio + async def test_stop_cancels_tasks_and_closes_devices(self): + from whisper_local.hotkey._evdev import EvdevHotkeyListener + + listener = EvdevHotkeyListener("KEY_F12") + + async def never_ending(device): + while True: + await asyncio.sleep(1) + + fake_device = MagicMock() + fake_device.close = MagicMock() + + with patch( + "whisper_local.hotkey._evdev.find_keyboard_devices", + return_value=[fake_device], + ): + listener._read_device = never_ending + listen_task = asyncio.create_task(listener.listen()) + await asyncio.sleep(0.05) # Loop-Schritt, damit listen() startet + + assert len(listener._tasks) == 1 + listener.stop() + await asyncio.sleep(0.05) + + assert listener._tasks == [] + fake_device.close.assert_called_once() + + await asyncio.wait_for(listen_task, timeout=1.0) diff --git a/whisper_local/hotkey/_evdev.py b/whisper_local/hotkey/_evdev.py index e057d32..6ad87f8 100644 --- a/whisper_local/hotkey/_evdev.py +++ b/whisper_local/hotkey/_evdev.py @@ -37,6 +37,8 @@ class EvdevHotkeyListener: raise ValueError(f"Unbekannter Key-Name: {key_name}") self.on_press: AsyncCallback | None = None self.on_release: AsyncCallback | None = None + self._tasks: list[asyncio.Task] = [] + self._devices: list[InputDevice] = [] async def _handle_key_event(self, key_down: bool) -> None: """Ruft den passenden Callback auf.""" @@ -57,13 +59,21 @@ class EvdevHotkeyListener: await self._handle_key_event(key_down=False) def stop(self) -> None: - """Stub — evdev-Listener läuft bis zum Prozessende.""" - pass + """Cancelt laufende Read-Tasks und schließt Devices.""" + for task in self._tasks: + task.cancel() + for dev in self._devices: + try: + dev.close() + except Exception: + pass + self._tasks = [] + self._devices = [] async def listen(self) -> None: """Lauscht auf evdev-Events der konfigurierten Taste auf allen passenden Devices.""" - devices = find_keyboard_devices(self.key_name) - for dev in devices: + self._devices = find_keyboard_devices(self.key_name) + for dev in self._devices: logger.info("Lausche auf %s auf %s (%s)", self.key_name, dev.name, dev.path) - tasks = [asyncio.create_task(self._read_device(dev)) for dev in devices] - await asyncio.gather(*tasks) + self._tasks = [asyncio.create_task(self._read_device(dev)) for dev in self._devices] + await asyncio.gather(*self._tasks, return_exceptions=True)