fix(shortcuts): record unmapped F19/F24 and XF86-aliased F13–F23 (#844)
* fix(shortcuts): record unmapped F19/F24 and XF86-aliased F13–F23 * fix(shortcuts): keep split-keyboard modifiers while recording The recorder only looked at the keyboard that sent the function key, so a modifier held on the other half was dropped. A dead device also stayed in the watch list, which left F19/F24 uncapturable until Record was restarted. Union held keys across every open keyboard before committing, and drop a device on hangup or read error while scanning for a replacement. * fix(shortcuts): keep capture tests off real input devices CI opened real /dev/input paths because the harness patched a stale evdev module, so io watches never attached. Patch the live module, and move the capture device protocol to common_types. * fix(tests): patch the recorder's own GLib globals test_settings_dialog reimports the dialog module and leaves the replacement in sys.modules. Recorder methods still close over the original dict, so the harness never recorded timeouts or io watches.
This commit is contained in:
Jatin K Malik
@@ -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."""
|
||||
...
|
||||
|
||||
@@ -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(
|
||||
"<i>Press a modifier + key (e.g. Alt+R), or an F-key. Press Esc to cancel.</i>"
|
||||
)
|
||||
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(
|
||||
"<span foreground='#e01b24'>Need a modifier + key, or an F1–F24 "
|
||||
"function key alone. Try again or press Esc to cancel.</span>"
|
||||
)
|
||||
self.shortcut_info_label.set_markup(
|
||||
"<span foreground='#e01b24'>Need a modifier + key, or an F1–F24 "
|
||||
"function key alone. Try again or press Esc to cancel.</span>"
|
||||
)
|
||||
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)
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user