fix: EvdevHotkeyListener.stop() cancelt Tasks und schließt Devices
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -127,3 +127,36 @@ class TestPynputHotkeyListenerStop:
|
|||||||
await asyncio.sleep(0) # Loop einen Schritt weiter
|
await asyncio.sleep(0) # Loop einen Schritt weiter
|
||||||
listener.stop()
|
listener.stop()
|
||||||
await asyncio.wait_for(listen_task, timeout=1.0)
|
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)
|
||||||
|
|||||||
@@ -37,6 +37,8 @@ class EvdevHotkeyListener:
|
|||||||
raise ValueError(f"Unbekannter Key-Name: {key_name}")
|
raise ValueError(f"Unbekannter Key-Name: {key_name}")
|
||||||
self.on_press: AsyncCallback | None = None
|
self.on_press: AsyncCallback | None = None
|
||||||
self.on_release: 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:
|
async def _handle_key_event(self, key_down: bool) -> None:
|
||||||
"""Ruft den passenden Callback auf."""
|
"""Ruft den passenden Callback auf."""
|
||||||
@@ -57,13 +59,21 @@ class EvdevHotkeyListener:
|
|||||||
await self._handle_key_event(key_down=False)
|
await self._handle_key_event(key_down=False)
|
||||||
|
|
||||||
def stop(self) -> None:
|
def stop(self) -> None:
|
||||||
"""Stub — evdev-Listener läuft bis zum Prozessende."""
|
"""Cancelt laufende Read-Tasks und schließt Devices."""
|
||||||
pass
|
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:
|
async def listen(self) -> None:
|
||||||
"""Lauscht auf evdev-Events der konfigurierten Taste auf allen passenden Devices."""
|
"""Lauscht auf evdev-Events der konfigurierten Taste auf allen passenden Devices."""
|
||||||
devices = find_keyboard_devices(self.key_name)
|
self._devices = find_keyboard_devices(self.key_name)
|
||||||
for dev in devices:
|
for dev in self._devices:
|
||||||
logger.info("Lausche auf %s auf %s (%s)", self.key_name, dev.name, dev.path)
|
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]
|
self._tasks = [asyncio.create_task(self._read_device(dev)) for dev in self._devices]
|
||||||
await asyncio.gather(*tasks)
|
await asyncio.gather(*self._tasks, return_exceptions=True)
|
||||||
|
|||||||
Reference in New Issue
Block a user