fix(hotkeys): consume the dictation shortcut so it stops leaking to apps (#873)
* fix(hotkeys): consume the dictation shortcut so it stops leaking to apps Grab each monitored keyboard with EVIOCGRAB and re-emit every event the shortcut does not own through a paired uinput clone. The configured shortcut key is now consumed by Vocalinux instead of also reaching the focused app (right-alt opening menus in Thunderbird/Firefox, F11/F12 triggering browser actions). - Pure-modifier gestures reserve the modifier's events entirely; bare function-key shortcuts consume that key; modifier+key combos consume the main key only while the combo owns it, so typing and chords like Alt+Tab keep working and injected text is untouched - BUS_VIRTUAL devices (our clones, ydotool, remappers) are never grabbed - EV_REP is stripped from clones and repeats are forwarded verbatim, so repeat rate is preserved without doubling - Feedback event types (EV_LED/EV_SND/EV_FF) are never re-injected - Falls back to ungrabbed listening when /dev/uinput or the grab is unavailable, warning once that keys will also reach the app Fixes #871 * test(hotkeys): cover grab/forward edge cases Race cleanup on duplicate open, forwarder close errors during stop and device removal, mid-SYN_DROPPED forwarding, write failures, and the combo consume edge cases (unowned repeat/release, missing main key). * fix(installer): grant read on /dev/uinput so key suppression works The evdev backend's uinput clone opens /dev/uinput O_RDWR, but the installed 80-dotool.rules gave the input group write-only access (0620), so shortcut suppression silently fell back to ungrabbed monitoring on stock installs. Bump the rule to 0660 and refresh it when the installed copy differs so existing installs pick up the fix. * fix(hotkeys): harden evdev grab/forward lifecycle and udev install - release the device grab and close the clone when a uinput write fails - close all devices if the monitor exits unexpectedly (releases grabs) - monitor virtual devices too; skip only our own uinput clones - track consumed combo presses per device so releases stay paired - after SYN_DROPPED, release keys the clone still believes are held - append the uinput udev rule instead of overwriting local rules - annotate new signatures and narrow the clone-creation catch * fix(hotkeys): reconcile swallowed combo releases after SYN_DROPPED - resync the swallowed-press pairing even when the clone holds no keys; a dropped combo release otherwise leaves the next ordinary press consumed, eating a keystroke - narrow every new close-cleanup catch to OSError/IOError/RuntimeError with debug context instead of blanket Exception passes --------- Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
This commit is contained in:
devin-ai-integration[bot]
co-authored by
Jatin K Malik
Jatin K Malik
parent
004696bbe0
commit
ff108595fd
@@ -752,11 +752,16 @@ install_text_input_tools() {
|
||||
print_warning "You will need to LOG OUT and back in for text injection to work with ydotool/dotool"
|
||||
fi
|
||||
|
||||
# Install udev rule for ydotool/dotool
|
||||
if [ ! -f /etc/udev/rules.d/80-dotool.rules ]; then
|
||||
# Install udev rule for ydotool/dotool — and for the evdev
|
||||
# backend's uinput clone, which opens /dev/uinput O_RDWR, so the
|
||||
# rule needs 0660 (0620 is write-only). Appending (not overwriting)
|
||||
# preserves any rules the user already keeps in this file; when an
|
||||
# older MODE=0620 line is present our later line still wins.
|
||||
UINPUT_UDEV_RULE='KERNEL=="uinput", GROUP="input", MODE="0660", OPTIONS+="static_node=uinput"'
|
||||
if ! grep -qxF "$UINPUT_UDEV_RULE" /etc/udev/rules.d/80-dotool.rules 2>/dev/null; then
|
||||
print_info "Installing udev rule for input device access..."
|
||||
echo 'KERNEL=="uinput", GROUP="input", MODE="0620", OPTIONS+="static_node=uinput"' \
|
||||
| sudo tee /etc/udev/rules.d/80-dotool.rules >/dev/null 2>&1 || print_warning "Failed to install udev rule"
|
||||
printf '\n%s\n' "$UINPUT_UDEV_RULE" \
|
||||
| sudo tee -a /etc/udev/rules.d/80-dotool.rules >/dev/null 2>&1 || print_warning "Failed to install udev rule"
|
||||
sudo udevadm control --reload 2>/dev/null || true
|
||||
sudo udevadm trigger 2>/dev/null || true
|
||||
fi
|
||||
|
||||
@@ -3,6 +3,12 @@ evdev keyboard backend for Wayland support.
|
||||
|
||||
This backend uses python-evdev to read keyboard events directly from
|
||||
input devices, which works on both X11 and Wayland (with proper permissions).
|
||||
|
||||
When /dev/uinput is writable, each opened keyboard is grabbed (EVIOCGRAB)
|
||||
and its events are re-emitted through a paired uinput clone, so the keys
|
||||
that make up the dictation shortcut are consumed here and never reach the
|
||||
focused application. Without uinput access the backend still listens
|
||||
ungrabbed, but shortcut presses also pass through to apps.
|
||||
"""
|
||||
|
||||
import errno
|
||||
@@ -12,17 +18,20 @@ import re
|
||||
import select
|
||||
import threading
|
||||
import time
|
||||
from typing import Optional, TextIO
|
||||
from typing import Optional, Sequence, TextIO
|
||||
|
||||
# Try to import evdev
|
||||
try:
|
||||
import evdev
|
||||
from evdev import InputDevice, ecodes
|
||||
from evdev import InputDevice, InputEvent, UInput, UInputError, ecodes
|
||||
|
||||
EVDEV_AVAILABLE = True
|
||||
except ImportError:
|
||||
evdev = None # type: ignore
|
||||
InputDevice = None # type: ignore
|
||||
InputEvent = None # type: ignore
|
||||
UInput = None # type: ignore
|
||||
UInputError = OSError # type: ignore
|
||||
ecodes = None # type: ignore
|
||||
EVDEV_AVAILABLE = False
|
||||
|
||||
@@ -49,6 +58,23 @@ KEY_LEFTMETA = 125 # Super/Windows key
|
||||
KEY_RIGHTMETA = 126
|
||||
DEVICE_RESCAN_SECONDS = 2.0
|
||||
|
||||
# Event types re-emitted on the uinput clone. Feedback events (EV_LED,
|
||||
# EV_SND, EV_FF) are kernel->device traffic already delivered to every
|
||||
# capable device — re-injecting them could feed back into the LED layer.
|
||||
if EVDEV_AVAILABLE:
|
||||
_FORWARDED_EVENT_TYPES = frozenset(
|
||||
{
|
||||
ecodes.EV_KEY,
|
||||
ecodes.EV_REL,
|
||||
ecodes.EV_ABS,
|
||||
ecodes.EV_MSC,
|
||||
ecodes.EV_SW,
|
||||
getattr(ecodes, "EV_ROT", -1),
|
||||
}
|
||||
)
|
||||
else:
|
||||
_FORWARDED_EVENT_TYPES = frozenset()
|
||||
|
||||
# Map modifier key names to evdev key codes
|
||||
MODIFIER_KEY_CODES: dict[str, set[int]] = {
|
||||
"ctrl": {KEY_LEFTCTRL, KEY_RIGHTCTRL},
|
||||
@@ -311,11 +337,25 @@ class EvdevKeyboardBackend(KeyboardBackend):
|
||||
self._combo_all_codes: set[int] = set()
|
||||
self._combo_pressed: set[int] = set() # currently-held combo-relevant codes
|
||||
self._combo_active = False # True while a push-to-talk combo hold is live
|
||||
# fds whose combo main-key press was consumed but not yet released —
|
||||
# per-device so one keyboard's release can't consume another's press.
|
||||
self._combo_swallowed: set[int] = set()
|
||||
self._resolve_combo_targets()
|
||||
|
||||
self._devices_lock = threading.Lock()
|
||||
self._dropped_devices: set[int] = set() # fds with SYN_DROPPED pending
|
||||
self._device_paths_by_fd: dict[int, str] = {}
|
||||
# Grabbed fd -> uinput clone that re-emits events we do not consume.
|
||||
self._forwarders: dict[int, UInput] = {}
|
||||
# Grabbed fd -> its clone's /dev/input/eventN path, and the live set
|
||||
# of clone paths. Clones are never monitored: their events are the
|
||||
# ones we forwarded, so reading them would loop input back in.
|
||||
self._forwarder_paths: dict[int, str] = {}
|
||||
self._clone_paths: set[str] = set()
|
||||
# Grabbed fd -> key codes the clone currently believes are held.
|
||||
# Needed to reconcile after SYN_DROPPED eats release events.
|
||||
self._forwarded_held: dict[int, set[int]] = {}
|
||||
self._uinput_warned = False
|
||||
|
||||
if not EVDEV_AVAILABLE:
|
||||
logger.error("python-evdev not available")
|
||||
@@ -336,6 +376,7 @@ class EvdevKeyboardBackend(KeyboardBackend):
|
||||
self._combo_all_codes = set()
|
||||
self._combo_pressed = set()
|
||||
self._combo_active = False
|
||||
self._combo_swallowed = set()
|
||||
|
||||
spec = getattr(self, "_spec", None)
|
||||
if spec is None or not spec.is_combo:
|
||||
@@ -374,6 +415,12 @@ class EvdevKeyboardBackend(KeyboardBackend):
|
||||
push-to-talk hold could stay stuck on. Clearing the pressed set fails
|
||||
safe (the user simply re-presses the modifier), and ending any live hold
|
||||
prevents a stuck session.
|
||||
|
||||
``_combo_swallowed`` is deliberately left alone: it records consumed
|
||||
presses per device, so clearing it here would let the matching
|
||||
release of a still-held key leak to the application without its
|
||||
press. Stale entries are pruned per-device on removal and during
|
||||
the SYN_DROPPED resync instead.
|
||||
"""
|
||||
self._combo_pressed = set()
|
||||
# Ends an active push-to-talk hold (fires the release callback); no-op
|
||||
@@ -486,6 +533,11 @@ class EvdevKeyboardBackend(KeyboardBackend):
|
||||
self.key_pressed_devices = set()
|
||||
self._dropped_devices = set()
|
||||
self._device_paths_by_fd = {}
|
||||
self._forwarders = {}
|
||||
self._forwarder_paths = {}
|
||||
self._clone_paths = set()
|
||||
self._forwarded_held = {}
|
||||
self._uinput_warned = False
|
||||
|
||||
# Refresh combo targets and clear any stale held-key state.
|
||||
self._resolve_combo_targets()
|
||||
@@ -520,20 +572,39 @@ class EvdevKeyboardBackend(KeyboardBackend):
|
||||
self.monitor_thread.join(timeout=2.0)
|
||||
self.monitor_thread = None
|
||||
|
||||
# Close devices
|
||||
self._close_all_devices()
|
||||
|
||||
def _close_all_devices(self) -> None:
|
||||
"""Close every monitored device and uinput clone.
|
||||
|
||||
Closing a grabbed device releases its EVIOCGRAB, and closing a
|
||||
uinput clone makes the kernel release any keys the clone still
|
||||
holds — together they hand input delivery back to applications.
|
||||
"""
|
||||
with self._devices_lock:
|
||||
devices = list(self.devices)
|
||||
forwarders = list(self._forwarders.values())
|
||||
self.devices = []
|
||||
self.device_fds = []
|
||||
self.device_paths = set()
|
||||
self._dropped_devices = set()
|
||||
self._device_paths_by_fd = {}
|
||||
self._forwarders = {}
|
||||
self._forwarder_paths = {}
|
||||
self._clone_paths = set()
|
||||
self._forwarded_held = {}
|
||||
self._combo_swallowed = set()
|
||||
|
||||
for device in devices:
|
||||
try:
|
||||
device.close()
|
||||
except Exception:
|
||||
pass
|
||||
except (OSError, IOError, RuntimeError) as e:
|
||||
logger.debug(f"Ignoring device close failure during cleanup: {e}")
|
||||
for forwarder in forwarders:
|
||||
try:
|
||||
forwarder.close()
|
||||
except (OSError, IOError, RuntimeError) as e:
|
||||
logger.debug(f"Ignoring clone close failure during cleanup: {e}")
|
||||
|
||||
def _open_keyboard_device(self, device_path: str) -> bool:
|
||||
"""Open a keyboard device if it is not already monitored."""
|
||||
@@ -548,22 +619,229 @@ class EvdevKeyboardBackend(KeyboardBackend):
|
||||
logger.warning(f"Cannot open {device_path}: {e}")
|
||||
return False
|
||||
|
||||
# Only Vocalinux's own uinput clones are never monitored — reading
|
||||
# them would loop the events we forwarded straight back in. Clones
|
||||
# are matched by their registered device path; the " (vocalinux)"
|
||||
# name suffix also covers clones of other Vocalinux processes whose
|
||||
# paths we never registered. Other virtual devices (key remappers,
|
||||
# accessibility keyboards) ARE monitored: users can bind shortcuts
|
||||
# to them, and grabbing is what lets the shortcut reach us at all
|
||||
# when a remapper has already grabbed the physical device.
|
||||
device_name = str(getattr(device, "name", "") or "")
|
||||
if device_path in self._clone_paths or device_name.endswith(" (vocalinux)"):
|
||||
logger.debug(f"Skipping Vocalinux clone device: {device_path} ({device_name})")
|
||||
try:
|
||||
device.close()
|
||||
except (OSError, IOError, RuntimeError) as e:
|
||||
logger.debug(f"Ignoring close failure for clone {device_path}: {e}")
|
||||
return False
|
||||
|
||||
# Exclusive grab hides the device from the compositor; the paired
|
||||
# clone re-emits every event we do not consume so apps keep typing.
|
||||
forwarder = self._create_forwarder(device)
|
||||
if forwarder is not None:
|
||||
try:
|
||||
device.grab()
|
||||
except (OSError, IOError) as e:
|
||||
logger.warning(
|
||||
f"Cannot grab {device_path} ({e}); "
|
||||
"shortcut keys will also reach the focused app"
|
||||
)
|
||||
try:
|
||||
forwarder.close()
|
||||
except (OSError, IOError, RuntimeError) as e:
|
||||
logger.debug(f"Ignoring clone close failure for {device_path}: {e}")
|
||||
forwarder = None
|
||||
|
||||
with self._devices_lock:
|
||||
if device_path in self.device_paths or fd in self.device_fds:
|
||||
try:
|
||||
device.close()
|
||||
except Exception:
|
||||
pass
|
||||
except (OSError, IOError, RuntimeError) as e:
|
||||
logger.debug(f"Ignoring close failure for duplicate {device_path}: {e}")
|
||||
if forwarder is not None:
|
||||
try:
|
||||
forwarder.close()
|
||||
except (OSError, IOError, RuntimeError) as e:
|
||||
logger.debug(
|
||||
f"Ignoring clone close failure for duplicate {device_path}: {e}"
|
||||
)
|
||||
return False
|
||||
|
||||
self.devices.append(device)
|
||||
self.device_fds.append(fd)
|
||||
self.device_paths.add(device_path)
|
||||
self._device_paths_by_fd[fd] = device_path
|
||||
if forwarder is not None:
|
||||
self._forwarders[fd] = forwarder
|
||||
clone_path = self._forwarder_device_path(forwarder)
|
||||
if clone_path is not None:
|
||||
self._forwarder_paths[fd] = clone_path
|
||||
self._clone_paths.add(clone_path)
|
||||
self._forwarded_held[fd] = set()
|
||||
|
||||
logger.debug(f"Opened keyboard device: {device_path} ({device.name})")
|
||||
logger.debug(
|
||||
f"Opened keyboard device: {device_path} ({device.name})"
|
||||
f"{' [grabbed]' if forwarder is not None else ''}"
|
||||
)
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def _forwarder_device_path(forwarder: UInput) -> Optional[str]:
|
||||
"""Return the clone's /dev/input/eventN path, or None if unknowable.
|
||||
|
||||
The path is what lets ``_open_keyboard_device`` recognize (and skip)
|
||||
our clones even when the clone name was truncated to the uinput
|
||||
80-byte limit.
|
||||
"""
|
||||
try:
|
||||
path = forwarder.device.path
|
||||
except (AttributeError, OSError, IOError):
|
||||
return None
|
||||
return path if isinstance(path, str) else None
|
||||
|
||||
def _create_forwarder(self, device: InputDevice) -> Optional[UInput]:
|
||||
"""Create a uinput clone of a keyboard for pass-through forwarding.
|
||||
|
||||
EV_REP is filtered out so the clone never autorepeats: the physical
|
||||
device's own repeat events are forwarded verbatim, preserving its
|
||||
repeat rate without doubling repeats.
|
||||
"""
|
||||
try:
|
||||
capabilities: dict[int, Sequence[int]] = {
|
||||
k: v for k, v in device.capabilities().items()
|
||||
}
|
||||
for event_type in (ecodes.EV_SYN, ecodes.EV_FF, ecodes.EV_REP):
|
||||
capabilities.pop(event_type, None)
|
||||
return UInput(
|
||||
events=capabilities,
|
||||
name=f"{device.name} (vocalinux)",
|
||||
bustype=ecodes.BUS_VIRTUAL,
|
||||
)
|
||||
except (OSError, IOError, TypeError, ValueError, UInputError) as e:
|
||||
if not self._uinput_warned:
|
||||
self._uinput_warned = True
|
||||
logger.warning(
|
||||
"Cannot create a virtual keyboard for key suppression "
|
||||
f"(/dev/uinput not writable?): {e}. The dictation shortcut "
|
||||
"will also reach the focused application."
|
||||
)
|
||||
return None
|
||||
|
||||
def _release_failed_forwarder(self, fd: int) -> None:
|
||||
"""Drop a dead uinput clone and release the source device's grab.
|
||||
|
||||
A keyboard that stays grabbed while its clone can no longer accept
|
||||
events swallows all user input; releasing EVIOCGRAB hands delivery
|
||||
back to the kernel so applications see keys again (the shortcut can
|
||||
then no longer be suppressed, matching the ungrabbed fallback). If
|
||||
the grab itself cannot be released, the device is removed outright
|
||||
— a grabbed device with a dead clone is worse than no listener.
|
||||
"""
|
||||
with self._devices_lock:
|
||||
device = next((d for d in self.devices if d.fileno() == fd), None)
|
||||
forwarder = self._forwarders.pop(fd, None)
|
||||
clone_path = self._forwarder_paths.pop(fd, None)
|
||||
if clone_path is not None:
|
||||
self._clone_paths.discard(clone_path)
|
||||
self._forwarded_held.pop(fd, None)
|
||||
|
||||
if forwarder is not None:
|
||||
try:
|
||||
forwarder.close()
|
||||
except (OSError, IOError, RuntimeError) as e:
|
||||
logger.debug(f"Ignoring clone close failure for fd {fd}: {e}")
|
||||
if device is None:
|
||||
return
|
||||
try:
|
||||
device.ungrab()
|
||||
except (OSError, IOError) as e:
|
||||
logger.error(
|
||||
f"Cannot release grab on fd {fd} ({e}); "
|
||||
"removing the device so it stops swallowing input"
|
||||
)
|
||||
self._remove_keyboard_device(fd, device)
|
||||
return
|
||||
logger.warning(
|
||||
f"Released grab on {getattr(device, 'name', 'device')} (fd={fd}) "
|
||||
"after a forwarding failure; keys pass through to applications "
|
||||
"but the dictation shortcut can no longer be suppressed"
|
||||
)
|
||||
|
||||
def _forward_event(self, fd: int, event: InputEvent) -> None:
|
||||
"""Re-emit an event on the uinput clone paired with a grabbed device.
|
||||
|
||||
Successful EV_KEY writes also update ``_forwarded_held`` so
|
||||
``_resync_clone_key_state`` can tell which keys the clone still
|
||||
believes are held after a SYN_DROPPED burst.
|
||||
"""
|
||||
if event.type != ecodes.EV_SYN and event.type not in _FORWARDED_EVENT_TYPES:
|
||||
return
|
||||
forwarder = self._forwarders.get(fd)
|
||||
if forwarder is None:
|
||||
return
|
||||
try:
|
||||
forwarder.write_event(event)
|
||||
except (OSError, IOError) as e:
|
||||
logger.error(f"Failed to forward event on fd {fd}: {e}")
|
||||
self._release_failed_forwarder(fd)
|
||||
return
|
||||
if event.type == ecodes.EV_KEY:
|
||||
held = self._forwarded_held.get(fd)
|
||||
if held is not None:
|
||||
if event.value == 0:
|
||||
held.discard(event.code)
|
||||
else:
|
||||
held.add(event.code)
|
||||
|
||||
def _event_is_shortcut(self, fd: int, event: InputEvent) -> bool:
|
||||
"""Return True if an EV_KEY event belongs to the dictation shortcut.
|
||||
|
||||
Consumed events are handled locally and never re-emitted, so the
|
||||
focused application never sees them:
|
||||
|
||||
- Pure-modifier gestures (``right_alt+right_alt``, ``ctrl+ctrl``):
|
||||
every event of the configured modifier side(s) is consumed — the
|
||||
modifier is reserved for Vocalinux while the listener runs.
|
||||
- Bare function-key shortcuts (``f11``): every event of that key.
|
||||
- Modifier+key combos (``alt+r``): the main key is consumed only
|
||||
while the combo owns it (press with all modifiers held, plus the
|
||||
matching repeat/release). Required modifiers and unrelated keys
|
||||
pass through, so typing and chords like Alt+Tab keep working.
|
||||
|
||||
Consumption of a combo main-key press is tracked per device fd:
|
||||
when two keyboards both press the main key while the modifiers are
|
||||
held, each device's release must still find its own consumed press
|
||||
— a shared flag would let the first release expose the second
|
||||
keyboard's release without its press. Stale entries are pruned on
|
||||
device removal and by the post-SYN_DROPPED resync.
|
||||
"""
|
||||
if event.type != ecodes.EV_KEY:
|
||||
return False
|
||||
spec = getattr(self, "_spec", None)
|
||||
if spec is None:
|
||||
return False
|
||||
|
||||
if not spec.is_combo:
|
||||
return event.code in self._get_target_key_codes()
|
||||
|
||||
main_code = self._combo_main_code
|
||||
if event.code != main_code or main_code is None:
|
||||
return False
|
||||
if not spec.modifiers:
|
||||
return True # bare function key
|
||||
if event.value == 1 and self._required_modifiers_held():
|
||||
self._combo_swallowed.add(fd)
|
||||
return True
|
||||
if event.value == 0:
|
||||
# Consume the release of a swallowed press on THIS device exactly
|
||||
# once, so apps never see a press/release pair we half-forwarded.
|
||||
if fd not in self._combo_swallowed:
|
||||
return False
|
||||
self._combo_swallowed.discard(fd)
|
||||
return True
|
||||
return fd in self._combo_swallowed
|
||||
|
||||
def _scan_for_new_devices(self) -> int:
|
||||
"""Find and open keyboard devices that appeared after startup."""
|
||||
new_device_count = 0
|
||||
@@ -587,8 +865,8 @@ class EvdevKeyboardBackend(KeyboardBackend):
|
||||
"""Close and forget a disconnected keyboard device."""
|
||||
try:
|
||||
device.close()
|
||||
except Exception:
|
||||
pass
|
||||
except (OSError, IOError, RuntimeError) as e:
|
||||
logger.debug(f"Ignoring close failure for disconnected fd {fd}: {e}")
|
||||
|
||||
with self._devices_lock:
|
||||
try:
|
||||
@@ -604,94 +882,182 @@ class EvdevKeyboardBackend(KeyboardBackend):
|
||||
self.device_paths.discard(device_path)
|
||||
self._dropped_devices.discard(fd)
|
||||
self.key_pressed_devices.discard(id(device))
|
||||
# A disconnect can swallow the modifier release (e.g. a wireless
|
||||
# split half dropping mid-hold); don't leave the combo logically held.
|
||||
self._reset_combo_state()
|
||||
forwarder = self._forwarders.pop(fd, None)
|
||||
clone_path = self._forwarder_paths.pop(fd, None)
|
||||
if clone_path is not None:
|
||||
self._clone_paths.discard(clone_path)
|
||||
self._forwarded_held.pop(fd, None)
|
||||
# The fd can be recycled for a newly plugged device; a stale
|
||||
# swallowed press would then eat that device's first release.
|
||||
self._combo_swallowed.discard(fd)
|
||||
|
||||
if forwarder is not None:
|
||||
try:
|
||||
forwarder.close()
|
||||
except (OSError, IOError, RuntimeError) as e:
|
||||
logger.debug(f"Ignoring clone close failure for disconnected fd {fd}: {e}")
|
||||
# A disconnect can swallow the modifier release (e.g. a wireless
|
||||
# split half dropping mid-hold); don't leave the combo logically held.
|
||||
self._reset_combo_state()
|
||||
|
||||
def _resync_clone_key_state(self, fd: int, device: InputDevice) -> None:
|
||||
"""Release clone-side keys whose releases were lost to SYN_DROPPED.
|
||||
|
||||
The kernel's dropped-event burst may have skipped release events,
|
||||
but the paired clone only saw what we forwarded — a forwarded press
|
||||
whose release was dropped leaves the key held on the clone forever,
|
||||
and no later event supplies the missing release. The burst can also
|
||||
eat the release of a swallowed combo press, which the clone never
|
||||
saw at all. Compare what the clone believes (``_forwarded_held``)
|
||||
and what we paired (``_combo_swallowed``) against the device's live
|
||||
kernel key state and emit releases for the phantom keys.
|
||||
"""
|
||||
forwarder = self._forwarders.get(fd)
|
||||
held = self._forwarded_held.get(fd)
|
||||
if (forwarder is None or not held) and fd not in self._combo_swallowed:
|
||||
return
|
||||
try:
|
||||
actually_held = set(device.active_keys())
|
||||
except (OSError, IOError) as e:
|
||||
logger.warning(f"Cannot resync key state for fd {fd}: {e}")
|
||||
return
|
||||
if forwarder is not None and held:
|
||||
for code in sorted(held - actually_held):
|
||||
try:
|
||||
forwarder.write(ecodes.EV_KEY, code, 0)
|
||||
except (OSError, IOError) as e:
|
||||
logger.error(f"Failed to release stuck key {code} on fd {fd}: {e}")
|
||||
self._release_failed_forwarder(fd)
|
||||
return
|
||||
held.discard(code)
|
||||
if self._combo_main_code is not None and self._combo_main_code not in actually_held:
|
||||
# A dropped burst can also eat the release of a swallowed combo
|
||||
# press — a key the clone never saw, so ``held`` cannot reflect
|
||||
# it. Without this prune the next ordinary press of that key
|
||||
# would be consumed too, eating a keystroke.
|
||||
self._combo_swallowed.discard(fd)
|
||||
|
||||
def _monitor_devices(self) -> None:
|
||||
"""Monitor keyboard devices for events."""
|
||||
logger.debug("Starting device monitor thread")
|
||||
last_scan = time.monotonic()
|
||||
|
||||
while self.running:
|
||||
try:
|
||||
now = time.monotonic()
|
||||
if now - last_scan >= DEVICE_RESCAN_SECONDS:
|
||||
self._scan_for_new_devices()
|
||||
last_scan = now
|
||||
try:
|
||||
while self.running:
|
||||
try:
|
||||
now = time.monotonic()
|
||||
if now - last_scan >= DEVICE_RESCAN_SECONDS:
|
||||
self._scan_for_new_devices()
|
||||
last_scan = now
|
||||
|
||||
# Use select to wait for events on any device
|
||||
with self._devices_lock:
|
||||
device_fds = list(self.device_fds)
|
||||
# Use select to wait for events on any device
|
||||
with self._devices_lock:
|
||||
device_fds = list(self.device_fds)
|
||||
|
||||
if not device_fds:
|
||||
time.sleep(1.0)
|
||||
continue
|
||||
|
||||
readable, _, _ = select.select(device_fds, [], [], 1.0) # 1 second timeout
|
||||
|
||||
for fd in readable:
|
||||
try:
|
||||
# Find the device for this fd
|
||||
with self._devices_lock:
|
||||
device = None
|
||||
for d in self.devices:
|
||||
if d.fileno() == fd:
|
||||
device = d
|
||||
break
|
||||
|
||||
if device is None:
|
||||
continue
|
||||
|
||||
# Read events from this device
|
||||
for event in device.read():
|
||||
if event.type == ecodes.EV_SYN:
|
||||
if event.code == ecodes.SYN_DROPPED:
|
||||
# Kernel buffer overflowed — discard until SYN_REPORT
|
||||
self._dropped_devices.add(fd)
|
||||
logger.warning(
|
||||
f"SYN_DROPPED on {device.name} (fd={fd}), "
|
||||
"resetting key state"
|
||||
)
|
||||
elif event.code == ecodes.SYN_REPORT:
|
||||
if fd in self._dropped_devices:
|
||||
# End of dropped sequence — clear stale state
|
||||
self._dropped_devices.discard(fd)
|
||||
self.key_pressed_devices.discard(id(device))
|
||||
# A dropped modifier release must not leave
|
||||
# the combo logically held.
|
||||
self._reset_combo_state()
|
||||
continue
|
||||
if fd in self._dropped_devices:
|
||||
continue
|
||||
if event.type == ecodes.EV_KEY:
|
||||
self._handle_key_event(event, device)
|
||||
|
||||
except (OSError, IOError) as e:
|
||||
# Device was disconnected - remove it to avoid busy loop
|
||||
device_name = (
|
||||
device.name if device and hasattr(device, "name") else "unknown"
|
||||
)
|
||||
logger.info(f"Device disconnected: {device_name} (fd={fd})")
|
||||
if device is not None:
|
||||
self._remove_keyboard_device(fd, device)
|
||||
if not device_fds:
|
||||
time.sleep(1.0)
|
||||
continue
|
||||
|
||||
except (OSError, ValueError) as e:
|
||||
if self.running:
|
||||
logger.error(f"Error monitoring devices: {e}")
|
||||
break
|
||||
readable, _, _ = select.select(device_fds, [], [], 1.0) # 1 second timeout
|
||||
|
||||
for fd in readable:
|
||||
try:
|
||||
# Find the device for this fd
|
||||
with self._devices_lock:
|
||||
device = None
|
||||
for d in self.devices:
|
||||
if d.fileno() == fd:
|
||||
device = d
|
||||
break
|
||||
|
||||
if device is None:
|
||||
continue
|
||||
|
||||
# Read events from this device. On grabbed devices,
|
||||
# everything we do not consume is re-emitted on the
|
||||
# paired uinput clone so the focused app keeps typing.
|
||||
for event in device.read():
|
||||
if event.type == ecodes.EV_SYN:
|
||||
if event.code == ecodes.SYN_DROPPED:
|
||||
# Kernel buffer overflowed — discard until SYN_REPORT
|
||||
self._dropped_devices.add(fd)
|
||||
logger.warning(
|
||||
f"SYN_DROPPED on {device.name} (fd={fd}), "
|
||||
"resetting key state"
|
||||
)
|
||||
elif event.code == ecodes.SYN_REPORT:
|
||||
if fd in self._dropped_devices:
|
||||
# End of dropped sequence — clear stale state
|
||||
self._dropped_devices.discard(fd)
|
||||
self.key_pressed_devices.discard(id(device))
|
||||
# Release keys the clone still thinks
|
||||
# are held before the SYN_REPORT
|
||||
# reaches it, atomically.
|
||||
self._resync_clone_key_state(fd, device)
|
||||
# A dropped modifier release must not leave
|
||||
# the combo logically held.
|
||||
self._reset_combo_state()
|
||||
self._forward_event(fd, event)
|
||||
continue
|
||||
if fd in self._dropped_devices:
|
||||
# Handling shortcut state mid-drop is unsafe,
|
||||
# but surviving non-shortcut events still pass
|
||||
# through so the app sees what the kernel kept.
|
||||
if not self._event_is_shortcut(fd, event):
|
||||
self._forward_event(fd, event)
|
||||
continue
|
||||
if event.type == ecodes.EV_KEY:
|
||||
# Every key event updates shortcut state (combo
|
||||
# modifiers must be tracked AND forwarded); only
|
||||
# the forwarding decision differs.
|
||||
consumed = self._event_is_shortcut(fd, event)
|
||||
self._handle_key_event(event, device)
|
||||
if not consumed:
|
||||
self._forward_event(fd, event)
|
||||
else:
|
||||
self._forward_event(fd, event)
|
||||
|
||||
except (OSError, IOError) as e:
|
||||
# Device was disconnected - remove it to avoid busy loop
|
||||
device_name = (
|
||||
device.name if device and hasattr(device, "name") else "unknown"
|
||||
)
|
||||
logger.info(f"Device disconnected: {device_name} (fd={fd})")
|
||||
if device is not None:
|
||||
self._remove_keyboard_device(fd, device)
|
||||
continue
|
||||
|
||||
except (OSError, ValueError) as e:
|
||||
if self.running:
|
||||
logger.error(f"Error monitoring devices: {e}")
|
||||
break
|
||||
finally:
|
||||
if self.running:
|
||||
# Exiting while still "running" means the loop died
|
||||
# unexpectedly — a failed select, an fd closed underneath
|
||||
# it, or an error escaping the handler. Nothing will ever
|
||||
# read the grabbed keyboards again, so close every device
|
||||
# and clone: closing releases each grab and hands input
|
||||
# delivery back to the kernel instead of leaving the
|
||||
# user's keyboards dead.
|
||||
logger.error(
|
||||
"Keyboard monitor exited unexpectedly; closing devices "
|
||||
"to release grabs so keyboards keep working"
|
||||
)
|
||||
self.running = False
|
||||
self.active = False
|
||||
self._close_all_devices()
|
||||
|
||||
logger.debug("Device monitor thread stopped")
|
||||
|
||||
def _handle_key_event(self, event, device) -> None:
|
||||
def _handle_key_event(self, event: InputEvent, device: InputDevice) -> None:
|
||||
"""Handle a key event from evdev."""
|
||||
if getattr(self, "_spec", None) is not None and self._spec.is_combo:
|
||||
self._handle_combo_key_event(event)
|
||||
return
|
||||
self._handle_modifier_key_event(event, device)
|
||||
|
||||
def _handle_combo_key_event(self, event) -> None:
|
||||
def _handle_combo_key_event(self, event: InputEvent) -> None:
|
||||
"""Handle a key event for a modifier+key combo (e.g. Alt+R).
|
||||
|
||||
Combo state is tracked globally across all keyboard devices, which is
|
||||
@@ -748,7 +1114,7 @@ class EvdevKeyboardBackend(KeyboardBackend):
|
||||
logger.debug(f"Combo {self._shortcut} released (evdev)")
|
||||
threading.Thread(target=self.key_release_callback, daemon=True).start()
|
||||
|
||||
def _handle_modifier_key_event(self, event, device) -> None:
|
||||
def _handle_modifier_key_event(self, event: InputEvent, device: InputDevice) -> None:
|
||||
"""Handle a key event for a single-modifier gesture (double-tap / hold)."""
|
||||
try:
|
||||
code = event.code
|
||||
|
||||
@@ -26,6 +26,7 @@ from vocalinux.ui.keyboard_backends.evdev_backend import (
|
||||
EvdevKeyboardBackend,
|
||||
device_has_key,
|
||||
device_has_modifier_key,
|
||||
ecodes,
|
||||
find_keyboard_devices,
|
||||
)
|
||||
|
||||
@@ -1051,3 +1052,592 @@ class TestEvdevKeyboardBackendGetTargetKeyCodes:
|
||||
|
||||
assert 42 in result # KEY_LEFTSHIFT
|
||||
assert 54 in result # KEY_RIGHTSHIFT
|
||||
|
||||
|
||||
class TestEvdevGrabAndForwarding:
|
||||
"""Test EVIOCGRAB + uinput pass-through that consumes shortcut keys (#871)."""
|
||||
|
||||
def _key_event(self, code: int, value: int = 1) -> MagicMock:
|
||||
return MagicMock(type=1, code=code, value=value) # EV_KEY
|
||||
|
||||
def _syn_event(self) -> MagicMock:
|
||||
return MagicMock(type=0, code=0) # EV_SYN / SYN_REPORT
|
||||
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
||||
def test_open_grabs_device_and_registers_forwarder(self, mock_input_device, mock_uinput):
|
||||
"""A keyboard that can be cloned is grabbed and paired with a forwarder."""
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
device.info.bustype = 0x03 # BUS_USB
|
||||
mock_input_device.return_value = device
|
||||
forwarder = MagicMock()
|
||||
mock_uinput.return_value = forwarder
|
||||
|
||||
backend = EvdevKeyboardBackend()
|
||||
opened = backend._open_keyboard_device("/dev/input/event0")
|
||||
|
||||
assert opened is True
|
||||
mock_uinput.assert_called_once()
|
||||
device.grab.assert_called_once()
|
||||
assert backend._forwarders[10] is forwarder
|
||||
assert device in backend.devices
|
||||
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
||||
def test_open_monitors_other_virtual_devices(self, mock_input_device, mock_uinput):
|
||||
"""Remapper/accessibility virtual devices are monitored like real ones.
|
||||
|
||||
A user can record a shortcut on a keyd/kmonad-style virtual keyboard;
|
||||
ignoring every BUS_VIRTUAL device would leave that shortcut dead.
|
||||
"""
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
device.info.bustype = 0x06 # BUS_VIRTUAL
|
||||
device.name = "keyd virtual keyboard"
|
||||
mock_input_device.return_value = device
|
||||
forwarder = MagicMock()
|
||||
forwarder.device.path = "/dev/input/event20"
|
||||
mock_uinput.return_value = forwarder
|
||||
|
||||
backend = EvdevKeyboardBackend()
|
||||
opened = backend._open_keyboard_device("/dev/input/event5")
|
||||
|
||||
assert opened is True
|
||||
device.grab.assert_called_once()
|
||||
assert device in backend.devices
|
||||
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
||||
def test_open_skips_own_clone_by_name(self, mock_input_device, mock_uinput):
|
||||
"""Our clones are skipped by name so forwarded events never loop back."""
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
device.info.bustype = 0x06 # BUS_VIRTUAL
|
||||
device.name = "AT Translated Set 2 keyboard (vocalinux)"
|
||||
mock_input_device.return_value = device
|
||||
|
||||
backend = EvdevKeyboardBackend()
|
||||
opened = backend._open_keyboard_device("/dev/input/event5")
|
||||
|
||||
assert opened is False
|
||||
device.close.assert_called_once()
|
||||
mock_uinput.assert_not_called()
|
||||
device.grab.assert_not_called()
|
||||
assert backend.devices == []
|
||||
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
||||
def test_open_skips_own_clone_by_path(self, mock_input_device, mock_uinput):
|
||||
"""A clone whose name was truncated is still skipped via its path."""
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
device.info.bustype = 0x06
|
||||
device.name = "AT Translated Set 2 keyboard (vocal" # truncated name
|
||||
mock_input_device.return_value = device
|
||||
|
||||
backend = EvdevKeyboardBackend()
|
||||
backend._clone_paths.add("/dev/input/event20")
|
||||
|
||||
opened = backend._open_keyboard_device("/dev/input/event20")
|
||||
|
||||
assert opened is False
|
||||
device.close.assert_called_once()
|
||||
mock_uinput.assert_not_called()
|
||||
device.grab.assert_not_called()
|
||||
assert backend.devices == []
|
||||
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
||||
def test_open_falls_back_when_uinput_unavailable(self, mock_input_device, mock_uinput):
|
||||
"""Without /dev/uinput the device still works ungrabbed (keys leak)."""
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
device.info.bustype = 0x03
|
||||
mock_input_device.return_value = device
|
||||
mock_uinput.side_effect = OSError("uinput denied")
|
||||
|
||||
backend = EvdevKeyboardBackend()
|
||||
opened = backend._open_keyboard_device("/dev/input/event0")
|
||||
|
||||
assert opened is True
|
||||
device.grab.assert_not_called()
|
||||
assert backend._forwarders == {}
|
||||
assert device in backend.devices
|
||||
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
||||
def test_open_falls_back_when_grab_fails(self, mock_input_device, mock_uinput):
|
||||
"""A failed grab drops the forwarder but keeps the listener running."""
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
device.info.bustype = 0x03
|
||||
device.grab.side_effect = OSError(errno.EBUSY, "device grabbed")
|
||||
mock_input_device.return_value = device
|
||||
forwarder = MagicMock()
|
||||
mock_uinput.return_value = forwarder
|
||||
|
||||
backend = EvdevKeyboardBackend()
|
||||
opened = backend._open_keyboard_device("/dev/input/event0")
|
||||
|
||||
assert opened is True
|
||||
forwarder.close.assert_called_once()
|
||||
assert backend._forwarders == {}
|
||||
assert device in backend.devices
|
||||
|
||||
def test_event_is_shortcut_pure_modifier(self) -> None:
|
||||
"""The configured modifier is consumed entirely; other keys are not."""
|
||||
backend = EvdevKeyboardBackend(shortcut="right_alt+right_alt") # KEY_RIGHTALT=100
|
||||
|
||||
assert backend._event_is_shortcut(10, self._key_event(100, 1)) is True
|
||||
assert backend._event_is_shortcut(10, self._key_event(100, 2)) is True
|
||||
assert backend._event_is_shortcut(10, self._key_event(100, 0)) is True
|
||||
assert backend._event_is_shortcut(10, self._key_event(56, 1)) is False # left alt
|
||||
assert backend._event_is_shortcut(10, self._key_event(30, 1)) is False # KEY_A
|
||||
assert backend._event_is_shortcut(10, self._syn_event()) is False
|
||||
|
||||
def test_event_is_shortcut_bare_function_key(self) -> None:
|
||||
"""A bare function-key shortcut consumes only that key."""
|
||||
backend = EvdevKeyboardBackend(shortcut="f11") # KEY_F11=87
|
||||
|
||||
assert backend._event_is_shortcut(10, self._key_event(87, 1)) is True
|
||||
assert backend._event_is_shortcut(10, self._key_event(87, 0)) is True
|
||||
assert backend._event_is_shortcut(10, self._key_event(88, 1)) is False # KEY_F12
|
||||
assert backend._event_is_shortcut(10, self._key_event(30, 1)) is False
|
||||
|
||||
def test_event_is_shortcut_combo_only_while_engaged(self) -> None:
|
||||
"""alt+r: the main key is consumed only while the modifier is held."""
|
||||
backend = EvdevKeyboardBackend(shortcut="alt+r")
|
||||
key_r = backend._combo_main_code
|
||||
assert key_r == 19 # KEY_R
|
||||
|
||||
backend._combo_pressed = {56} # KEY_LEFTALT held
|
||||
assert backend._event_is_shortcut(10, self._key_event(key_r, 1)) is True
|
||||
# Matching repeat and release stay consumed while the combo owns the
|
||||
# key — even if the modifier was already released first.
|
||||
backend._combo_pressed = set()
|
||||
assert backend._event_is_shortcut(10, self._key_event(key_r, 2)) is True
|
||||
assert backend._event_is_shortcut(10, self._key_event(key_r, 0)) is True
|
||||
assert backend._combo_swallowed == set()
|
||||
|
||||
# 'r' typed without alt still reaches apps; the modifier always passes
|
||||
assert backend._event_is_shortcut(10, self._key_event(key_r, 1)) is False
|
||||
assert backend._event_is_shortcut(10, self._key_event(key_r, 0)) is False
|
||||
assert backend._event_is_shortcut(10, self._key_event(56, 1)) is False
|
||||
assert backend._event_is_shortcut(10, self._key_event(100, 1)) is False
|
||||
|
||||
def test_event_is_shortcut_combo_tracks_presses_per_device(self) -> None:
|
||||
"""Two keyboards pressing the combo main key each pair their release."""
|
||||
backend = EvdevKeyboardBackend(shortcut="alt+r")
|
||||
key_r = backend._combo_main_code
|
||||
backend._combo_pressed = {56} # KEY_LEFTALT held
|
||||
|
||||
# Both keyboards press the main key; both presses are consumed.
|
||||
assert backend._event_is_shortcut(10, self._key_event(key_r, 1)) is True
|
||||
assert backend._event_is_shortcut(11, self._key_event(key_r, 1)) is True
|
||||
|
||||
# The first release must not clear the second keyboard's swallowed
|
||||
# press — otherwise its release would reach the app without a press.
|
||||
assert backend._event_is_shortcut(10, self._key_event(key_r, 0)) is True
|
||||
assert backend._event_is_shortcut(11, self._key_event(key_r, 0)) is True
|
||||
assert backend._combo_swallowed == set()
|
||||
|
||||
def test_reset_combo_state_keeps_other_devices_swallowed(self) -> None:
|
||||
"""A dropped-event reset on one device can't orphan another's press."""
|
||||
backend = EvdevKeyboardBackend(shortcut="alt+r")
|
||||
backend._combo_swallowed = {10, 11}
|
||||
|
||||
backend._reset_combo_state()
|
||||
|
||||
assert backend._combo_swallowed == {10, 11}
|
||||
|
||||
def test_forward_event_filters_feedback_types(self):
|
||||
"""Only real input events and SYN are re-emitted — never EV_LED."""
|
||||
backend = EvdevKeyboardBackend()
|
||||
forwarder = MagicMock()
|
||||
backend._forwarders = {10: forwarder}
|
||||
|
||||
backend._forward_event(10, self._key_event(30, 1))
|
||||
backend._forward_event(10, self._syn_event())
|
||||
backend._forward_event(10, MagicMock(type=0x11, code=0, value=1)) # EV_LED
|
||||
|
||||
assert forwarder.write_event.call_count == 2
|
||||
|
||||
def test_forward_event_noop_without_forwarder(self):
|
||||
"""Ungrabbed devices have no forwarder; nothing is written."""
|
||||
backend = EvdevKeyboardBackend()
|
||||
backend._forward_event(10, self._key_event(30, 1)) # no crash, no write
|
||||
|
||||
def test_monitor_consumes_shortcut_key_but_forwards_syn(self):
|
||||
"""Grabbed device: the shortcut press is handled locally, not re-emitted."""
|
||||
backend = EvdevKeyboardBackend() # default right_alt push-to-talk
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
forwarder = MagicMock()
|
||||
backend._forwarders = {10: forwarder}
|
||||
key_event = self._key_event(100, 1) # KEY_RIGHTALT press
|
||||
syn_event = self._syn_event()
|
||||
device.read.side_effect = lambda: setattr(backend, "running", False) or [
|
||||
key_event,
|
||||
syn_event,
|
||||
]
|
||||
backend.running = True
|
||||
backend.devices = [device]
|
||||
backend.device_fds = [10]
|
||||
backend._handle_key_event = MagicMock()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"vocalinux.ui.keyboard_backends.evdev_backend.time.monotonic",
|
||||
side_effect=[0.0, 0.0],
|
||||
),
|
||||
patch(
|
||||
"vocalinux.ui.keyboard_backends.evdev_backend.select.select",
|
||||
return_value=([10], [], []),
|
||||
),
|
||||
patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes,
|
||||
):
|
||||
mock_ecodes.EV_SYN = 0
|
||||
mock_ecodes.SYN_DROPPED = 3
|
||||
mock_ecodes.SYN_REPORT = 0
|
||||
mock_ecodes.EV_KEY = 1
|
||||
backend._monitor_devices()
|
||||
|
||||
backend._handle_key_event.assert_called_once_with(key_event, device)
|
||||
forwarder.write_event.assert_called_once_with(syn_event)
|
||||
|
||||
def test_monitor_forwards_non_shortcut_key(self):
|
||||
"""Grabbed device: an unrelated key is re-emitted, not consumed."""
|
||||
backend = EvdevKeyboardBackend()
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
forwarder = MagicMock()
|
||||
backend._forwarders = {10: forwarder}
|
||||
key_event = self._key_event(30, 1) # KEY_A
|
||||
syn_event = self._syn_event()
|
||||
device.read.side_effect = lambda: setattr(backend, "running", False) or [
|
||||
key_event,
|
||||
syn_event,
|
||||
]
|
||||
backend.running = True
|
||||
backend.devices = [device]
|
||||
backend.device_fds = [10]
|
||||
backend._handle_key_event = MagicMock()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"vocalinux.ui.keyboard_backends.evdev_backend.time.monotonic",
|
||||
side_effect=[0.0, 0.0],
|
||||
),
|
||||
patch(
|
||||
"vocalinux.ui.keyboard_backends.evdev_backend.select.select",
|
||||
return_value=([10], [], []),
|
||||
),
|
||||
patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes,
|
||||
):
|
||||
mock_ecodes.EV_SYN = 0
|
||||
mock_ecodes.SYN_DROPPED = 3
|
||||
mock_ecodes.SYN_REPORT = 0
|
||||
mock_ecodes.EV_KEY = 1
|
||||
backend._monitor_devices()
|
||||
|
||||
backend._handle_key_event.assert_called_once_with(key_event, device)
|
||||
assert forwarder.write_event.call_count == 2
|
||||
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
||||
def test_stop_closes_forwarders(self, mock_input_device, mock_find_devices, mock_uinput):
|
||||
"""stop() releases grabs (via close) and closes every uinput clone."""
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
device.info.bustype = 0x03
|
||||
mock_input_device.return_value = device
|
||||
forwarder = MagicMock()
|
||||
mock_uinput.return_value = forwarder
|
||||
mock_find_devices.return_value = ["/dev/input/event0"]
|
||||
|
||||
backend = EvdevKeyboardBackend()
|
||||
backend.start()
|
||||
backend.stop()
|
||||
|
||||
forwarder.close.assert_called_once()
|
||||
assert backend._forwarders == {}
|
||||
|
||||
def test_remove_device_closes_paired_forwarder(self):
|
||||
"""Disconnecting a keyboard closes its uinput clone too."""
|
||||
backend = EvdevKeyboardBackend()
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
forwarder = MagicMock()
|
||||
backend.devices = [device]
|
||||
backend.device_fds = [10]
|
||||
backend.device_paths = {"/dev/input/event0"}
|
||||
backend._device_paths_by_fd = {10: "/dev/input/event0"}
|
||||
backend._forwarders = {10: forwarder}
|
||||
|
||||
backend._remove_keyboard_device(10, device)
|
||||
|
||||
forwarder.close.assert_called_once()
|
||||
assert backend._forwarders == {}
|
||||
|
||||
def test_event_is_shortcut_without_spec(self) -> None:
|
||||
"""No parsed spec means nothing is consumed."""
|
||||
backend = EvdevKeyboardBackend()
|
||||
backend._spec = None
|
||||
|
||||
assert backend._event_is_shortcut(10, self._key_event(100, 1)) is False
|
||||
|
||||
def test_event_is_shortcut_combo_edges(self) -> None:
|
||||
"""Combo consumption only ever applies to the main key."""
|
||||
backend = EvdevKeyboardBackend(shortcut="alt+r")
|
||||
key_r = backend._combo_main_code
|
||||
|
||||
# Modifier and unrelated keys are never consumed, held or not.
|
||||
backend._combo_pressed = {56}
|
||||
assert backend._event_is_shortcut(10, self._key_event(56, 1)) is False
|
||||
assert backend._event_is_shortcut(10, self._key_event(30, 1)) is False
|
||||
# An unowned repeat is forwarded like any other event.
|
||||
assert backend._event_is_shortcut(10, self._key_event(key_r, 2)) is False
|
||||
# A release without a swallowed press is forwarded too.
|
||||
assert backend._event_is_shortcut(10, self._key_event(key_r, 0)) is False
|
||||
|
||||
# No resolvable main key -> nothing consumed.
|
||||
backend._combo_main_code = None
|
||||
backend._combo_pressed = {56}
|
||||
assert backend._event_is_shortcut(10, self._key_event(19, 1)) is False
|
||||
|
||||
def test_forward_event_logs_write_failure(self) -> None:
|
||||
"""A dead uinput clone logs the failure instead of crashing the loop."""
|
||||
backend = EvdevKeyboardBackend()
|
||||
forwarder = MagicMock()
|
||||
forwarder.write_event.side_effect = OSError("gone")
|
||||
backend._forwarders = {10: forwarder}
|
||||
|
||||
with patch("vocalinux.ui.keyboard_backends.evdev_backend.logger") as mock_logger:
|
||||
backend._forward_event(10, self._key_event(30, 1))
|
||||
|
||||
mock_logger.error.assert_called_once()
|
||||
assert backend._forwarders == {}
|
||||
forwarder.close.assert_called_once()
|
||||
|
||||
def test_forward_failure_releases_grab(self) -> None:
|
||||
"""A write failure on the clone releases EVIOCGRAB so keys reach apps."""
|
||||
backend = EvdevKeyboardBackend()
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
device.name = "Keyboard"
|
||||
forwarder = MagicMock()
|
||||
forwarder.write_event.side_effect = OSError("uinput gone")
|
||||
backend.devices = [device]
|
||||
backend.device_fds = [10]
|
||||
backend._forwarders = {10: forwarder}
|
||||
backend._forwarded_held = {10: set()}
|
||||
|
||||
backend._forward_event(10, self._key_event(30, 1))
|
||||
|
||||
device.ungrab.assert_called_once()
|
||||
forwarder.close.assert_called_once()
|
||||
assert backend._forwarders == {}
|
||||
assert 10 not in backend._forwarded_held
|
||||
# The device stays monitored, ungrabbed — the shortcut still detects.
|
||||
assert device in backend.devices
|
||||
|
||||
def test_forward_failure_removes_device_when_ungrab_fails(self) -> None:
|
||||
"""If the grab cannot be released, the device is closed outright."""
|
||||
backend = EvdevKeyboardBackend()
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
device.name = "Keyboard"
|
||||
device.ungrab.side_effect = OSError("ungrab failed")
|
||||
forwarder = MagicMock()
|
||||
forwarder.write_event.side_effect = OSError("uinput gone")
|
||||
backend.devices = [device]
|
||||
backend.device_fds = [10]
|
||||
backend.device_paths = {"/dev/input/event0"}
|
||||
backend._device_paths_by_fd = {10: "/dev/input/event0"}
|
||||
backend._forwarders = {10: forwarder}
|
||||
backend._forwarded_held = {10: set()}
|
||||
|
||||
backend._forward_event(10, self._key_event(30, 1))
|
||||
|
||||
# Closed rather than left grabbed with a dead clone.
|
||||
device.close.assert_called_once()
|
||||
assert backend.devices == []
|
||||
assert backend._forwarders == {}
|
||||
assert backend.device_paths == set()
|
||||
|
||||
def test_forward_event_tracks_held_keys(self) -> None:
|
||||
"""Forwarded presses/releases keep the clone's believed key state."""
|
||||
backend = EvdevKeyboardBackend()
|
||||
forwarder = MagicMock()
|
||||
backend._forwarders = {10: forwarder}
|
||||
backend._forwarded_held = {10: set()}
|
||||
|
||||
backend._forward_event(10, self._key_event(30, 1))
|
||||
backend._forward_event(10, self._key_event(56, 1))
|
||||
assert backend._forwarded_held[10] == {30, 56}
|
||||
|
||||
backend._forward_event(10, self._key_event(30, 0))
|
||||
assert backend._forwarded_held[10] == {56}
|
||||
|
||||
def test_resync_releases_phantom_keys_after_syn_dropped(self) -> None:
|
||||
"""A dropped release is rewritten to the clone from live key state."""
|
||||
backend = EvdevKeyboardBackend()
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
device.active_keys.return_value = [] # nothing actually held
|
||||
forwarder = MagicMock()
|
||||
backend._forwarders = {10: forwarder}
|
||||
backend._forwarded_held = {10: {30, 56}}
|
||||
|
||||
backend._resync_clone_key_state(10, device)
|
||||
|
||||
forwarder.write.assert_any_call(ecodes.EV_KEY, 30, 0)
|
||||
forwarder.write.assert_any_call(ecodes.EV_KEY, 56, 0)
|
||||
assert forwarder.write.call_count == 2
|
||||
assert backend._forwarded_held[10] == set()
|
||||
|
||||
def test_resync_keeps_still_held_keys(self) -> None:
|
||||
"""Keys still physically held are not released on the clone."""
|
||||
backend = EvdevKeyboardBackend()
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
device.active_keys.return_value = [30, 56]
|
||||
forwarder = MagicMock()
|
||||
backend._forwarders = {10: forwarder}
|
||||
backend._forwarded_held = {10: {30, 56}}
|
||||
|
||||
backend._resync_clone_key_state(10, device)
|
||||
|
||||
forwarder.write.assert_not_called()
|
||||
assert backend._forwarded_held[10] == {30, 56}
|
||||
|
||||
def test_resync_prunes_swallowed_combo_when_clone_held_nothing(self) -> None:
|
||||
"""A dropped combo release is reconciled even when the clone holds no keys.
|
||||
|
||||
The swallowed press was never forwarded, so ``_forwarded_held`` is
|
||||
empty — but its dropped release still needs the pairing pruned or
|
||||
the next ordinary press of that key is consumed too.
|
||||
"""
|
||||
backend = EvdevKeyboardBackend()
|
||||
backend._combo_main_code = 30
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
device.active_keys.return_value = [] # combo key physically up
|
||||
forwarder = MagicMock()
|
||||
backend._forwarders = {10: forwarder}
|
||||
backend._forwarded_held = {10: set()}
|
||||
backend._combo_swallowed = {10}
|
||||
|
||||
backend._resync_clone_key_state(10, device)
|
||||
|
||||
assert backend._combo_swallowed == set()
|
||||
forwarder.write.assert_not_called()
|
||||
|
||||
def test_resync_keeps_swallowed_combo_while_key_still_held(self) -> None:
|
||||
"""A swallowed combo press stays paired while the key is physically down."""
|
||||
backend = EvdevKeyboardBackend()
|
||||
backend._combo_main_code = 30
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
device.active_keys.return_value = [30]
|
||||
backend._combo_swallowed = {10}
|
||||
|
||||
backend._resync_clone_key_state(10, device)
|
||||
|
||||
assert backend._combo_swallowed == {10}
|
||||
|
||||
def test_monitor_exit_releases_grabs_on_select_failure(self) -> None:
|
||||
"""An unexpected monitor exit closes devices, releasing every grab."""
|
||||
backend = EvdevKeyboardBackend()
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
forwarder = MagicMock()
|
||||
backend.running = True
|
||||
backend.active = True
|
||||
backend.devices = [device]
|
||||
backend.device_fds = [10]
|
||||
backend.device_paths = {"/dev/input/event0"}
|
||||
backend._device_paths_by_fd = {10: "/dev/input/event0"}
|
||||
backend._forwarders = {10: forwarder}
|
||||
|
||||
with (
|
||||
patch(
|
||||
"vocalinux.ui.keyboard_backends.evdev_backend.time.monotonic",
|
||||
side_effect=[0.0, 0.0],
|
||||
),
|
||||
patch(
|
||||
"vocalinux.ui.keyboard_backends.evdev_backend.select.select",
|
||||
side_effect=OSError("select failed"),
|
||||
),
|
||||
):
|
||||
backend._monitor_devices()
|
||||
|
||||
device.close.assert_called_once()
|
||||
forwarder.close.assert_called_once()
|
||||
assert backend.devices == []
|
||||
assert backend._forwarders == {}
|
||||
assert backend.running is False
|
||||
assert backend.active is False
|
||||
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
|
||||
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
||||
def test_open_device_race_closes_device_and_forwarder(self, mock_input_device, mock_uinput):
|
||||
"""A device that loses the registration race is cleaned up fully."""
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
device.info.bustype = 0x03
|
||||
mock_input_device.return_value = device
|
||||
forwarder = MagicMock()
|
||||
mock_uinput.return_value = forwarder
|
||||
|
||||
backend = EvdevKeyboardBackend()
|
||||
backend.device_fds = [10] # same fd won by another open
|
||||
|
||||
opened = backend._open_keyboard_device("/dev/input/event7")
|
||||
|
||||
assert opened is False
|
||||
device.close.assert_called_once()
|
||||
forwarder.close.assert_called_once()
|
||||
assert backend.devices == []
|
||||
assert backend._forwarders == {}
|
||||
|
||||
def test_monitor_forwards_survivors_during_syn_dropped(self):
|
||||
"""Mid-drop: shortcut keys are eaten, surviving other keys forwarded."""
|
||||
backend = EvdevKeyboardBackend()
|
||||
device = MagicMock()
|
||||
device.fileno.return_value = 10
|
||||
forwarder = MagicMock()
|
||||
backend._forwarders = {10: forwarder}
|
||||
backend._dropped_devices = {10}
|
||||
shortcut_event = self._key_event(100, 1) # KEY_RIGHTALT press
|
||||
other_event = self._key_event(30, 1) # KEY_A
|
||||
device.read.side_effect = lambda: setattr(backend, "running", False) or [
|
||||
shortcut_event,
|
||||
other_event,
|
||||
]
|
||||
backend.running = True
|
||||
backend.devices = [device]
|
||||
backend.device_fds = [10]
|
||||
backend._handle_key_event = MagicMock()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"vocalinux.ui.keyboard_backends.evdev_backend.time.monotonic",
|
||||
side_effect=[0.0, 0.0],
|
||||
),
|
||||
patch(
|
||||
"vocalinux.ui.keyboard_backends.evdev_backend.select.select",
|
||||
return_value=([10], [], []),
|
||||
),
|
||||
patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes,
|
||||
):
|
||||
mock_ecodes.EV_SYN = 0
|
||||
mock_ecodes.SYN_DROPPED = 3
|
||||
mock_ecodes.SYN_REPORT = 0
|
||||
mock_ecodes.EV_KEY = 1
|
||||
backend._monitor_devices()
|
||||
|
||||
backend._handle_key_event.assert_not_called()
|
||||
forwarder.write_event.assert_called_once_with(other_event)
|
||||
|
||||
Reference in New Issue
Block a user