diff --git a/src/vocalinux/common_types.py b/src/vocalinux/common_types.py index db4d1f1..1bdfc4b 100644 --- a/src/vocalinux/common_types.py +++ b/src/vocalinux/common_types.py @@ -4,7 +4,7 @@ This module provides type definitions to avoid circular imports. """ from enum import Enum, auto -from typing import Callable, Optional, Protocol # noqa: F401 +from typing import Any, Callable, Iterable, Optional, Protocol # noqa: F401 class RecognitionState(Enum): @@ -79,3 +79,23 @@ class TextInjectorProtocol(Protocol): def inject_text(self, text: str) -> bool: """Inject text into the active application.""" ... + + +class _EvdevCaptureDevice(Protocol): + """Minimal evdev InputDevice surface used by the shortcut recorder.""" + + def fileno(self) -> int: + """Return the device file descriptor for GLib.io_add_watch.""" + ... + + def read(self) -> Iterable[Any]: + """Return pending input events.""" + ... + + def close(self) -> None: + """Close the device file.""" + ... + + def active_keys(self) -> Iterable[int]: + """Return evdev codes currently down on this device.""" + ... diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index c0afd4e..fb8e145 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -22,7 +22,7 @@ import os import re import threading import time -from typing import TYPE_CHECKING, Any, NamedTuple, Optional +from typing import TYPE_CHECKING, Any, Callable, NamedTuple, Optional import gi @@ -31,7 +31,7 @@ gi.require_version("Gdk", "3.0") # Need GLib for idle_add from gi.repository import Gdk, GLib, GObject, Gtk, Pango # noqa: E402 -from ..common_types import RecognitionState # noqa: E402 +from ..common_types import RecognitionState, _EvdevCaptureDevice # noqa: E402 from ..speech_recognition.silero_vad import is_silero_available # noqa: E402 from ..utils import parakeet_model_info as parakeet # noqa: E402 from ..utils.faster_whisper_model_info import ( @@ -114,6 +114,7 @@ from .keyboard_backends import ( # noqa: E402 is_valid_shortcut, parse_shortcut_spec, ) +from .keyboard_backends.evdev_backend import MODIFIER_KEY_CODES # noqa: E402 # Avoid circular imports for type checking if TYPE_CHECKING: @@ -1644,6 +1645,42 @@ def _gdk_keyname_to_token(name: Optional[str]) -> Optional[str]: return None +def function_token_from_gdk_hardware_keycode(hardware_keycode: int) -> Optional[str]: + """Map XKB/GDK hardware keycodes 191–202 (FK13–FK24) to f13–f24. + + Default evdev maps those keys to XF86* keysyms or NoSymbol; the XKB + keycode is the unambiguous signal. F1–F12 use other keycodes and already + arrive as F1–F12 keysyms. + """ + try: + code = int(hardware_keycode) + except (TypeError, ValueError): + return None + if 191 <= code <= 202: + return f"f{code - 178}" + return None + + +def function_token_from_evdev_code(code: int) -> Optional[str]: + """Map evdev KEY_F13–KEY_F24 (183–194) to f13–f24.""" + try: + key = int(code) + except (TypeError, ValueError): + return None + if 183 <= key <= 194: + return f"f{key - 170}" + return None + + +def modifiers_from_active_evdev_codes(active_codes: set[int]) -> list[str]: + """Return ctrl/alt/shift/super currently held in *active_codes*, in that order.""" + held: list[str] = [] + for name in ("ctrl", "alt", "shift", "super"): + if active_codes & MODIFIER_KEY_CODES[name]: + held.append(name) + return held + + def _shortcut_from_capture(modifiers: list[str], token: Optional[str]) -> Optional[str]: """Build a canonical shortcut from recorded modifiers and a main key.""" if token is None: @@ -1652,6 +1689,255 @@ def _shortcut_from_capture(modifiers: list[str], token: Optional[str]) -> Option return candidate if is_valid_shortcut(candidate) else None +def _gdk_capture_to_shortcut( + modifiers: list[str], + keyval_name: Optional[str], + hardware_keycode: int = 0, +) -> Optional[str]: + """Build a shortcut from recorder inputs, preferring FK13–FK24 keycodes.""" + token = function_token_from_gdk_hardware_keycode(hardware_keycode) + if token is None: + token = _gdk_keyname_to_token(keyval_name) + return _shortcut_from_capture(modifiers, token) + + +# linux/input.h EV_KEY; python-evdev uses the same value. +_EVDEV_EV_KEY = 1 + + +class _EvdevShortcutRecorder: + """Read-only F13–F24 capture from keyboard evdev nodes (no grab, no thread). + + GTK 3 on Wayland drops XKB NoSymbol keys before they become GdkEvents, so + FK19/FK24 never reach the dialog key-press handler on the default map. + Unions modifiers across watched keyboards, and drops a failed device while + still scanning until recording stops. + """ + + def __init__(self, on_shortcut: Callable[[str], None]) -> None: + self._on_shortcut = on_shortcut + self._active = False + self._devices: dict[str, _EvdevCaptureDevice] = {} + self._watch_ids: dict[str, Any] = {} + self._held: dict[str, set[int]] = {} + self._scan_timeout_id: Optional[Any] = None + self._input_device_cls: Any = None + self._find_keyboard_devices: Optional[Callable[[], list[str]]] = None + + def start(self) -> None: + try: + from .keyboard_backends.evdev_backend import ( + DEVICE_RESCAN_SECONDS, + EVDEV_AVAILABLE, + find_keyboard_devices, + ) + except ImportError: + logger.debug("evdev unavailable for shortcut recording; using GDK events only") + return + if not EVDEV_AVAILABLE: + logger.debug("evdev unavailable for shortcut recording; using GDK events only") + return + try: + from evdev import InputDevice + except ImportError: + logger.debug("evdev unavailable for shortcut recording; using GDK events only") + return + + self._active = True + self._input_device_cls = InputDevice + self._find_keyboard_devices = find_keyboard_devices + + paths: list[str] = [] + listed = False + try: + paths = list(find_keyboard_devices()) + listed = True + except Exception as e: + logger.warning("Cannot list keyboard devices for shortcut recording: %s", e) + + open_error: Optional[tuple[str, BaseException]] = None + for path in paths: + err = self._attach_device(path) + if err is not None and open_error is None: + open_error = (path, err) + + if not self._devices: + if open_error is not None: + logger.warning( + "Cannot open keyboard devices for shortcut recording " + "(%s: %s); F-keys without keysyms will not be captured", + open_error[0], + open_error[1], + ) + elif listed and not paths: + logger.debug("No keyboard devices found for shortcut recording") + elif listed: + logger.debug("No keyboard devices opened for shortcut recording") + + if self._active: + self._scan_timeout_id = GLib.timeout_add( + int(DEVICE_RESCAN_SECONDS * 1000), + self._on_device_rescan, + ) + + def stop(self) -> None: + self._active = False + timeout_id = self._scan_timeout_id + self._scan_timeout_id = None + if timeout_id is not None: + try: + GLib.source_remove(timeout_id) + except Exception: + pass + for watch_id in list(self._watch_ids.values()): + try: + GLib.source_remove(watch_id) + except Exception: + pass + self._watch_ids.clear() + for device in list(self._devices.values()): + self._close_device(device) + self._devices.clear() + self._held.clear() + + def _close_device(self, device: _EvdevCaptureDevice) -> None: + try: + device.close() + except Exception: + pass + + def _attach_device(self, path: str, *, hotplug: bool = False) -> Optional[BaseException]: + """Open and watch *path* if recording is active and it is not already watched. + + Returns the InputDevice open error, or None if attached/skipped. + """ + if not self._active or path in self._devices or self._input_device_cls is None: + return None + try: + device = self._input_device_cls(path) + except (OSError, IOError) as e: + return e + if not self._active: + self._close_device(device) + return None + # Do not grab: compositor and the runtime backend must still see keys. + try: + held = set(device.active_keys()) + except Exception: + held = set() + try: + watch_id = GLib.io_add_watch( + device.fileno(), + GLib.IO_IN | GLib.IO_ERR | GLib.IO_HUP, + self._make_io_callback(device), + ) + except Exception as e: + logger.debug("GLib.io_add_watch failed for %s: %s", path, e) + self._close_device(device) + return None + if not self._active: + try: + GLib.source_remove(watch_id) + except Exception: + pass + self._close_device(device) + return None + self._devices[path] = device + self._watch_ids[path] = watch_id + self._held[path] = held + if hotplug: + logger.debug("Added keyboard for shortcut recording: %s", path) + return None + + def _drop_device(self, path: str) -> None: + """Close and forget *path* without source_remove of its in-flight watch.""" + self._watch_ids.pop(path, None) + device = self._devices.pop(path, None) + self._held.pop(path, None) + if device is not None: + self._close_device(device) + logger.debug("Dropped keyboard from shortcut recording: %s", path) + + def _union_held_codes(self) -> set[int]: + union: set[int] = set() + for codes in self._held.values(): + union.update(codes) + return union + + def _path_for_device(self, device: _EvdevCaptureDevice) -> Optional[str]: + for path, tracked in self._devices.items(): + if tracked is device: + return path + return None + + def _on_device_rescan(self) -> bool: + if not self._active: + return False + finder = self._find_keyboard_devices + if finder is None: + return False + try: + paths = list(finder()) + except Exception: + return True + for path in paths: + if not self._active: + return False + self._attach_device(path, hotplug=True) + return self._active + + def _make_io_callback(self, device: _EvdevCaptureDevice) -> Callable[[int, int], bool]: + def _on_io(fd: int, condition: int) -> bool: + return self._on_fd(device, condition) + + return _on_io + + def _on_fd(self, device: _EvdevCaptureDevice, condition: int) -> bool: + if not self._active: + return False + path = self._path_for_device(device) + if path is None: + return False + try: + if condition & (GLib.IO_ERR | GLib.IO_HUP): + self._drop_device(path) + return False + except Exception: + pass + held = self._held.setdefault(path, set()) + try: + events = list(device.read()) + except BlockingIOError: + return True + except (OSError, IOError): + self._drop_device(path) + return False + + for event in events: + if getattr(event, "type", None) != _EVDEV_EV_KEY: + continue + code = event.code + value = event.value + if value == 1: + held.add(code) + token = function_token_from_evdev_code(code) + if token is None: + continue + shortcut = _shortcut_from_capture( + modifiers_from_active_evdev_codes(self._union_held_codes()), + token, + ) + if shortcut: + # Pop before on_shortcut: the dialog stop() must not + # source_remove this in-flight watch (returning False does). + self._watch_ids.pop(path, None) + self._on_shortcut(shortcut) + return False + elif value == 0: + held.discard(code) + return True + + def _row_matches_query(query: str, title: str, subtitle: str = "", keywords=()) -> bool: """Return whether a settings row matches a search query. @@ -3667,7 +3953,9 @@ class SettingsDialog(Gtk.Dialog): # Key-capture state for the Record button. self._recording_shortcut = False + self._evdev_shortcut_recorder = None self.connect("key-press-event", self._on_shortcut_key_press) + self.connect("destroy", self._on_shortcut_recorder_destroy) self.shortcuts_tab.pack_start(group, False, False, 0) @@ -3816,11 +4104,50 @@ class SettingsDialog(Gtk.Dialog): self.shortcut_info_label.set_markup( "Press a modifier + key (e.g. Alt+R), or an F-key. Press Esc to cancel." ) + self._start_evdev_shortcut_recorder() def _stop_recording_shortcut(self): """Exit key-capture mode and restore the Record button.""" self._recording_shortcut = False - self.record_shortcut_button.set_label("Record") + self._stop_evdev_shortcut_recorder() + if getattr(self, "record_shortcut_button", None) is not None: + try: + self.record_shortcut_button.set_label("Record") + except Exception: + pass + + def _start_evdev_shortcut_recorder(self) -> None: + """Listen on keyboard evdev nodes for F13–F24 while recording.""" + self._stop_evdev_shortcut_recorder() + recorder = _EvdevShortcutRecorder(self._on_evdev_recorded_shortcut) + recorder.start() + self._evdev_shortcut_recorder = recorder + + def _stop_evdev_shortcut_recorder(self) -> None: + recorder = getattr(self, "_evdev_shortcut_recorder", None) + if recorder is not None: + recorder.stop() + self._evdev_shortcut_recorder = None + + def _on_shortcut_recorder_destroy(self, widget) -> None: + """Drop evdev watches if the dialog closes while recording.""" + self._recording_shortcut = False + self._stop_evdev_shortcut_recorder() + + def _commit_recorded_shortcut(self, shortcut: Optional[str]) -> bool: + """Apply a captured shortcut if still recording. Returns True if consumed.""" + if not getattr(self, "_recording_shortcut", False): + return False + if not shortcut or not is_valid_shortcut(shortcut): + return False + self.custom_shortcut_entry.set_text(shortcut) + self._stop_recording_shortcut() + self._apply_custom_shortcut(shortcut) + return True + + def _on_evdev_recorded_shortcut(self, shortcut: str) -> None: + """Apply an F13–F24 capture from the evdev watcher (main loop).""" + self._commit_recorded_shortcut(shortcut) def _gdk_event_to_shortcut(self, event) -> Optional[str]: """Build a canonical shortcut string from a GDK key-press event.""" @@ -3834,8 +4161,8 @@ class SettingsDialog(Gtk.Dialog): modifiers.append("shift") if state & Gdk.ModifierType.SUPER_MASK: modifiers.append("super") - token = _gdk_keyname_to_token(Gdk.keyval_name(event.keyval)) - return _shortcut_from_capture(modifiers, token) + hardware = getattr(event, "hardware_keycode", 0) or 0 + return _gdk_capture_to_shortcut(modifiers, Gdk.keyval_name(event.keyval), hardware) def _on_shortcut_key_press(self, widget, event): """Capture a pressed combo while recording; otherwise pass through.""" @@ -3847,20 +4174,19 @@ class SettingsDialog(Gtk.Dialog): return True # wait for the non-modifier key shortcut = self._gdk_event_to_shortcut(event) + if self._commit_recorded_shortcut(shortcut): + return True + if not getattr(self, "_recording_shortcut", False): + return True if keyname == "Escape" and not shortcut: self._stop_recording_shortcut() self.shortcut_info_label.set_text("Recording cancelled.") return True - if shortcut and is_valid_shortcut(shortcut): - self.custom_shortcut_entry.set_text(shortcut) - self._stop_recording_shortcut() - self._apply_custom_shortcut(shortcut) - else: - self.shortcut_info_label.set_markup( - "Need a modifier + key, or an F1–F24 " - "function key alone. Try again or press Esc to cancel." - ) + self.shortcut_info_label.set_markup( + "Need a modifier + key, or an F1–F24 " + "function key alone. Try again or press Esc to cancel." + ) return True def _update_shortcut_ui_for_mode(self, mode: str): @@ -7271,6 +7597,8 @@ For now, the engine has been reverted to VOSK.""" def _on_dialog_destroy(self, widget): """Clean up callbacks when dialog is destroyed.""" + self._recording_shortcut = False + self._stop_evdev_shortcut_recorder() if hasattr(self, "speech_engine") and self.speech_engine: if self._on_recognition_state_changed in self.speech_engine.state_callbacks: self.speech_engine.state_callbacks.remove(self._on_recognition_state_changed) diff --git a/tests/test_shortcut_capture.py b/tests/test_shortcut_capture.py index 7a1f42d..dabe12e 100644 --- a/tests/test_shortcut_capture.py +++ b/tests/test_shortcut_capture.py @@ -1,14 +1,48 @@ """Tests for the shortcut-capture keyname mapping in the settings dialog. -Covers the pure `_gdk_keyname_to_token` helper that turns a GDK key-symbol name -(from the "Record shortcut" capture) into a canonical main-key token. The GTK -dialog itself can't be instantiated under the mocked-GTK test harness, but this -module-level helper is pure and directly testable. +Covers the pure helpers that turn a GDK key-symbol name and/or hardware +keycode (from the "Record shortcut" capture) into a canonical main-key token. +The GTK dialog itself can't be instantiated under the mocked-GTK test harness, +but these module-level helpers are pure and directly testable. + +Also drives `_EvdevShortcutRecorder` with mocked evdev devices and GLib watches +(no `/dev/input`, no SettingsDialog construction). """ +import importlib +import types +from typing import Any, Optional +from unittest.mock import MagicMock + import pytest -from vocalinux.ui.settings_dialog import _gdk_keyname_to_token, _shortcut_from_capture +from vocalinux.ui.keyboard_backends.evdev_backend import DEVICE_RESCAN_SECONDS +from vocalinux.ui.settings_dialog import ( + SettingsDialog, + _EvdevShortcutRecorder, + _gdk_capture_to_shortcut, + _gdk_keyname_to_token, + _shortcut_from_capture, + function_token_from_evdev_code, + function_token_from_gdk_hardware_keycode, + modifiers_from_active_evdev_codes, +) + +# linux/input.h / python-evdev values used by the recorder. +_EV_KEY = 1 +_KEY_A = 30 +_KEY_LEFTCTRL = 29 +_KEY_LEFTALT = 56 +_KEY_LEFTSHIFT = 42 +_KEY_F13 = 183 +_KEY_F19 = 189 +_KEY_F24 = 194 + +_IO_IN = 1 +_IO_OUT = 4 +_IO_PRI = 2 +_IO_ERR = 8 +_IO_HUP = 16 @pytest.mark.parametrize( @@ -48,6 +82,7 @@ def test_rejects_modifiers_and_unknown(name: str | None) -> None: ([], "space", None), ([], None, None), (["shift"], "f10", "shift+f10"), + ([], "f19", "f19"), ], ) def test_shortcut_from_capture( @@ -56,3 +91,453 @@ def test_shortcut_from_capture( expected: str | None, ) -> None: assert _shortcut_from_capture(modifiers, token) == expected + + +@pytest.mark.parametrize( + "hardware,expected", + [ + (191, "f13"), + (197, "f19"), + (202, "f24"), + (190, None), + (203, None), + (0, None), + (67, None), # F1 XKB keycode; F1–F12 come from keyval names + ], +) +def test_function_token_from_gdk_hardware_keycode(hardware: int, expected: str | None) -> None: + assert function_token_from_gdk_hardware_keycode(hardware) == expected + + +@pytest.mark.parametrize( + "code,expected", + [ + (183, "f13"), + (189, "f19"), + (194, "f24"), + (182, None), + (195, None), + ], +) +def test_function_token_from_evdev_code(code: int, expected: str | None) -> None: + assert function_token_from_evdev_code(code) == expected + + +def test_capture_f19_from_hardware_without_keyval() -> None: + assert _gdk_capture_to_shortcut([], "", 197) == "f19" + assert _gdk_capture_to_shortcut([], None, 197) == "f19" + + +def test_capture_f13_from_hardware_despite_xf86_keyval() -> None: + assert _gdk_capture_to_shortcut([], "XF86Tools", 191) == "f13" + + +def test_capture_f19_from_keyval_with_hardware_zero() -> None: + assert _gdk_capture_to_shortcut([], "F19", 0) == "f19" + + +def test_xf86_name_is_not_guessed_without_hardware() -> None: + assert _gdk_capture_to_shortcut([], "XF86Tools", 0) is None + assert _gdk_keyname_to_token("XF86Tools") is None + + +def test_modifiers_from_active_evdev_codes() -> None: + assert modifiers_from_active_evdev_codes({29}) == ["ctrl"] + assert modifiers_from_active_evdev_codes({189, 56}) == ["alt"] + assert ( + _shortcut_from_capture( + modifiers_from_active_evdev_codes({189, 56}), + function_token_from_evdev_code(189), + ) + == "alt+f19" + ) + + +class _FakeEvdevEvent: + def __init__(self, code: int, value: int, ev_type: int = _EV_KEY) -> None: + self.type = ev_type + self.code = code + self.value = value + + +class _FakeEvdevDevice: + def __init__( + self, + path: str, + fd: int, + *, + active_keys: Optional[set[int]] = None, + active_keys_error: Optional[BaseException] = None, + ) -> None: + self.path = path + self._fd = fd + self._active_keys = set(active_keys or ()) + self._active_keys_error = active_keys_error + self.pending: list[_FakeEvdevEvent] = [] + self.read_error: Optional[BaseException] = None + self.closed = False + self.read_calls = 0 + + def fileno(self) -> int: + return self._fd + + def read(self) -> list[_FakeEvdevEvent]: + self.read_calls += 1 + if self.read_error is not None: + raise self.read_error + events = self.pending + self.pending = [] + return events + + def close(self) -> None: + self.closed = True + + def active_keys(self) -> set[int]: + if self._active_keys_error is not None: + raise self._active_keys_error + return set(self._active_keys) + + +class _EvdevRecorderHarness: + """Patch GLib watches, evdev availability, and InputDevice for one test.""" + + def __init__(self) -> None: + self.watches: list[dict[str, Any]] = [] + self.timeouts: list[dict[str, Any]] = [] + self.removed: list[Any] = [] + self.devices: dict[str, _FakeEvdevDevice] = {} + self.find_keyboard_devices = MagicMock(return_value=[]) + self._next_id = 1 + self._next_fd = 100 + + def add_device( + self, + path: str, + *, + active_keys: Optional[set[int]] = None, + active_keys_error: Optional[BaseException] = None, + ) -> _FakeEvdevDevice: + device = _FakeEvdevDevice( + path, + self._next_fd, + active_keys=active_keys, + active_keys_error=active_keys_error, + ) + self._next_fd += 1 + self.devices[path] = device + return device + + def io_add_watch(self, fd: int, flags: int, callback: Any) -> int: + watch_id = self._next_id + self._next_id += 1 + self.watches.append({"id": watch_id, "fd": fd, "flags": flags, "callback": callback}) + return watch_id + + def timeout_add(self, interval_ms: int, callback: Any) -> int: + timeout_id = self._next_id + self._next_id += 1 + self.timeouts.append({"id": timeout_id, "interval_ms": interval_ms, "callback": callback}) + return timeout_id + + def source_remove(self, source_id: Any) -> bool: + self.removed.append(source_id) + return True + + def _input_device(self, path: str) -> _FakeEvdevDevice: + if path not in self.devices: + raise OSError(f"No such device: {path}") + return self.devices[path] + + def watch_for(self, path: str) -> dict[str, Any]: + fd = self.devices[path].fileno() + matches = [watch for watch in self.watches if watch["fd"] == fd] + assert matches, f"no io watch for {path}" + return matches[-1] + + def start_recorder( + self, + paths: list[str], + on_shortcut: Optional[Any] = None, + ) -> tuple[_EvdevShortcutRecorder, list[str]]: + self.find_keyboard_devices.return_value = list(paths) + committed: list[str] = [] + callback = on_shortcut if on_shortcut is not None else committed.append + recorder = _EvdevShortcutRecorder(callback) + recorder.start() + return recorder, committed + + def fire_io(self, path: str, condition: int = _IO_IN) -> bool: + watch = self.watch_for(path) + return bool(watch["callback"](watch["fd"], condition)) + + +@pytest.fixture +def evdev_recorder(monkeypatch: pytest.MonkeyPatch) -> _EvdevRecorderHarness: + harness = _EvdevRecorderHarness() + fake_glib = types.SimpleNamespace( + IO_IN=_IO_IN, + IO_OUT=_IO_OUT, + IO_PRI=_IO_PRI, + IO_ERR=_IO_ERR, + IO_HUP=_IO_HUP, + io_add_watch=harness.io_add_watch, + timeout_add=harness.timeout_add, + source_remove=harness.source_remove, + ) + # Methods close over the original module dict, not a reimported sys.modules entry. + monkeypatch.setitem(_EvdevShortcutRecorder.start.__globals__, "GLib", fake_glib) + monkeypatch.setattr( + "vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", + True, + ) + monkeypatch.setattr( + "vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices", + harness.find_keyboard_devices, + ) + # Keyboard tests pop sys.modules["evdev"]; patch the live module, not a stale import. + evdev_mod = importlib.import_module("evdev") + monkeypatch.setattr(evdev_mod, "InputDevice", harness._input_device) + try: + evdev_device = importlib.import_module("evdev.device") + except ImportError: + pass + else: + monkeypatch.setattr(evdev_device, "InputDevice", harness._input_device) + return harness + + +def test_f19_press_commits_once_non_function_and_release_do_not( + evdev_recorder: _EvdevRecorderHarness, +) -> None: + path = "/dev/input/event0" + device = evdev_recorder.add_device(path) + committed: list[str] = [] + recorder_box: list[_EvdevShortcutRecorder] = [] + + def on_shortcut(shortcut: str) -> None: + committed.append(shortcut) + recorder_box[0].stop() + + recorder, _ = evdev_recorder.start_recorder([path], on_shortcut) + recorder_box.append(recorder) + + assert len(evdev_recorder.timeouts) == 1 + assert evdev_recorder.timeouts[0]["interval_ms"] == int(DEVICE_RESCAN_SECONDS * 1000) + in_flight = evdev_recorder.watch_for(path) + + device.pending.append(_FakeEvdevEvent(_KEY_A, 1)) + assert evdev_recorder.fire_io(path) is True + assert committed == [] + + device.pending.append(_FakeEvdevEvent(_KEY_F19, 0)) + assert evdev_recorder.fire_io(path) is True + assert committed == [] + + device.pending.append(_FakeEvdevEvent(_KEY_F19, 1)) + assert evdev_recorder.fire_io(path) is False + assert committed == ["f19"] + assert in_flight["id"] not in evdev_recorder.removed + assert evdev_recorder.timeouts[0]["id"] in evdev_recorder.removed + + +def test_modifiers_split_across_devices_alt_f19( + evdev_recorder: _EvdevRecorderHarness, +) -> None: + left_path = "/dev/input/left" + right_path = "/dev/input/right" + left = evdev_recorder.add_device(left_path) + right = evdev_recorder.add_device(right_path) + recorder, committed = evdev_recorder.start_recorder([left_path, right_path]) + + left.pending.append(_FakeEvdevEvent(_KEY_LEFTALT, 1)) + assert evdev_recorder.fire_io(left_path) is True + assert committed == [] + + right.pending.append(_FakeEvdevEvent(_KEY_F19, 1)) + assert evdev_recorder.fire_io(right_path) is False + assert committed == ["alt+f19"] + + +def test_modifiers_from_active_keys_at_open_ctrl_f13( + evdev_recorder: _EvdevRecorderHarness, +) -> None: + left_path = "/dev/input/left" + right_path = "/dev/input/right" + evdev_recorder.add_device(left_path, active_keys={_KEY_LEFTCTRL}) + right = evdev_recorder.add_device(right_path) + _recorder, committed = evdev_recorder.start_recorder([left_path, right_path]) + + right.pending.append(_FakeEvdevEvent(_KEY_F13, 1)) + assert evdev_recorder.fire_io(right_path) is False + assert committed == ["ctrl+f13"] + + +def test_per_device_held_sets_release_on_one_half_keeps_the_other( + evdev_recorder: _EvdevRecorderHarness, +) -> None: + left_path = "/dev/input/left" + right_path = "/dev/input/right" + left = evdev_recorder.add_device(left_path) + right = evdev_recorder.add_device(right_path) + _recorder, committed = evdev_recorder.start_recorder([left_path, right_path]) + + left.pending.append(_FakeEvdevEvent(_KEY_LEFTALT, 1)) + assert evdev_recorder.fire_io(left_path) is True + right.pending.append(_FakeEvdevEvent(_KEY_LEFTSHIFT, 1)) + assert evdev_recorder.fire_io(right_path) is True + + left.pending.append(_FakeEvdevEvent(_KEY_LEFTALT, 0)) + assert evdev_recorder.fire_io(left_path) is True + + right.pending.append(_FakeEvdevEvent(_KEY_F19, 1)) + assert evdev_recorder.fire_io(right_path) is False + assert committed == ["shift+f19"] + + +def test_ctrl_and_alt_on_separate_halves_commit_canonical_order( + evdev_recorder: _EvdevRecorderHarness, +) -> None: + left_path = "/dev/input/left" + right_path = "/dev/input/right" + left = evdev_recorder.add_device(left_path) + right = evdev_recorder.add_device(right_path) + _recorder, committed = evdev_recorder.start_recorder([left_path, right_path]) + + left.pending.append(_FakeEvdevEvent(_KEY_LEFTCTRL, 1)) + assert evdev_recorder.fire_io(left_path) is True + right.pending.append(_FakeEvdevEvent(_KEY_LEFTALT, 1)) + assert evdev_recorder.fire_io(right_path) is True + left.pending.append(_FakeEvdevEvent(_KEY_F24, 1)) + assert evdev_recorder.fire_io(left_path) is False + assert committed == ["ctrl+alt+f24"] + + +def test_read_oserror_drops_only_that_device_and_rescan_attaches_replacement( + evdev_recorder: _EvdevRecorderHarness, +) -> None: + left_path = "/dev/input/left" + right_path = "/dev/input/right" + left = evdev_recorder.add_device(left_path) + right = evdev_recorder.add_device(right_path) + recorder, committed = evdev_recorder.start_recorder([left_path, right_path]) + timeout = evdev_recorder.timeouts[0] + right_watch_id = evdev_recorder.watch_for(right_path)["id"] + + left.read_error = OSError("disconnected") + assert evdev_recorder.fire_io(left_path) is False + assert left.closed + assert not right.closed + assert right_watch_id not in evdev_recorder.removed + assert timeout["id"] not in evdev_recorder.removed + assert committed == [] + + replacement_path = "/dev/input/event99" + replacement = evdev_recorder.add_device(replacement_path) + evdev_recorder.find_keyboard_devices.return_value = [right_path, replacement_path] + assert timeout["callback"]() is True + assert evdev_recorder.watch_for(replacement_path)["fd"] == replacement.fileno() + assert recorder._active is True + + +def test_io_hup_and_io_err_drop_without_reading( + evdev_recorder: _EvdevRecorderHarness, +) -> None: + hup_path = "/dev/input/hup" + err_path = "/dev/input/err" + keep_path = "/dev/input/keep" + hup_dev = evdev_recorder.add_device(hup_path) + err_dev = evdev_recorder.add_device(err_path) + keep = evdev_recorder.add_device(keep_path) + _recorder, committed = evdev_recorder.start_recorder([hup_path, err_path, keep_path]) + timeout_id = evdev_recorder.timeouts[0]["id"] + + hup_dev.pending.append(_FakeEvdevEvent(_KEY_F19, 1)) + assert evdev_recorder.fire_io(hup_path, _IO_HUP) is False + assert hup_dev.read_calls == 0 + assert hup_dev.closed + assert committed == [] + + err_dev.pending.append(_FakeEvdevEvent(_KEY_F19, 1)) + assert evdev_recorder.fire_io(err_path, _IO_ERR) is False + assert err_dev.read_calls == 0 + assert err_dev.closed + assert committed == [] + + assert not keep.closed + assert timeout_id not in evdev_recorder.removed + + +def test_blocking_io_error_keeps_device_and_watch( + evdev_recorder: _EvdevRecorderHarness, +) -> None: + path = "/dev/input/event0" + device = evdev_recorder.add_device(path) + recorder, committed = evdev_recorder.start_recorder([path]) + watch_id = evdev_recorder.watch_for(path)["id"] + + device.read_error = BlockingIOError() + assert evdev_recorder.fire_io(path) is True + assert not device.closed + assert watch_id not in evdev_recorder.removed + assert recorder._active is True + assert committed == [] + + +def test_stop_is_idempotent_and_callback_after_stop_does_not_commit( + evdev_recorder: _EvdevRecorderHarness, +) -> None: + left_path = "/dev/input/left" + right_path = "/dev/input/right" + left = evdev_recorder.add_device(left_path) + right = evdev_recorder.add_device(right_path) + recorder, committed = evdev_recorder.start_recorder([left_path, right_path]) + saved_callback = evdev_recorder.watch_for(left_path)["callback"] + watch_ids = {watch["id"] for watch in evdev_recorder.watches} + timeout_id = evdev_recorder.timeouts[0]["id"] + + recorder.stop() + assert left.closed + assert right.closed + assert watch_ids <= set(evdev_recorder.removed) + assert timeout_id in evdev_recorder.removed + + recorder.stop() + + left.pending.append(_FakeEvdevEvent(_KEY_F19, 1)) + assert saved_callback(left.fileno(), _IO_IN) is False + assert committed == [] + + +def _settings_dialog_unbound(name: str) -> Any: + """Return a SettingsDialog method without instantiating the dialog. + + Gtk.Dialog is a MagicMock in this harness, so `class SettingsDialog` is + constructed as a MagicMock and the class body lives on ``return_value``. + """ + namespace = SettingsDialog.return_value + return namespace[name] + + +def test_destroy_clears_recording_flag_and_closes_devices( + evdev_recorder: _EvdevRecorderHarness, +) -> None: + path = "/dev/input/event0" + device = evdev_recorder.add_device(path) + recorder, _committed = evdev_recorder.start_recorder([path]) + destroy = _settings_dialog_unbound("_on_shortcut_recorder_destroy") + stop = _settings_dialog_unbound("_stop_evdev_shortcut_recorder") + + class _DialogStub: + _stop_evdev_shortcut_recorder = stop + + def __init__(self) -> None: + self._recording_shortcut = True + self._evdev_shortcut_recorder = recorder + + stub = _DialogStub() + destroy(stub, object()) + assert stub._recording_shortcut is False + assert stub._evdev_shortcut_recorder is None + assert device.closed + assert {watch["id"] for watch in evdev_recorder.watches} <= set(evdev_recorder.removed) + assert evdev_recorder.timeouts[0]["id"] in evdev_recorder.removed