d11bd1bf0d
* feat(post-processing): pipe transcriptions through user-defined script Add a configurable post-processing step that passes each transcription result to an external executable via stdin and uses its stdout as the text to type. Enables custom transforms (formatting, LLM rewriting, etc.) without modifying Vocalinux. Falls back to original text on script failure or timeout. * test(post-processor): add unit tests and extract apply_post_processing Move config-reading logic from main.py closure into apply_post_processing() in post_processor.py so it can be tested independently. Add 11 unit tests covering PostProcessor.process() and apply_post_processing(). * docs(post-processing): document post-processing script feature * fix formatting of one file * fix(post-processing): run the user script with a stripped host environment Rebase compat: the host_process AST guard requires every subprocess spawned for a host binary to pass env=host_env() so an AppImage's LD_LIBRARY_PATH/GI_TYPELIB_PATH/PYTHONHOME do not leak in. Also give apply_post_processing() a typed ConfigManager parameter and keep text_callback_wrapper's local variable typed as str for mypy. * fix(post-processing): preserve paragraph breaks and tighten script handling Take script stdout verbatim instead of rstripping every trailing newline, so the "new paragraph" voice command's break survives a passthrough script. A single trailing newline is still dropped, but only when the transcription had none — it is almost always just the line terminator echo/print-style tools append. Also stop writing transformed output to the log (it can carry sensitive dictation into retained logs), catch only the expected process/OS/encoding failures instead of every Exception, and add the missing signature annotations. * fix(main): run the post-processing script off the recognition thread A user script may take seconds per segment, and text callbacks run on the recognition thread that drains the bounded audio-segment queue; stalling it drops queued dictation once the queue fills. Submit post-processing and injection to a dedicated single-worker executor so the backlog waits instead of losing speech, while keeping injection order. On timeout or failure the worker falls back to the original text, and an unexpected error is logged with a traceback and still injects the original rather than silently dropping the segment. Also pin the callback tests' get_str mock to no-script so they exercise the synchronous path deterministically, and add a callback test proving transformed text (including preserved paragraph breaks) reaches the injector. * docs(post-processing): annotate new callbacks and clarify stdout contract Add the missing type annotations on the new Settings page callbacks per the repository guide, and document that script stdout is used verbatim (a lone trailing newline added by line-oriented tools is dropped only when the transcription had none) and that scripts run on a dedicated worker. * test(main): annotate new post-processing callback tests * fix(main): bind queued dictation to its target app and drain worker on quit Route every transcription segment through the single post-processing worker (a pass-through when no script is configured) so clearing the script path mid-queue can no longer let a later segment overtake an earlier one still running. A segment that can sit in the backlog — a script is configured or a job is still running — is bound to the application focused when it was dictated; the worker drops it if focus has verifiably moved to another application, so delayed dictation cannot land in the wrong app. Wire the worker into the tray quit path via an on_quit hook: it clears the acceptance flag, cancels pending jobs, and waits — bounded by the script's own timeout — so nothing injects after the injector stops. * fix(main): keep voice actions ordered and probe focus off the recognition thread - Queue voice-command actions on the post-processing worker so an action can no longer overtake dictated text still waiting to be injected. - Move the submit-time focus probe onto a dedicated executor; the recognition thread only submits the probe, and the job joins the result at inject time, so compositor tools can no longer stall consumption of the bounded audio-segment queue. - Re-check the quit flag after the inject-time focus re-probe, then shut both workers down without waiting on a running script so quitting does not freeze on the script timeout. * fix(main): capture submit-time focus on dedicated threads and serialize quit - Replace the shared focus-probe executor with a fresh daemon thread per submission so a slow earlier probe cannot delay a later one past the focus change it is meant to capture; jobs read the result off a one-slot queue at inject time. - Give queued voice actions the same submit-time focus binding as text segments so a queued shortcut is dropped instead of fired into a newly focused application. - Hold injection_lock across the accept-check plus inject/action call so quit waits only for an in-flight injection instead of racing it. - Queue the IDLE reset of last_injected_text behind pending worker jobs so a queued 'delete that' still sees the text it targets. * fix(main): keep actions instant on a free worker and bound quit wait - An action now starts a submit-time focus probe only when it actually queues behind other jobs; with a free worker there is no delay window to bind against, so the probe's compositor calls are skipped and the action stays immediate even when a script is configured. - Probe and executor submission now happen inside pending_jobs_lock so queue order matches callback order — a job that saw an empty queue can no longer slip behind one that arrived while it was submitting. - Every inject-path subprocess call (xdotool window/key/type, wtype and ydotool type) is bounded by a timeout, and the quit hook's injection_lock wait is capped, so quitting can no longer freeze on a stalled injector subprocess. * fix(main): bind every action to submit-time focus without slowing it down - Every voice action now starts a submit-time focus probe: submission never guarantees immediate execution, so skipping the binding left a focus-change window in which a shortcut could fire into another app. The worker waits on the probe for at most one second — milliseconds on a healthy desktop — so immediate actions stay immediate while a genuinely stalled probe falls back to the injector's own targeting. - Quit waits for an in-flight injection without a cap again: every inject-path subprocess is individually bounded, so the wait always ends with the injection and a long transcription is never cut off. - The wtype/ydotool typing timeout is now derived from the configured per-key delay (a generous multiple), so a legitimately slow type can never be cut mid-text; a timed-out type no longer falls through to the clipboard fallback, which would duplicate the partial fragment. * fix(main): drop unverified actions and make quit-time injection abortable - An action whose submit-time focus probe cannot answer inside its one-second bound is now dropped instead of running unverified: a stalled probe can no longer leave a shortcut pointing at whatever application happens to be focused. - Quit asks the injector to abort in-flight injections, and injection paths honour it: the xdotool chunk loop stops between chunks, the wtype/ydotool type call runs as a polling Popen that can be killed, and paste/shortcut/backspace entry points exit early. The quit hook still waits for an in-flight injection, but the wait is bounded by at most one already-running subprocess call — never the injection's full length — and no subprocess is ever stopped mid-call by a timeout cap on the wait itself. - Annotate the remaining untyped test helpers. * fix(main): wait out slow focus probes; restore run() in wayland typing - An action no longer races its submit-time focus probe on a one-second bound: the job waits for the probe's own answer, whose compositor calls already carry per-call timeouts and short-circuit absent tools. A slow but living probe delays the command instead of discarding it, so a valid "undo" can no longer be lost to timing alone, and focus verification still decides every drop. - Wayland typing goes back to subprocess.run so tests keep intercepting the same entry point. Abortability is preserved by typing in 200-character chunks with the shutdown check between calls — a whole-string call could not be interrupted, but no single chunk holds the injector for more than its own scaled budget. - The abort flag gains a class-level default so injectors built without __init__ in tests still answer the checks; each real instance gets its own event in __init__. - Update call-shape assertions for the typed-call timeout and finish annotating the test methods touched by the callback refactor. * fix(inject): never retype text a failed chunked call already sent - _PartiallyTyped carries the confirmed-delivered prefix length out of the chunked type loops, so every downstream fallback continues from the untyped remainder: the Wayland-to-xdotool switch types only the tail, and the clipboard fallback copies only the tail, never the whole string a second time. - The xdotool retry loop gains the same accounting — a retry now resumes at the first untyped chunk instead of restarting at the top and duplicating everything the failed attempt already delivered. - Annotate the remaining test signature the repository guide requires. * fix(inject): report partial typing as failure even when clipboard lands A chunked type call that dies mid-text leaves only a prefix on screen, and the remainder-fallback keeps it that way: it copies the tail for a manual paste but now returns False, so the caller no longer records the full transcription as injected and a following "delete that" cannot erase text that was never typed. * fix(inject): keep deletion state aligned with delivered text inject_text now records how many characters a call confirmed on screen as last_typed_count: the whole length on success, the typed prefix after a partial failure, -1 when the delivered amount is unknowable. The transcription caller uses it to keep last_injected_text honest — a following "delete that" erases the typed prefix or nothing, never the previous segment or text that never arrived. * fix(inject): never replay input a timed-out portal job may have delivered - RemoteDesktopPortalError.delivered=None marks an unreported job's count as unknowable; all three portal fallbacks demote and skip instead of replaying, and text injection surfaces as TimeoutExpired so no full-text clipboard copy is offered - _PartiallyTyped counts raised after a portal demote are rebased onto the original text (typed_prefix) so the caller clips the real prefix - evdev: a consumed pure-modifier press is withheld and replayed ahead of the first non-target press, so holding RightAlt for AltGr composition still reaches the application - history: extend_latest gains an expected_latest CAS guard so a late segment from an older worker merges only into its own session's snippet, never into a newer session's entry - quit path aborts in-flight injections before waiting on the lock so quitting cannot freeze behind a long transcription's typing budget - tests: stop _ensure_ydotoold leaking a real Popen in Wayland tests * fix(shortcut): replay withheld modifiers across devices and after SYN_DROPPED - Replay withheld modifier presses from every device, not only the fd the character arrived on: split keyboards deliver AltGr and the character on separate devices, and a per-fd check forwarded a plain character. - Let the post-SYN_DROPPED resync run when only withheld modifiers exist for the fd, so a press whose release the dropped burst ate is pruned instead of replaying a stale modifier on the next ordinary key. * fix(history,pads): keep late segments pad-bound deliveries and pad deletions correct - Track the worker behind the ended session's snippet so late segments from an older worker cannot merge into a newer session's entry - Skip the focus-change drop for dictation-pad segments and pad-bound editing commands, which never target the focused application - Refuse 'delete that' when a manual pad edit blurred the segment boundaries instead of deleting the stale recorded length * style: drop unneeded nonlocal for mutated worker set * fix(shortcut): share one evdev device layer across shortcut listeners Each KeyboardShortcutManager built its own EvdevKeyboardBackend, and on evdev every backend opened and grabbed the same physical keyboards: the first grabber received all events and the rest received none, so per-language shortcuts could never fire on Wayland. All backends now bind to a process-wide EvdevDeviceHub that opens each keyboard once, owns the grabs and the paired uinput clones, and fans every event out to each registered engine. An event is re-emitted to applications only when no engine consumed it, so suppression semantics stay the same while per-engine matching state (combo progress, withheld modifiers, swallowed presses) remains independent. * fix(history): own late segments per worker and decide pad routing once - ended_worker_snippets maps each ended session's worker to its own (snippet, commit epoch): late deliveries merge into their session's entry however many sessions have committed since, so one session's stragglers never split across Recent Snippets entries. - The leftover-segment commit reads the live history epoch inside the session lock: a mid-session clear() no longer makes the stale ended epoch refuse valid post-clear dictation. - The pad-bound routing decision is captured once per job and passed to the injector/action handler, so a toggle flip between the focus check and delivery cannot reroute text or shortcuts into the focused app. - Dead recognition workers are pruned from the tracking set and map so long runs don't accumulate finished threads. - The custom-terms switch now suppresses GTK's own state-set on save failure so it shows the value that actually persisted. * fix(shortcut): serialize evdev device layer teardown against restart The last listener's unregister released the engine lock before joining the monitor thread and closing devices, so a new listener registering in that gap opened keyboards the stale teardown then closed — it stayed active but dead, and briefly grabbed keyboards stopped forwarding input. A lifecycle lock now covers the whole close/open transition in register, unregister, and the monitor's unexpected-exit teardown, so a cold start can only open devices after the previous generation finished closing (also making the new generation's grabs succeed). A generation stamp tells a monitor thread that outlived its join() its incarnation is over even when running has gone True again. * fix(injection): log focus-probe failures and narrow clipboard fallback catch A crashed focus probe turned into an unknown result with no logged error, indistinguishable from unavailable focus information; the partial-typing clipboard fallback caught every exception. Log the probe failure with context and narrow the clipboard catch to the exceptions the fallback path can actually raise. * fix(shortcut): join the evdev monitor outside the lifecycle lock If the monitor exited unexpectedly while the last listener unregistered, its cleanup queued on the lifecycle lock that unregister() held for the join — stalling shutdown for the join's full 2s timeout. The close now finishes inside the lock and the join waits after it is released, so the monitor's own cleanup can run and exit instead of queueing behind it. Closing before the join also dies the monitor's read/select on the closed fds right away. A _closed gate on _open_keyboard_device keeps a rescan from opening a device between detach and close that nothing would ever close. * fix(injection): narrow focus-probe catch to the raisable exception set get_focused_window() reports Optional and never raises; the catch now covers only what can realistically escape the queue write and logs it with context, keeping the waiting job released. * fix(shortcut): tag evdev device fds with the generation that opened them A monitor that outlived join() could still touch a newer generation's devices: the loop re-checked the generation only at the top, so mid-iteration it could dispatch from, forward to, or remove devices a restarted listener had just opened. Every fd is now tagged with the generation that opened it, and the tag is verified at each point a monitor could mutate shared state — fd lookup for dispatch, clone forwarding, forwarder release, device removal, and rescan opens — while dispatch re-checks the generation between events and the per-fd loop breaks on it. Cold starts bump the generation before opening so fresh devices are tagged for the new generation. Devices an older generation opened stay tagged stale, so its reader leaves them alone. --------- Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
2053 lines
84 KiB
Python
2053 lines
84 KiB
Python
"""
|
|
Tests for evdev keyboard backend.
|
|
|
|
This module tests:
|
|
- find_keyboard_devices() function
|
|
- device_has_modifier_key() function
|
|
- EvdevKeyboardBackend class initialization
|
|
- start() and stop() methods
|
|
- is_available() and get_permission_hint()
|
|
- _handle_key_event() method
|
|
- _monitor_devices() thread
|
|
"""
|
|
|
|
import errno
|
|
import os
|
|
import threading
|
|
import time
|
|
from unittest.mock import MagicMock, Mock, mock_open, patch
|
|
|
|
import pytest
|
|
|
|
from vocalinux.ui.keyboard_backends.evdev_backend import (
|
|
DEVICE_RESCAN_SECONDS,
|
|
EVDEV_AVAILABLE,
|
|
MODIFIER_KEY_CODES,
|
|
EvdevDeviceHub,
|
|
EvdevKeyboardBackend,
|
|
device_has_key,
|
|
device_has_modifier_key,
|
|
ecodes,
|
|
find_keyboard_devices,
|
|
)
|
|
|
|
|
|
class TestFindKeyboardDevices:
|
|
"""Test find_keyboard_devices() function."""
|
|
|
|
def test_find_keyboard_devices_no_devices(self):
|
|
"""Test when no keyboard devices are found."""
|
|
mock_proc_content = """I: Bus=0011 Vendor=0001 Product=0001 Version=ab83
|
|
N: Name="Power Button"
|
|
H: Handlers=
|
|
"""
|
|
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
|
|
with patch("os.path.exists", return_value=False):
|
|
result = find_keyboard_devices()
|
|
assert result == []
|
|
|
|
def test_find_keyboard_devices_single_device(self):
|
|
"""Test finding a single keyboard device."""
|
|
mock_proc_content = """I: Bus=0011 Vendor=0001 Product=0001 Version=ab83
|
|
N: Name="AT Translated Set 2 keyboard"
|
|
H: Handlers=sysrq kbd event0
|
|
B: KEY=10000 7ff 202100 3953b001 68ffe0 1 20000 2000000000000 0
|
|
"""
|
|
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
|
|
with patch("os.path.exists", return_value=True):
|
|
result = find_keyboard_devices()
|
|
assert "/dev/input/event0" in result
|
|
|
|
def test_find_keyboard_devices_multiple_devices(self):
|
|
"""Test finding multiple keyboard devices."""
|
|
mock_proc_content = """I: Bus=0011 Vendor=0001 Product=0001 Version=ab83
|
|
N: Name="AT Translated Set 2 keyboard"
|
|
H: Handlers=sysrq kbd event0
|
|
B: KEY=10000 7ff 202100 3953b001 68ffe0 1 20000 2000000000000 0
|
|
I: Bus=0018 Vendor=04f3 Product=0033 Version=0500
|
|
N: Name="Elan Touchpad"
|
|
H: Handlers=mouse0 event1
|
|
B: KEY=ff000000000000 0 0 0
|
|
"""
|
|
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
|
|
with patch("os.path.exists", return_value=True):
|
|
result = find_keyboard_devices()
|
|
# Should find event0 (has KEY capability)
|
|
assert len(result) >= 1
|
|
|
|
def test_find_keyboard_devices_io_error(self):
|
|
"""Test error handling when /proc/bus/input/devices cannot be read."""
|
|
with patch("builtins.open", side_effect=IOError("Permission denied")):
|
|
with patch(
|
|
"vocalinux.ui.keyboard_backends.evdev_backend._find_keyboard_devices_from_evdev",
|
|
return_value=[],
|
|
):
|
|
result = find_keyboard_devices()
|
|
assert result == []
|
|
|
|
def test_find_keyboard_devices_file_not_found(self):
|
|
"""Test error handling when /proc/bus/input/devices doesn't exist."""
|
|
with patch("builtins.open", side_effect=FileNotFoundError()):
|
|
with patch(
|
|
"vocalinux.ui.keyboard_backends.evdev_backend._find_keyboard_devices_from_evdev",
|
|
return_value=[],
|
|
):
|
|
result = find_keyboard_devices()
|
|
assert result == []
|
|
|
|
def test_find_keyboard_devices_falls_back_to_evdev_list(self):
|
|
"""When /proc is denied (e.g. snap), discover keyboards via list_devices."""
|
|
with patch("builtins.open", side_effect=PermissionError(13, "Permission denied")):
|
|
with patch(
|
|
"vocalinux.ui.keyboard_backends.evdev_backend._find_keyboard_devices_from_evdev",
|
|
return_value=["/dev/input/event10"],
|
|
) as fallback:
|
|
result = find_keyboard_devices()
|
|
assert result == ["/dev/input/event10"]
|
|
fallback.assert_called_once()
|
|
|
|
def test_find_keyboard_devices_no_keyboard_capability(self):
|
|
"""Test filtering devices without keyboard capability."""
|
|
mock_proc_content = """I: Bus=0019 Vendor=0000 Product=0003 Version=0000
|
|
N: Name="Power Button"
|
|
H: Handlers=power kbd event2
|
|
B: KEY=0
|
|
"""
|
|
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
|
|
result = find_keyboard_devices()
|
|
# Device with KEY=0 should not be included
|
|
assert "/dev/input/event2" not in result
|
|
|
|
def test_find_keyboard_devices_no_event_handler(self):
|
|
"""Test filtering devices without event handler."""
|
|
mock_proc_content = """I: Bus=0011 Vendor=0001 Product=0001 Version=ab83
|
|
N: Name="Device with no event"
|
|
H: Handlers=kbd
|
|
B: KEY=10000 7ff 202100 3953b001 68ffe0 1 20000 2000000000000 0
|
|
"""
|
|
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
|
|
result = find_keyboard_devices()
|
|
# Device without eventX handler should not be included
|
|
assert result == []
|
|
|
|
def test_find_keyboard_devices_device_not_exists(self):
|
|
"""Test filtering devices that don't actually exist in /dev/input."""
|
|
mock_proc_content = """I: Bus=0011 Vendor=0001 Product=0001 Version=ab83
|
|
N: Name="AT Translated Set 2 keyboard"
|
|
H: Handlers=sysrq kbd event0
|
|
B: KEY=10000 7ff 202100 3953b001 68ffe0 1 20000 2000000000000 0
|
|
"""
|
|
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
|
|
with patch("os.path.exists", return_value=False):
|
|
result = find_keyboard_devices()
|
|
assert "/dev/input/event0" not in result
|
|
|
|
|
|
class TestDeviceHasModifierKey:
|
|
"""Test device_has_modifier_key() function."""
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", False)
|
|
def test_device_has_modifier_key_evdev_not_available(self):
|
|
"""Test returns False when evdev is not available."""
|
|
result = device_has_modifier_key("/dev/input/event0", "ctrl")
|
|
assert result is False
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_device_has_modifier_key_has_ctrl(self, mock_input_device):
|
|
"""Test device with ctrl modifier key capability."""
|
|
mock_device = MagicMock()
|
|
mock_device.capabilities.return_value = {
|
|
1: [29, 97] # EV_KEY: [KEY_LEFTCTRL, KEY_RIGHTCTRL]
|
|
}
|
|
mock_input_device.return_value = mock_device
|
|
|
|
# Mock ecodes
|
|
with patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes:
|
|
mock_ecodes.EV_KEY = 1
|
|
result = device_has_modifier_key("/dev/input/event0", "ctrl")
|
|
assert result is True
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_device_has_modifier_key_missing_modifier(self, mock_input_device):
|
|
"""Test device without specific modifier key capability."""
|
|
mock_device = MagicMock()
|
|
mock_device.capabilities.return_value = {
|
|
1: [1, 2, 3] # EV_KEY with different keys (not ctrl)
|
|
}
|
|
mock_input_device.return_value = mock_device
|
|
|
|
with patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes:
|
|
mock_ecodes.EV_KEY = 1
|
|
result = device_has_modifier_key("/dev/input/event0", "ctrl")
|
|
assert result is False
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_device_has_modifier_key_alt(self, mock_input_device):
|
|
"""Test device with alt modifier key capability."""
|
|
mock_device = MagicMock()
|
|
mock_device.capabilities.return_value = {
|
|
1: [56, 100] # EV_KEY: [KEY_LEFTALT, KEY_RIGHTALT]
|
|
}
|
|
mock_input_device.return_value = mock_device
|
|
|
|
with patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes:
|
|
mock_ecodes.EV_KEY = 1
|
|
result = device_has_modifier_key("/dev/input/event0", "alt")
|
|
assert result is True
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_device_has_modifier_key_left_ctrl_specific(self, mock_input_device):
|
|
"""Test device with left_ctrl specifically."""
|
|
mock_device = MagicMock()
|
|
mock_device.capabilities.return_value = {1: [29]} # EV_KEY: [KEY_LEFTCTRL only]
|
|
mock_input_device.return_value = mock_device
|
|
|
|
with patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes:
|
|
mock_ecodes.EV_KEY = 1
|
|
result = device_has_modifier_key("/dev/input/event0", "left_ctrl")
|
|
assert result is True
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_device_has_modifier_key_os_error(self, mock_input_device):
|
|
"""Test handling of OSError when opening device."""
|
|
mock_input_device.side_effect = OSError("Permission denied")
|
|
|
|
result = device_has_modifier_key("/dev/input/event0", "ctrl")
|
|
assert result is False
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_device_has_modifier_key_invalid_modifier(self, mock_input_device):
|
|
"""Test with invalid modifier name."""
|
|
result = device_has_modifier_key("/dev/input/event0", "invalid")
|
|
assert result is False
|
|
|
|
|
|
class TestDeviceHasKey:
|
|
"""Test device_has_key() against EV_KEY capabilities."""
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_device_has_key_f10(self, mock_input_device):
|
|
mock_device = MagicMock()
|
|
mock_device.capabilities.return_value = {1: [68]} # KEY_F10
|
|
mock_input_device.return_value = mock_device
|
|
|
|
with patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes:
|
|
mock_ecodes.EV_KEY = 1
|
|
mock_ecodes.KEY_F10 = 68
|
|
assert device_has_key("/dev/input/event0", "f10") is True
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_device_has_key_missing(self, mock_input_device):
|
|
mock_device = MagicMock()
|
|
mock_device.capabilities.return_value = {1: [1, 2, 3]}
|
|
mock_input_device.return_value = mock_device
|
|
|
|
with patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes:
|
|
mock_ecodes.EV_KEY = 1
|
|
mock_ecodes.KEY_F10 = 68
|
|
assert device_has_key("/dev/input/event0", "f10") is False
|
|
|
|
|
|
class TestEvdevKeyboardBackendInit:
|
|
"""Test EvdevKeyboardBackend initialization."""
|
|
|
|
def test_init_default_values(self):
|
|
"""Test initialization with default values."""
|
|
backend = EvdevKeyboardBackend()
|
|
assert backend.shortcut == "right_alt+right_alt"
|
|
assert backend.mode == "push_to_talk"
|
|
assert backend.active is False
|
|
assert backend.devices == []
|
|
assert backend.device_fds == []
|
|
assert backend.running is False
|
|
assert backend.monitor_thread is None
|
|
|
|
def test_init_custom_shortcut(self):
|
|
"""Test initialization with custom shortcut."""
|
|
backend = EvdevKeyboardBackend(shortcut="alt+alt")
|
|
assert backend.shortcut == "alt+alt"
|
|
assert backend.modifier_key == "alt"
|
|
|
|
def test_init_custom_mode(self):
|
|
"""Test initialization with custom mode."""
|
|
backend = EvdevKeyboardBackend(mode="push_to_talk")
|
|
assert backend.mode == "push_to_talk"
|
|
|
|
def test_init_double_tap_threshold(self):
|
|
"""Test that double tap threshold is initialized."""
|
|
backend = EvdevKeyboardBackend()
|
|
assert backend.double_tap_threshold == 0.3
|
|
|
|
def test_init_left_shift_shortcut(self):
|
|
"""Test initialization with left_shift shortcut."""
|
|
backend = EvdevKeyboardBackend(shortcut="left_shift+left_shift")
|
|
assert backend.modifier_key == "left_shift"
|
|
|
|
def test_init_key_pressed_devices_empty(self):
|
|
"""Test that key_pressed_devices starts empty."""
|
|
backend = EvdevKeyboardBackend()
|
|
assert backend.key_pressed_devices == set()
|
|
|
|
|
|
class TestEvdevKeyboardBackendIsAvailable:
|
|
"""Test is_available() method."""
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", False)
|
|
def test_is_available_evdev_not_available(self):
|
|
"""Test is_available returns False when evdev not available."""
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.is_available()
|
|
assert result is False
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
def test_is_available_no_devices_found(self, mock_find_devices):
|
|
"""Test is_available returns False when no devices found."""
|
|
mock_find_devices.return_value = []
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.is_available()
|
|
assert result is False
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.device_has_modifier_key")
|
|
def test_is_available_device_with_modifier(self, mock_has_mod, mock_find_devices):
|
|
"""Test is_available returns True when device with modifier found."""
|
|
mock_find_devices.return_value = ["/dev/input/event0"]
|
|
mock_has_mod.return_value = True
|
|
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.is_available()
|
|
assert result is True
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.device_has_modifier_key")
|
|
def test_is_available_no_device_with_modifier(self, mock_has_mod, mock_find_devices):
|
|
"""Test is_available returns False when no device has required modifier."""
|
|
mock_find_devices.return_value = ["/dev/input/event0"]
|
|
mock_has_mod.return_value = False
|
|
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.is_available()
|
|
assert result is False
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
def test_is_available_exception_handling(self, mock_find_devices):
|
|
"""Test is_available handles exceptions gracefully."""
|
|
mock_find_devices.side_effect = Exception("Test error")
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.is_available()
|
|
assert result is False
|
|
|
|
|
|
class TestEvdevKeyboardBackendPermissionHint:
|
|
"""Test get_permission_hint() method."""
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", False)
|
|
def test_permission_hint_evdev_not_available(self):
|
|
"""Test permission hint when evdev not installed."""
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.get_permission_hint()
|
|
assert "pip install" in result or "evdev" in result
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
def test_permission_hint_no_devices(self, mock_find_devices):
|
|
"""Test permission hint when no devices found."""
|
|
mock_find_devices.return_value = []
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.get_permission_hint()
|
|
assert result is None
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_permission_hint_permission_denied(self, mock_input_device, mock_find_devices):
|
|
"""Test permission hint when permission denied."""
|
|
mock_find_devices.return_value = ["/dev/input/event0"]
|
|
error = OSError(errno.EACCES, "Permission denied")
|
|
mock_input_device.side_effect = error
|
|
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.get_permission_hint()
|
|
assert result is not None
|
|
assert "input" in result or "group" in result
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_permission_hint_device_accessible(self, mock_input_device, mock_find_devices):
|
|
"""Test permission hint when device is accessible."""
|
|
mock_find_devices.return_value = ["/dev/input/event0"]
|
|
mock_device = MagicMock()
|
|
mock_input_device.return_value = mock_device
|
|
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.get_permission_hint()
|
|
assert result is None
|
|
|
|
@patch.dict("os.environ", {"SNAP": "/snap/vocalinux/x1"}, clear=False)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
def test_permission_hint_snap_no_devices(self, mock_find_devices):
|
|
"""Snap with no listed devices should ask for both input plugs."""
|
|
mock_find_devices.return_value = []
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.get_permission_hint()
|
|
assert result is not None
|
|
assert "raw-input" in result
|
|
assert "hardware-observe" in result
|
|
|
|
@patch.dict("os.environ", {"SNAP": "/snap/vocalinux/x1"}, clear=False)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_permission_hint_snap_permission_denied(self, mock_input_device, mock_find_devices):
|
|
"""Snap that cannot open /dev/input should ask for both input plugs."""
|
|
mock_find_devices.return_value = ["/dev/input/event0"]
|
|
mock_input_device.side_effect = OSError(errno.EACCES, "Permission denied")
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.get_permission_hint()
|
|
assert result is not None
|
|
assert "raw-input" in result
|
|
assert "hardware-observe" in result
|
|
|
|
|
|
class TestEvdevKeyboardBackendStart:
|
|
"""Test start() method."""
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", False)
|
|
def test_start_evdev_not_available(self):
|
|
"""Test start returns False when evdev not available."""
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.start()
|
|
assert result is False
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
def test_start_no_devices(self, mock_find_devices):
|
|
"""Test start returns False when no devices found."""
|
|
mock_find_devices.return_value = []
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.start()
|
|
assert result is False
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_start_successful(self, mock_input_device, mock_find_devices):
|
|
"""Test successful start."""
|
|
mock_find_devices.return_value = ["/dev/input/event0"]
|
|
mock_device = MagicMock()
|
|
mock_device.fileno.return_value = 10
|
|
mock_input_device.return_value = mock_device
|
|
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.start()
|
|
|
|
assert result is True
|
|
assert backend.active is True
|
|
assert backend.running is True
|
|
assert len(backend.devices) == 1
|
|
assert 10 in backend.device_fds
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
def test_start_already_active(self):
|
|
"""Test that start returns True if already active."""
|
|
backend = EvdevKeyboardBackend()
|
|
backend.active = True
|
|
|
|
result = backend.start()
|
|
assert result is True
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_start_creates_monitor_thread(self, mock_input_device, mock_find_devices):
|
|
"""Test that monitor thread is created."""
|
|
mock_find_devices.return_value = ["/dev/input/event0"]
|
|
mock_device = MagicMock()
|
|
mock_device.fileno.return_value = 10
|
|
mock_input_device.return_value = mock_device
|
|
|
|
backend = EvdevKeyboardBackend()
|
|
backend.start()
|
|
|
|
assert backend.monitor_thread is not None
|
|
assert isinstance(backend.monitor_thread, threading.Thread)
|
|
|
|
# Clean up
|
|
backend.stop()
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_start_fails_when_devices_cannot_be_opened(self, mock_input_device, mock_find_devices):
|
|
"""Test start fails when discovery finds devices but none can be opened."""
|
|
mock_find_devices.return_value = ["/dev/input/event0"]
|
|
mock_input_device.side_effect = OSError("permission denied")
|
|
|
|
backend = EvdevKeyboardBackend()
|
|
result = backend.start()
|
|
|
|
assert result is False
|
|
assert backend.active is False
|
|
assert backend.devices == []
|
|
assert backend.device_fds == []
|
|
|
|
|
|
class TestEvdevKeyboardBackendStop:
|
|
"""Test stop() method."""
|
|
|
|
def test_stop_when_not_active(self):
|
|
"""Test that stop is safe when not active."""
|
|
backend = EvdevKeyboardBackend()
|
|
backend.stop() # Should not raise
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.EVDEV_AVAILABLE", True)
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_stop_active_backend(self, mock_input_device, mock_find_devices):
|
|
"""Test stopping an active backend."""
|
|
mock_find_devices.return_value = ["/dev/input/event0"]
|
|
mock_device = MagicMock()
|
|
mock_device.fileno.return_value = 10
|
|
mock_input_device.return_value = mock_device
|
|
|
|
backend = EvdevKeyboardBackend()
|
|
backend.start()
|
|
|
|
# Give thread a moment to start
|
|
time.sleep(0.1)
|
|
|
|
backend.stop()
|
|
|
|
assert backend.active is False
|
|
assert backend.running is False
|
|
assert backend.devices == []
|
|
assert backend.device_fds == []
|
|
|
|
def test_stop_active_without_thread_handles_close_failure(self):
|
|
"""Test active stop without a monitor thread still clears device state."""
|
|
backend = EvdevKeyboardBackend()
|
|
mock_device = MagicMock()
|
|
mock_device.close.side_effect = RuntimeError("already closed")
|
|
backend.active = True
|
|
backend.running = True
|
|
backend.devices = [mock_device]
|
|
backend.device_fds = [10]
|
|
backend.device_paths = {"/dev/input/event0"}
|
|
backend._dropped_devices = {10}
|
|
backend._device_paths_by_fd = {10: "/dev/input/event0"}
|
|
|
|
backend.stop()
|
|
|
|
assert backend.active is False
|
|
assert backend.running is False
|
|
assert backend.devices == []
|
|
assert backend.device_fds == []
|
|
assert backend.device_paths == set()
|
|
assert backend._dropped_devices == set()
|
|
assert backend._device_paths_by_fd == {}
|
|
|
|
|
|
class TestEvdevKeyboardBackendHotplug:
|
|
"""Test hotplug discovery for evdev keyboard devices."""
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_scan_for_new_devices_opens_hotplugged_keyboard(
|
|
self, mock_input_device, mock_find_devices
|
|
):
|
|
"""Test that rescanning opens keyboard devices that appear after startup."""
|
|
existing_device = MagicMock()
|
|
existing_device.fileno.return_value = 10
|
|
new_device = MagicMock()
|
|
new_device.fileno.return_value = 11
|
|
new_device.name = "External Keyboard"
|
|
mock_input_device.return_value = new_device
|
|
mock_find_devices.return_value = ["/dev/input/event0", "/dev/input/event1"]
|
|
|
|
backend = EvdevKeyboardBackend()
|
|
backend.devices = [existing_device]
|
|
backend.device_fds = [10]
|
|
backend.device_paths = {"/dev/input/event0"}
|
|
backend._device_paths_by_fd = {10: "/dev/input/event0"}
|
|
|
|
added = backend._scan_for_new_devices()
|
|
|
|
assert added == 1
|
|
mock_input_device.assert_called_once_with("/dev/input/event1")
|
|
assert new_device in backend.devices
|
|
assert 11 in backend.device_fds
|
|
assert "/dev/input/event1" in backend.device_paths
|
|
assert backend._device_paths_by_fd[11] == "/dev/input/event1"
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_open_keyboard_device_skips_known_path(self, mock_input_device):
|
|
"""Test opening a device already tracked by path is a no-op."""
|
|
backend = EvdevKeyboardBackend()
|
|
backend.device_paths = {"/dev/input/event0"}
|
|
|
|
opened = backend._open_keyboard_device("/dev/input/event0")
|
|
|
|
assert opened is False
|
|
mock_input_device.assert_not_called()
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_open_keyboard_device_handles_open_failure(self, mock_input_device):
|
|
"""Test open failures leave backend device state unchanged."""
|
|
mock_input_device.side_effect = OSError("permission denied")
|
|
backend = EvdevKeyboardBackend()
|
|
|
|
opened = backend._open_keyboard_device("/dev/input/event7")
|
|
|
|
assert opened is False
|
|
assert backend.devices == []
|
|
assert backend.device_fds == []
|
|
assert backend.device_paths == set()
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_open_keyboard_device_closes_duplicate_fd(self, mock_input_device):
|
|
"""Test a duplicate fd opened through a new path is discarded."""
|
|
duplicate_device = MagicMock()
|
|
duplicate_device.fileno.return_value = 10
|
|
duplicate_device.close.side_effect = RuntimeError("close failed")
|
|
mock_input_device.return_value = duplicate_device
|
|
backend = EvdevKeyboardBackend()
|
|
backend.device_fds = [10]
|
|
|
|
opened = backend._open_keyboard_device("/dev/input/event9")
|
|
|
|
assert opened is False
|
|
assert backend.devices == []
|
|
assert backend.device_paths == set()
|
|
duplicate_device.close.assert_called_once()
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
def test_scan_for_new_devices_handles_discovery_failure(self, mock_find_devices):
|
|
"""Test rescan failures do not stop the monitor loop."""
|
|
mock_find_devices.side_effect = RuntimeError("proc read failed")
|
|
backend = EvdevKeyboardBackend()
|
|
|
|
added = backend._scan_for_new_devices()
|
|
|
|
assert added == 0
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
def test_scan_for_new_devices_no_new_devices(self, mock_find_devices):
|
|
"""Test rescanning with only already-tracked paths returns zero."""
|
|
mock_find_devices.return_value = ["/dev/input/event0"]
|
|
backend = EvdevKeyboardBackend()
|
|
backend.device_paths = {"/dev/input/event0"}
|
|
|
|
added = backend._scan_for_new_devices()
|
|
|
|
assert added == 0
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
def test_removed_device_path_can_be_reopened(self, mock_input_device):
|
|
"""Test that disconnected device paths are forgotten for later replug."""
|
|
old_device = MagicMock()
|
|
old_device.fileno.return_value = 10
|
|
old_device.name = "External Keyboard"
|
|
new_device = MagicMock()
|
|
new_device.fileno.return_value = 12
|
|
new_device.name = "External Keyboard"
|
|
mock_input_device.return_value = new_device
|
|
|
|
backend = EvdevKeyboardBackend()
|
|
backend.devices = [old_device]
|
|
backend.device_fds = [10]
|
|
backend.device_paths = {"/dev/input/event4"}
|
|
backend._device_paths_by_fd = {10: "/dev/input/event4"}
|
|
backend.key_pressed_devices = {id(old_device)}
|
|
|
|
backend._remove_keyboard_device(10, old_device)
|
|
|
|
assert backend.devices == []
|
|
assert backend.device_fds == []
|
|
assert backend.device_paths == set()
|
|
assert backend.key_pressed_devices == set()
|
|
old_device.close.assert_called_once()
|
|
|
|
reopened = backend._open_keyboard_device("/dev/input/event4")
|
|
|
|
assert reopened is True
|
|
mock_input_device.assert_called_once_with("/dev/input/event4")
|
|
assert new_device in backend.devices
|
|
assert backend._device_paths_by_fd[12] == "/dev/input/event4"
|
|
|
|
def test_remove_keyboard_device_handles_stale_state(self):
|
|
"""Test removing a device tolerates already-pruned backend state."""
|
|
stale_device = MagicMock()
|
|
stale_device.close.side_effect = RuntimeError("already closed")
|
|
backend = EvdevKeyboardBackend()
|
|
backend.devices = []
|
|
backend.device_fds = []
|
|
backend.device_paths = {"/dev/input/event8"}
|
|
backend._device_paths_by_fd = {}
|
|
backend._dropped_devices = {12}
|
|
backend.key_pressed_devices = {id(stale_device)}
|
|
|
|
backend._remove_keyboard_device(12, stale_device)
|
|
|
|
assert backend.device_paths == {"/dev/input/event8"}
|
|
assert backend._dropped_devices == set()
|
|
assert backend.key_pressed_devices == set()
|
|
|
|
def test_monitor_scans_when_all_devices_are_disconnected(self):
|
|
"""Test that the monitor keeps rescanning after the last fd is removed."""
|
|
backend = EvdevKeyboardBackend()
|
|
backend.running = True
|
|
backend.device_fds = []
|
|
backend._scan_for_new_devices = MagicMock(return_value=0)
|
|
|
|
def stop_after_sleep(seconds):
|
|
backend.running = False
|
|
|
|
with (
|
|
patch(
|
|
"vocalinux.ui.keyboard_backends.evdev_backend.time.monotonic",
|
|
side_effect=[0.0, DEVICE_RESCAN_SECONDS + 0.1],
|
|
),
|
|
patch(
|
|
"vocalinux.ui.keyboard_backends.evdev_backend.time.sleep",
|
|
side_effect=stop_after_sleep,
|
|
),
|
|
):
|
|
backend._monitor_devices()
|
|
|
|
backend._scan_for_new_devices.assert_called_once()
|
|
|
|
def test_monitor_dispatches_events_from_readable_device(self):
|
|
"""Test readable fds are resolved to devices and dispatched."""
|
|
backend = EvdevKeyboardBackend()
|
|
device = MagicMock()
|
|
device.fileno.return_value = 10
|
|
device.read.return_value = [MagicMock(type=1, code=29, value=1)]
|
|
backend.running = True
|
|
backend.devices = [device]
|
|
backend.device_fds = [10]
|
|
backend._handle_key_event = MagicMock(
|
|
side_effect=lambda event, device: setattr(backend, "running", False)
|
|
)
|
|
|
|
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.EV_KEY = 1
|
|
backend._monitor_devices()
|
|
|
|
backend._handle_key_event.assert_called_once()
|
|
|
|
def test_monitor_ignores_readable_fd_without_device(self):
|
|
"""Test readable fds that no longer map to a device are ignored."""
|
|
backend = EvdevKeyboardBackend()
|
|
backend.running = True
|
|
backend.devices = []
|
|
backend.device_fds = [10]
|
|
|
|
def select_unknown_fd(read_fds, write_fds, error_fds, timeout):
|
|
backend.running = False
|
|
return [10], [], []
|
|
|
|
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=select_unknown_fd,
|
|
),
|
|
):
|
|
backend._monitor_devices()
|
|
|
|
def test_monitor_handles_syn_dropped_recovery(self):
|
|
"""Test SYN_DROPPED state is cleared on the following SYN_REPORT."""
|
|
backend = EvdevKeyboardBackend()
|
|
device = MagicMock()
|
|
device.fileno.return_value = 10
|
|
device.name = "External Keyboard"
|
|
dropped_event = MagicMock(type=0, code=3)
|
|
report_event = MagicMock(type=0, code=0)
|
|
|
|
def read_events():
|
|
backend.running = False
|
|
return [dropped_event, report_event]
|
|
|
|
device.read.side_effect = read_events
|
|
backend.running = True
|
|
backend.devices = [device]
|
|
backend.device_fds = [10]
|
|
backend.key_pressed_devices = {id(device)}
|
|
|
|
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
|
|
backend._monitor_devices()
|
|
|
|
assert backend._dropped_devices == set()
|
|
assert backend.key_pressed_devices == set()
|
|
|
|
def test_monitor_skips_events_while_device_is_dropped(self):
|
|
"""Test key events are ignored while a device is in SYN_DROPPED state."""
|
|
backend = EvdevKeyboardBackend()
|
|
device = MagicMock()
|
|
device.fileno.return_value = 10
|
|
key_event = MagicMock(type=1, code=29, value=1)
|
|
|
|
def read_events():
|
|
backend.running = False
|
|
return [key_event]
|
|
|
|
device.read.side_effect = read_events
|
|
backend.running = True
|
|
backend.devices = [device]
|
|
backend.device_fds = [10]
|
|
backend._dropped_devices = {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
|
|
backend._monitor_devices()
|
|
|
|
backend._handle_key_event.assert_not_called()
|
|
|
|
def test_monitor_removes_device_on_read_error(self):
|
|
"""Test disconnected devices are removed from the monitor loop."""
|
|
backend = EvdevKeyboardBackend()
|
|
device = MagicMock()
|
|
device.fileno.return_value = 10
|
|
device.name = "External Keyboard"
|
|
device.read.side_effect = OSError("device removed")
|
|
backend.running = True
|
|
backend.devices = [device]
|
|
backend.device_fds = [10]
|
|
backend._remove_keyboard_device = MagicMock(
|
|
side_effect=lambda fd, device, generation: setattr(backend, "running", False)
|
|
)
|
|
|
|
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], [], []),
|
|
),
|
|
):
|
|
backend._monitor_devices()
|
|
|
|
backend._remove_keyboard_device.assert_called_once_with(
|
|
10, device, backend._hub._generation
|
|
)
|
|
|
|
def test_monitor_handles_fd_lookup_error(self):
|
|
"""Test fd lookup errors do not try to remove an unknown device."""
|
|
backend = EvdevKeyboardBackend()
|
|
device = MagicMock()
|
|
device.fileno.side_effect = OSError("bad fd")
|
|
backend.running = True
|
|
backend.devices = [device]
|
|
backend.device_fds = [10]
|
|
backend._remove_keyboard_device = MagicMock()
|
|
|
|
def select_bad_fd(read_fds, write_fds, error_fds, timeout):
|
|
backend.running = False
|
|
return [10], [], []
|
|
|
|
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=select_bad_fd,
|
|
),
|
|
):
|
|
backend._monitor_devices()
|
|
|
|
backend._remove_keyboard_device.assert_not_called()
|
|
|
|
def test_monitor_stops_on_select_error(self):
|
|
"""Test select errors stop the monitor loop."""
|
|
backend = EvdevKeyboardBackend()
|
|
backend.running = True
|
|
backend.device_fds = [10]
|
|
|
|
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()
|
|
|
|
|
|
class TestEvdevKeyboardBackendHandleKeyEvent:
|
|
"""Test _handle_key_event() method."""
|
|
|
|
def test_handle_key_event_toggle_mode_double_tap(self):
|
|
"""Test double-tap detection in toggle mode."""
|
|
backend = EvdevKeyboardBackend(shortcut="ctrl+ctrl", mode="toggle")
|
|
callback = MagicMock()
|
|
backend.register_toggle_callback(callback)
|
|
|
|
mock_device = MagicMock()
|
|
mock_event = MagicMock()
|
|
mock_event.code = 29 # KEY_LEFTCTRL
|
|
mock_event.value = 1 # Press
|
|
|
|
# Set up timing for double-tap
|
|
backend.last_key_press_time = time.time() - 0.1
|
|
backend.last_trigger_time = time.time() - 1.0
|
|
|
|
with patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes:
|
|
mock_ecodes.EV_KEY = 1
|
|
backend._handle_key_event(mock_event, mock_device)
|
|
|
|
time.sleep(0.1)
|
|
|
|
def test_handle_key_event_non_target_key(self):
|
|
"""Test that non-target keys are ignored."""
|
|
backend = EvdevKeyboardBackend(shortcut="ctrl+ctrl")
|
|
callback = MagicMock()
|
|
backend.register_toggle_callback(callback)
|
|
|
|
mock_device = MagicMock()
|
|
mock_event = MagicMock()
|
|
mock_event.code = 1 # Some non-modifier key
|
|
mock_event.value = 1
|
|
|
|
with patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes:
|
|
mock_ecodes.EV_KEY = 1
|
|
backend._handle_key_event(mock_event, mock_device)
|
|
|
|
# Callback should not be called
|
|
|
|
def test_handle_key_event_push_to_talk_press(self):
|
|
"""Test push-to-talk mode triggers on press."""
|
|
backend = EvdevKeyboardBackend(shortcut="alt+alt", mode="push_to_talk")
|
|
callback = MagicMock()
|
|
backend.register_press_callback(callback)
|
|
|
|
mock_device = MagicMock()
|
|
mock_event = MagicMock()
|
|
mock_event.code = 56 # KEY_LEFTALT
|
|
mock_event.value = 1 # Press
|
|
|
|
with patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes:
|
|
mock_ecodes.EV_KEY = 1
|
|
backend._handle_key_event(mock_event, mock_device)
|
|
|
|
time.sleep(0.1)
|
|
|
|
def test_handle_key_event_push_to_talk_release(self):
|
|
"""Test push-to-talk mode triggers on release."""
|
|
backend = EvdevKeyboardBackend(shortcut="shift+shift", mode="push_to_talk")
|
|
callback = MagicMock()
|
|
backend.register_release_callback(callback)
|
|
|
|
mock_device = MagicMock()
|
|
mock_event = MagicMock()
|
|
mock_event.code = 42 # KEY_LEFTSHIFT
|
|
mock_event.value = 0 # Release
|
|
|
|
with patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes:
|
|
mock_ecodes.EV_KEY = 1
|
|
backend._handle_key_event(mock_event, mock_device)
|
|
|
|
time.sleep(0.1)
|
|
|
|
def test_handle_key_event_exception_handling(self):
|
|
"""Test exception handling in _handle_key_event."""
|
|
backend = EvdevKeyboardBackend()
|
|
|
|
mock_device = MagicMock()
|
|
mock_event = MagicMock()
|
|
mock_event.code = 29
|
|
|
|
# Make _get_target_key_codes raise an exception
|
|
backend._get_target_key_codes = MagicMock(side_effect=Exception("Test error"))
|
|
|
|
backend._handle_key_event(mock_event, mock_device) # Should not raise
|
|
|
|
|
|
class TestEvdevKeyboardBackendGetTargetKeyCodes:
|
|
"""Test _get_target_key_codes() method."""
|
|
|
|
def test_get_target_key_codes_ctrl(self):
|
|
"""Test getting key codes for ctrl modifier."""
|
|
backend = EvdevKeyboardBackend(shortcut="ctrl+ctrl")
|
|
result = backend._get_target_key_codes()
|
|
|
|
assert 29 in result # KEY_LEFTCTRL
|
|
assert 97 in result # KEY_RIGHTCTRL
|
|
|
|
def test_get_target_key_codes_left_ctrl(self):
|
|
"""Test getting key codes for left_ctrl modifier."""
|
|
backend = EvdevKeyboardBackend(shortcut="left_ctrl+left_ctrl")
|
|
result = backend._get_target_key_codes()
|
|
|
|
assert 29 in result # KEY_LEFTCTRL
|
|
assert 97 not in result # KEY_RIGHTCTRL should not be included
|
|
|
|
def test_get_target_key_codes_alt(self):
|
|
"""Test getting key codes for alt modifier."""
|
|
backend = EvdevKeyboardBackend(shortcut="alt+alt")
|
|
result = backend._get_target_key_codes()
|
|
|
|
assert 56 in result # KEY_LEFTALT
|
|
assert 100 in result # KEY_RIGHTALT
|
|
|
|
def test_get_target_key_codes_shift(self):
|
|
"""Test getting key codes for shift modifier."""
|
|
backend = EvdevKeyboardBackend(shortcut="shift+shift")
|
|
result = backend._get_target_key_codes()
|
|
|
|
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)
|
|
|
|
def test_pure_modifier_press_stays_consumed(self) -> None:
|
|
"""A real PTT hold never reaches the application."""
|
|
backend = EvdevKeyboardBackend(shortcut="right_alt+right_alt")
|
|
fd = 10
|
|
backend._forwarders[fd] = MagicMock()
|
|
backend._forwarded_held[fd] = set()
|
|
backend._withheld_modifier[fd] = {}
|
|
|
|
assert backend._event_is_shortcut(fd, self._key_event(100, 1)) is True
|
|
assert backend._event_is_shortcut(fd, self._key_event(100, 0)) is True
|
|
backend._forwarders[fd].write_event.assert_not_called()
|
|
|
|
def test_altgr_chord_replays_withheld_press(self) -> None:
|
|
"""RightAlt held for composition replays its press and release."""
|
|
backend = EvdevKeyboardBackend(shortcut="right_alt+right_alt")
|
|
fd = 10
|
|
forwarder = MagicMock()
|
|
backend._forwarders[fd] = forwarder
|
|
backend._forwarded_held[fd] = set()
|
|
backend._withheld_modifier[fd] = {}
|
|
|
|
press = self._key_event(100, 1) # KEY_RIGHTALT down
|
|
key_e = self._key_event(18, 1) # KEY_E down (AltGr+e)
|
|
release = self._key_event(100, 0) # KEY_RIGHTALT up
|
|
|
|
assert backend._event_is_shortcut(fd, press) is True
|
|
forwarder.write_event.assert_not_called()
|
|
|
|
assert backend._event_is_shortcut(fd, key_e) is False
|
|
forwarder.write_event.assert_called_once_with(press)
|
|
assert 100 in backend._forwarded_held[fd]
|
|
|
|
assert backend._event_is_shortcut(fd, release) is True
|
|
# The replayed modifier's release passes through so the clone's key
|
|
# state matches the physical keyboard.
|
|
assert forwarder.write_event.call_count == 2
|
|
forwarder.write_event.assert_called_with(release)
|
|
|
|
def test_withheld_state_drops_on_device_removal(self) -> None:
|
|
"""Disconnecting a keyboard forgets its withheld modifier presses."""
|
|
backend = EvdevKeyboardBackend(shortcut="right_alt+right_alt")
|
|
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._withheld_modifier[10] = {100: [self._key_event(100, 1), False]}
|
|
|
|
backend._remove_keyboard_device(10, device)
|
|
|
|
assert backend._withheld_modifier == {}
|
|
|
|
def test_split_keyboard_altgr_replays_modifier_from_other_device(self) -> None:
|
|
"""A chord spanning two devices still replays the withheld modifier.
|
|
|
|
On a split keyboard RightAlt arrives on one device and the character
|
|
key on another: the modifier is withheld under its own fd, so the
|
|
replay must scan every device — checking only the character key's fd
|
|
would forward a plain character and lose the AltGr composition.
|
|
"""
|
|
backend = EvdevKeyboardBackend(shortcut="right_alt+right_alt")
|
|
mod_fd, char_fd = 10, 20
|
|
mod_forwarder, char_forwarder = MagicMock(), MagicMock()
|
|
backend._forwarders = {mod_fd: mod_forwarder, char_fd: char_forwarder}
|
|
backend._forwarded_held = {mod_fd: set(), char_fd: set()}
|
|
backend._withheld_modifier = {mod_fd: {}, char_fd: {}}
|
|
|
|
press = self._key_event(100, 1) # KEY_RIGHTALT down on the left half
|
|
key_e = self._key_event(18, 1) # KEY_E down on the right half
|
|
|
|
assert backend._event_is_shortcut(mod_fd, press) is True
|
|
mod_forwarder.write_event.assert_not_called()
|
|
|
|
assert backend._event_is_shortcut(char_fd, key_e) is False
|
|
# The modifier press is replayed through ITS OWN device's clone —
|
|
# before the character is forwarded — so the app sees AltGr+e.
|
|
mod_forwarder.write_event.assert_called_once_with(press)
|
|
assert 100 in backend._forwarded_held[mod_fd]
|
|
|
|
def test_syn_dropped_prunes_unreplayed_withheld_modifier(self) -> None:
|
|
"""A withheld press whose release SYN_DROPPED ate leaves no stale entry.
|
|
|
|
When the modifier was never replayed the clone holds nothing, so the
|
|
old early return skipped the withheld-state cleanup entirely and a
|
|
later ordinary press would replay a modifier that is not held.
|
|
"""
|
|
backend = EvdevKeyboardBackend(shortcut="right_alt+right_alt")
|
|
fd = 10
|
|
device = MagicMock()
|
|
device.active_keys.return_value = [] # modifier already released
|
|
backend._forwarders = {fd: MagicMock()}
|
|
backend._forwarded_held = {fd: set()} # nothing forwarded
|
|
backend._withheld_modifier = {fd: {100: [self._key_event(100, 1), False]}}
|
|
|
|
backend._resync_clone_key_state(fd, device)
|
|
|
|
assert backend._withheld_modifier[fd] == {}
|
|
|
|
|
|
class TestSharedEvdevDeviceLayer:
|
|
"""Several backends share one evdev reader instead of competing grabs.
|
|
|
|
Regression tests for the "language listeners compete for keyboards"
|
|
finding: each EvdevKeyboardBackend used to open and grab its own
|
|
InputDevice on the same keyboard, so the first grabber won and every
|
|
other listener received no events — language shortcuts could never
|
|
fire on Wayland.
|
|
"""
|
|
|
|
def _key_event(self, code: int, value: int = 1) -> MagicMock:
|
|
return MagicMock(type=ecodes.EV_KEY, code=code, value=value)
|
|
|
|
def _registered(self, *engines: EvdevKeyboardBackend) -> EvdevDeviceHub:
|
|
"""Register engines on the shared hub without opening devices."""
|
|
hub = engines[0]._hub
|
|
hub.running = True
|
|
for engine in engines:
|
|
assert hub.register(engine) is True
|
|
return hub
|
|
|
|
def test_backends_share_one_device_layer(self) -> None:
|
|
"""Every backend binds to the same process-wide hub."""
|
|
first = EvdevKeyboardBackend(shortcut="ctrl+ctrl")
|
|
second = EvdevKeyboardBackend(shortcut="alt+d")
|
|
|
|
assert first._hub is second._hub
|
|
assert second.devices is first.devices
|
|
assert second._forwarders is first._forwarders
|
|
assert second._device_paths_by_fd is first._device_paths_by_fd
|
|
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
|
|
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
|
|
def test_second_start_registers_without_reopening_devices(
|
|
self, mock_find: Mock, mock_input_device: Mock, mock_uinput: Mock
|
|
) -> None:
|
|
"""The first backend opens and grabs once; later ones just register."""
|
|
mock_find.return_value = ["/dev/input/event0"]
|
|
device = MagicMock()
|
|
device.name = "test-kbd"
|
|
device.fileno.return_value = 10
|
|
device.capabilities.return_value = {}
|
|
mock_input_device.return_value = device
|
|
forwarder = MagicMock()
|
|
forwarder.device.path = "/dev/input/event9"
|
|
mock_uinput.return_value = forwarder
|
|
|
|
first = EvdevKeyboardBackend(shortcut="ctrl+ctrl")
|
|
with patch("select.select", return_value=([], [], [])):
|
|
assert first.start() is True
|
|
second = EvdevKeyboardBackend(shortcut="alt+d")
|
|
assert second.start() is True
|
|
|
|
assert mock_input_device.call_count == 1 # device opened once
|
|
assert device.grab.call_count == 1 # grabbed once
|
|
assert set(first._hub._engine_snapshot()) == {first, second}
|
|
|
|
second.stop()
|
|
assert first._hub.running is True # layer outlives one listener
|
|
first.stop()
|
|
assert first._hub.running is False
|
|
device.close.assert_called()
|
|
|
|
def test_two_engines_both_fire_on_one_device(self) -> None:
|
|
"""Events on a shared keyboard reach every registered engine."""
|
|
german = EvdevKeyboardBackend(shortcut="ctrl+ctrl", mode="toggle")
|
|
french = EvdevKeyboardBackend(shortcut="alt+d", mode="toggle")
|
|
hub = self._registered(german, french)
|
|
|
|
german_fired = MagicMock()
|
|
french_fired = MagicMock()
|
|
german.register_toggle_callback(german_fired)
|
|
french.register_toggle_callback(french_fired)
|
|
|
|
device = MagicMock()
|
|
fd = 10
|
|
device.read.return_value = [
|
|
self._key_event(29, 1), # ctrl press (german tap 1)
|
|
self._key_event(29, 0), # ctrl release
|
|
self._key_event(29, 1), # ctrl press (german double-tap)
|
|
self._key_event(56, 1), # alt press (french modifier)
|
|
self._key_event(32, 1), # d press (french combo main key)
|
|
]
|
|
|
|
hub._dispatch_events(fd, device)
|
|
|
|
time.sleep(0.1)
|
|
german_fired.assert_called_once()
|
|
french_fired.assert_called_once()
|
|
|
|
def test_shortcut_events_consumed_once_across_engines(self) -> None:
|
|
"""An event claimed by any engine never reaches the application."""
|
|
pure = EvdevKeyboardBackend(shortcut="right_alt+right_alt")
|
|
combo = EvdevKeyboardBackend(shortcut="ctrl+f", mode="toggle")
|
|
hub = self._registered(pure, combo)
|
|
fd = 10
|
|
forwarder = MagicMock()
|
|
hub._forwarders = {fd: forwarder}
|
|
hub._forwarded_held = {fd: set()}
|
|
|
|
ctrl_press = self._key_event(29, 1)
|
|
f_press = self._key_event(33, 1)
|
|
a_press = self._key_event(30, 1)
|
|
ralt_press = self._key_event(100, 1)
|
|
ralt_release = self._key_event(100, 0)
|
|
device = MagicMock()
|
|
device.read.return_value = [ctrl_press, f_press, a_press, ralt_press, ralt_release]
|
|
|
|
hub._dispatch_events(fd, device)
|
|
|
|
# f is consumed by the combo engine while ctrl is held, both right-alt
|
|
# events by the pure-modifier engine; unrelated keys still pass through.
|
|
forwarded = [call.args[0] for call in forwarder.write_event.call_args_list]
|
|
assert forwarded == [ctrl_press, a_press]
|
|
|
|
def test_unregister_shuts_down_only_when_last_engine_leaves(self) -> None:
|
|
"""The shared layer stays up until its last listener stops."""
|
|
first = EvdevKeyboardBackend(shortcut="ctrl+ctrl")
|
|
second = EvdevKeyboardBackend(shortcut="alt+d")
|
|
hub = self._registered(first, second)
|
|
first.active = True
|
|
second.active = True
|
|
device = MagicMock()
|
|
hub.devices = [device]
|
|
|
|
first.stop()
|
|
assert hub.running is True
|
|
device.close.assert_not_called()
|
|
|
|
second.stop()
|
|
assert hub.running is False
|
|
assert hub.devices == []
|
|
device.close.assert_called_once()
|
|
|
|
def test_dispatch_reaches_only_registered_engines(self) -> None:
|
|
"""A backend that never started observes nothing."""
|
|
active = EvdevKeyboardBackend(shortcut="ctrl+ctrl")
|
|
idle = EvdevKeyboardBackend(shortcut="alt+d")
|
|
hub = self._registered(active)
|
|
active._handle_key_event = MagicMock()
|
|
idle._handle_key_event = MagicMock()
|
|
|
|
device = MagicMock()
|
|
device.read.return_value = [self._key_event(29, 1)]
|
|
hub._dispatch_events(10, device)
|
|
|
|
active._handle_key_event.assert_called_once()
|
|
idle._handle_key_event.assert_not_called()
|
|
|
|
def test_register_waits_for_in_flight_teardown(self) -> None:
|
|
"""Devices opened after teardown starts are never closed by it.
|
|
|
|
Regression test for "keyboard listener can lose devices": the last
|
|
engine's unregister used to release the lock before closing, so a
|
|
new register could open keyboards in the gap only for the stale
|
|
teardown to close them — the new listener stayed active but dead.
|
|
"""
|
|
hub = EvdevDeviceHub()
|
|
old_engine = EvdevKeyboardBackend(shortcut="ctrl+ctrl")
|
|
new_engine = EvdevKeyboardBackend(shortcut="alt+d")
|
|
old_device = MagicMock()
|
|
hub.running = True
|
|
hub._engines.add(old_engine)
|
|
hub.devices = [old_device]
|
|
|
|
teardown_started = threading.Event()
|
|
finish_teardown = threading.Event()
|
|
|
|
def blocking_close() -> None:
|
|
teardown_started.set()
|
|
finish_teardown.wait(timeout=5.0)
|
|
|
|
hub._close_all_devices = blocking_close
|
|
|
|
unregister_done = threading.Event()
|
|
|
|
def run_unregister() -> None:
|
|
hub.unregister(old_engine)
|
|
unregister_done.set()
|
|
|
|
threading.Thread(target=run_unregister, daemon=True).start()
|
|
assert teardown_started.wait(timeout=5.0)
|
|
|
|
new_device = MagicMock()
|
|
new_device.fileno.return_value = 42
|
|
result: list[bool] = []
|
|
with (
|
|
patch(
|
|
"vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices",
|
|
return_value=["/dev/input/event9"],
|
|
),
|
|
patch(
|
|
"vocalinux.ui.keyboard_backends.evdev_backend.InputDevice",
|
|
return_value=new_device,
|
|
),
|
|
patch(
|
|
"vocalinux.ui.keyboard_backends.evdev_backend.UInput",
|
|
return_value=MagicMock(),
|
|
),
|
|
):
|
|
register_thread = threading.Thread(
|
|
target=lambda: result.append(hub.register(new_engine)),
|
|
daemon=True,
|
|
)
|
|
register_thread.start()
|
|
register_thread.join(timeout=0.5)
|
|
assert register_thread.is_alive() # blocked while teardown runs
|
|
|
|
finish_teardown.set()
|
|
register_thread.join(timeout=5.0)
|
|
assert result == [True]
|
|
new_device.close.assert_not_called()
|
|
assert hub.running is True
|
|
assert hub.devices == [new_device]
|
|
assert new_engine in hub._engines
|
|
assert old_engine not in hub._engines
|
|
|
|
assert unregister_done.wait(timeout=5.0)
|
|
hub.reset()
|
|
|
|
def test_unregister_joins_monitor_outside_the_lifecycle_lock(self) -> None:
|
|
"""The monitor join must not hold the lock its cleanup queues on.
|
|
|
|
Regression test for "monitor cleanup delays shutdown": joining under
|
|
_lifecycle_lock stalled the full 2s timeout whenever the monitor's
|
|
unexpected-exit cleanup raced in for the same lock.
|
|
"""
|
|
hub = EvdevDeviceHub()
|
|
engine = EvdevKeyboardBackend(shortcut="ctrl+ctrl")
|
|
hub.running = True
|
|
hub._engines.add(engine)
|
|
monitor = MagicMock()
|
|
hub.monitor_thread = monitor
|
|
|
|
lock_was_free = threading.Event()
|
|
|
|
def join_and_probe(timeout: float) -> None:
|
|
if hub._lifecycle_lock.acquire(blocking=False):
|
|
lock_was_free.set()
|
|
hub._lifecycle_lock.release()
|
|
|
|
monitor.join = join_and_probe
|
|
|
|
hub.unregister(engine)
|
|
|
|
assert lock_was_free.is_set()
|
|
assert hub.running is False
|
|
assert hub.monitor_thread is None
|
|
assert engine not in hub._engines
|
|
hub.reset()
|
|
|
|
def test_stale_monitor_leaves_new_generation_devices_alone(self) -> None:
|
|
"""A monitor that outlived its generation can't touch new devices.
|
|
|
|
Regression test for "old monitor reaches new devices": once the
|
|
generation advanced, every fd the new listener opened is off-limits
|
|
to the stale reader — no dispatch, no forward, no removal — even
|
|
though its fd number is readable and mapped.
|
|
"""
|
|
hub = EvdevDeviceHub()
|
|
hub.running = True
|
|
hub._generation = 6
|
|
|
|
new_device = MagicMock()
|
|
new_device.fileno.return_value = 10
|
|
new_device.name = "External Keyboard"
|
|
new_clone = MagicMock()
|
|
hub.devices = [new_device]
|
|
hub.device_fds = [10]
|
|
hub.device_paths = {"/dev/input/event4"}
|
|
hub._device_paths_by_fd = {10: "/dev/input/event4"}
|
|
hub._forwarders = {10: new_clone}
|
|
hub._fd_generation = {10: 6}
|
|
hub._dispatch_events = MagicMock()
|
|
|
|
def select_then_stop(read_fds, write_fds, error_fds, timeout):
|
|
hub.running = False
|
|
return [10], [], []
|
|
|
|
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=select_then_stop,
|
|
),
|
|
):
|
|
hub._monitor_devices(generation=5)
|
|
|
|
hub._dispatch_events.assert_not_called()
|
|
|
|
hub._forward_event(10, MagicMock(type=1, code=30, value=1), generation=5)
|
|
new_clone.write_event.assert_not_called()
|
|
|
|
hub._remove_keyboard_device(10, new_device, generation=5)
|
|
new_device.close.assert_not_called()
|
|
assert hub.devices == [new_device]
|
|
assert hub.device_fds == [10]
|
|
|
|
hub.reset()
|