xavierk/voculinux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity
Files
0c48db5b5ef162d3e6e96f9b2a40e667e7fdf3e0
voculinux/tests/test_evdev_backend_ext.py
T
soubarnak ea7be5f111 feat(identity): rename user-visible identity to voculinux with one-time legacy import 
The fork takes its own name everywhere a user can see it: binaries
(voculinux, voculinux-gui), desktop entry, autostart, tray, D-Bus
namespace (com.voculinux.Voculinux, no alias for the old name),
config/data/cache dirs, single-instance lock, IBus engine identity,
tray icons, resource prefixes, VOCULINUX_* environment names, and the
update-checker user agent. The internal Python module stays vocalinux
and the distribution metadata keeps the upstream URLs as provenance.

On first run a previous Vocalinux install's config.json, custom
dictionary, and portal restore token are copied once into the new
locations (new config always wins, no merge, no write-back); models
re-resolve through the pinned-checksum download path. An offline
naming-sweep test proves no old-name strings remain on user-visible
surfaces outside the allowlist.

Refs #18
2026-11-06 01:29:39 +05:30

3596 lines
148 KiB
Python
Raw Blame History

"""
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 contextlib
import errno
import os
import threading
import time
from collections.abc import Callable
from unittest.mock import MagicMock, Mock, call, mock_open, patch
import pytest
from vocalinux.ui.keyboard_backends.evdev_backend import (
_MONITOR_STALL_SECONDS,
DEVICE_RESCAN_SECONDS,
EvdevDeviceHub,
EvdevKeyboardBackend,
InputEvent,
_clone_device_name,
_find_keyboard_devices_from_evdev,
_is_virtual_keyboard,
device_has_key,
device_has_modifier_key,
ecodes,
find_keyboard_devices,
)
def _proc_key_bitmap(*codes: int, word_bits: int = 64) -> str:
"""Format key codes as the words printed in /proc device bitmaps."""
bits = sum(1 << code for code in codes)
word_mask = (1 << word_bits) - 1
word_count = max(1, (bits.bit_length() + word_bits - 1) // word_bits)
return " ".join(
f"{(bits >> (word_index * word_bits)) & word_mask:x}"
for word_index in reversed(range(word_count))
)
def _keyboard_only_capabilities() -> dict[int, list[int]]:
return {ecodes.EV_KEY: [ecodes.KEY_A, ecodes.KEY_RIGHTCTRL]}
class TestIsVirtualKeyboard:
"""Test _is_virtual_keyboard() sysfs and name detection (#930)."""
@patch("vocalinux.ui.keyboard_backends.evdev_backend.os.path.realpath")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.os.path.exists")
def test_sysfs_virtual_path_is_virtual(self, mock_exists: Mock, mock_realpath: Mock) -> None:
"""A device node under /devices/virtual/input is uinput-spawned."""
mock_exists.return_value = True
mock_realpath.return_value = "/sys/devices/virtual/input/input30"
assert _is_virtual_keyboard("/dev/input/event9", "Whatever Keyboard") is True
mock_realpath.assert_called_once_with("/sys/class/input/event9/device")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.os.path.realpath")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.os.path.exists")
def test_sysfs_real_path_is_not_virtual(self, mock_exists: Mock, mock_realpath: Mock) -> None:
"""A bus-attached path wins over a remapper-sounding name."""
mock_exists.return_value = True
mock_realpath.return_value = "/sys/devices/pci0000:00/usb1/input/input7"
assert _is_virtual_keyboard("/dev/input/event9", "keyd virtual keyboard") is False
@patch("vocalinux.ui.keyboard_backends.evdev_backend.os.path.exists")
def test_name_fallback_matches_remap_names(self, mock_exists: Mock) -> None:
"""Without sysfs, known remapper names are treated as virtual."""
mock_exists.return_value = False
for name in ("keyd virtual keyboard", "KMonad output", "kanata", "xremap"):
assert _is_virtual_keyboard("/dev/input/event9", name) is True, name
@patch("vocalinux.ui.keyboard_backends.evdev_backend.os.path.exists")
def test_name_fallback_ignores_real_keyboards(self, mock_exists: Mock) -> None:
"""Without sysfs, ordinary keyboard names stay grabbed."""
mock_exists.return_value = False
assert _is_virtual_keyboard("/dev/input/event9", "AT Translated Set 2 keyboard") is False
@patch("vocalinux.ui.keyboard_backends.evdev_backend.os.path.realpath")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.os.path.exists")
def test_sysfs_oserror_falls_back_to_names(
self, mock_exists: Mock, mock_realpath: Mock
) -> None:
"""A sysfs read failure degrades to the name check, not a crash."""
mock_exists.side_effect = OSError("permission denied")
assert _is_virtual_keyboard("/dev/input/event9", "keyd virtual keyboard") is True
assert _is_virtual_keyboard("/dev/input/event9", "AT Translated Set 2 keyboard") is False
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=402000000 3803078f800d001 feffffdfffefffff fffffffffffffffe
"""
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=402000000 3803078f800d001 feffffdfffefffff fffffffffffffffe
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=402000000 3803078f800d001 feffffdfffefffff fffffffffffffffe
"""
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=402000000 3803078f800d001 feffffdfffefffff fffffffffffffffe
"""
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
def test_find_keyboard_devices_keeps_relative_combo_and_skips_mouse(self) -> None:
"""Keyboard keys keep a REL_X/Y combo; button-only pointer nodes stay excluded."""
mock_proc_content = f"""I: Bus=0011 Vendor=0001 Product=0001 Version=ab83
N: Name="AT Translated Set 2 keyboard"
H: Handlers=sysrq kbd event0
B: KEY=402000000 3803078f800d001 feffffdfffefffff fffffffffffffffe
I: Bus=0003 Vendor=046d Product=c08b Version=0110
N: Name="Logitech G502 HERO Gaming Mouse"
H: Handlers=mouse0 event5
B: KEY=1f0000 0 0 0 38000000
B: REL=143
B: MSC=10
I: Bus=0003 Vendor=046d Product=0001 Version=0110
N: Name="Test Mouse"
H: Handlers=mouse0 event6
B: KEY={_proc_key_bitmap(ecodes.BTN_LEFT, ecodes.BTN_RIGHT)}
B: REL=143
"""
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
with patch("os.path.exists", return_value=True):
result = find_keyboard_devices()
assert result == ["/dev/input/event0", "/dev/input/event5"]
def test_find_keyboard_devices_keeps_absolute_combo_and_skips_touchpad(self) -> None:
"""Keyboard keys keep an ABS_X/Y combo; button-only touchpads stay excluded."""
mock_proc_content = f"""I: Bus=0018 Vendor=04f3 Product=0033 Version=0500
N: Name="Elan Touchpad"
H: Handlers=mouse0 event1
B: KEY=ff000000000000 0 0 38000000
B: ABS=273000000000003
I: Bus=0018 Vendor=04f3 Product=0034 Version=0500
N: Name="Button-only Touchpad"
H: Handlers=mouse0 event2
B: KEY={_proc_key_bitmap(ecodes.BTN_TOUCH)}
B: ABS=273000000000003
"""
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
with patch("os.path.exists", return_value=True):
result = find_keyboard_devices()
assert result == ["/dev/input/event1"]
def test_find_keyboard_devices_keeps_wheel_only_device(self) -> None:
"""REL_WHEEL without X/Y motion does not mark a device as a pointer."""
mock_proc_content = """I: Bus=0005 Vendor=046d Product=b331 Version=0030
N: Name="Wireless Keyboard"
H: Handlers=kbd event3
B: KEY=402000000 3803078f800d001 feffffdfffefffff fffffffffffffffe
B: REL=100
"""
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
with patch("os.path.exists", return_value=True):
result = find_keyboard_devices()
assert result == ["/dev/input/event3"]
def test_find_keyboard_devices_keeps_hidpp_keyboard_pointer_combos(self) -> None:
"""All keyboard-capable HID++ nodes remain eligible regardless of name."""
# Real blocks from one Bolt receiver: keyboards and mouse share the
# same REL bitmap, only the name tells them apart.
mock_proc_content = """I: Bus=0003 Vendor=046d Product=408a Version=0111
N: Name="Logitech MX Keys"
H: Handlers=sysrq kbd leds mouse1 event3
B: EV=12001f
B: KEY=3f00733fff 0 0 483ffff17aff32d bfd4444600000000 \
ffff0001 130ff38b17d007 ffff7bfad941dfff ffbeffdfffefffff fffffffffffffffe
B: REL=1943
B: ABS=100000000
I: Bus=0003 Vendor=046d Product=4088 Version=0111
N: Name="Logitech ERGO K860"
H: Handlers=sysrq kbd leds mouse0 event4
B: EV=12001f
B: KEY=3f00733fff 0 0 483ffff17aff32d bfd4444600000000 \
ffff0001 130ff38b17d007 ffff7bfad941dfff ffbeffdfffefffff fffffffffffffffe
B: REL=1943
B: ABS=100000000
I: Bus=0003 Vendor=046d Product=4082 Version=0111
N: Name="Logitech MX Master 3"
H: Handlers=sysrq kbd leds mouse2 event5
B: EV=120017
B: KEY=ffff0000 1000000000007 ff9f207ac14057ff febeffdfffefffff fffffffffffffffe
B: REL=1943
"""
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
with patch("os.path.exists", return_value=True):
result = find_keyboard_devices()
assert result == [
"/dev/input/event3",
"/dev/input/event4",
"/dev/input/event5",
]
def test_find_keyboard_devices_keeps_split_bitmap_combo_and_skips_mouse(self) -> None:
"""Low-word REL_X/Y axes keep keyboard combos, not button-only mice."""
mock_proc_content = f"""I: Bus=0003 Vendor=046d Product=c52b Version=1201
N: Name="Logitech USB Receiver"
H: Handlers=mouse0 event4
B: KEY=ffff0000 0 0 0 38000000
B: REL=0 0 143
I: Bus=0003 Vendor=046d Product=0001 Version=0110
N: Name="Button-only Mouse"
H: Handlers=mouse0 event5
B: KEY={_proc_key_bitmap(ecodes.BTN_LEFT, ecodes.BTN_RIGHT)}
B: REL=0 0 143
"""
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
with patch("os.path.exists", return_value=True):
result = find_keyboard_devices()
assert result == ["/dev/input/event4"]
def test_find_keyboard_devices_skips_button_only_devices(self) -> None:
"""Button-only devices are excluded while keyboard-pointer combos remain eligible."""
mock_proc_content = """I: Bus=0019 Vendor=0000 Product=0001 Version=0000
N: Name="Power Button"
H: Handlers=kbd event4
B: KEY=1000000000000 0
I: Bus=0019 Vendor=0000 Product=0003 Version=0000
N: Name="Sleep Button"
H: Handlers=kbd event5
B: KEY=4000 0 0
I: Bus=0019 Vendor=0000 Product=0006 Version=0000
N: Name="Video Bus"
H: Handlers=kbd event6
B: KEY=3f000300000000 0 0 0 0
I: Bus=0019 Vendor=17aa Product=5054 Version=4101
N: Name="ThinkPad Extra Buttons"
H: Handlers=kbd event7 rfkill
B: KEY=40040002000000 c00017 21004000 300600 400028000000 0
I: Bus=0011 Vendor=0001 Product=0001 Version=ab83
N: Name="AT Translated Set 2 keyboard"
H: Handlers=sysrq kbd event0
B: KEY=402000000 3803078f800d001 feffffdfffefffff fffffffffffffffe
I: Bus=0003 Vendor=046d Product=c08b Version=0110
N: Name="Logitech G502 HERO Gaming Mouse Keyboard"
H: Handlers=sysrq kbd event8
B: KEY=ff 0 0 0 38000000
I: Bus=0006 Vendor=0000 Product=0000 Version=0001
N: Name="ydotoold virtual device"
H: Handlers=kbd mouse0 event9
B: KEY=ffff0000 0 0 38000000
B: REL=143
"""
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
with patch("os.path.exists", return_value=True):
result = find_keyboard_devices()
assert result == [
"/dev/input/event0",
"/dev/input/event8",
"/dev/input/event9",
]
def test_find_keyboard_devices_32bit_words(self) -> None:
"""On a 32-bit kernel each bitmap word holds 32 bits, not 64.
KEY_RIGHTCTRL (code 97) sits in the fourth 32-bit word — the first
word printed. A 64-bit reading would look for it in word 1 and
miss the keyboard entirely.
"""
mock_proc_content = """I: Bus=0019 Vendor=0000 Product=0001 Version=0000
N: Name="GPIO keyboard"
H: Handlers=kbd event2
B: KEY=2 0 0 0
"""
with patch("vocalinux.ui.keyboard_backends.evdev_backend._WORD_BITS", 32):
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
with patch("os.path.exists", return_value=True):
result = find_keyboard_devices()
assert result == ["/dev/input/event2"]
def test_find_keyboard_devices_skips_multitouch_only_touchpad(self) -> None:
"""A touchpad reporting only MT axes (no ABS_X/Y) is a pointer (#914)."""
mock_proc_content = f"""I: Bus=0018 Vendor=27c6 Product=01e9 Version=0100
N: Name="GXTP5100:00 27C6:01E9"
H: Handlers=mouse0 event2
B: PROP=85
B: EV=1b
B: KEY={_proc_key_bitmap(ecodes.BTN_TOUCH)}
B: ABS=60000000000000
B: MSC=10
"""
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
with patch("os.path.exists", return_value=True):
result = find_keyboard_devices()
assert result == []
def test_find_keyboard_devices_skips_mixed_single_and_mt_touchpad(self) -> None:
"""A button-only touchpad with both ABS and MT axes stays skipped."""
mock_proc_content = f"""I: Bus=0018 Vendor=27c6 Product=01e9 Version=0100
N: Name="GXTP5100:00 27C6:01E9"
H: Handlers=mouse0 event2
B: KEY={_proc_key_bitmap(ecodes.BTN_TOUCH)}
B: ABS=60000000000003
"""
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
with patch("os.path.exists", return_value=True):
result = find_keyboard_devices()
assert result == []
def test_find_keyboard_devices_mt_axes_32bit_words(self) -> None:
"""32-bit MT axes retain keyboard combos and skip button-only touchpads."""
mock_proc_content = f"""I: Bus=0018 Vendor=27c6 Product=01e9 Version=0100
N: Name="GXTP5100:00 27C6:01E9"
H: Handlers=mouse0 event2
B: KEY={_proc_key_bitmap(ecodes.KEY_RIGHTCTRL, word_bits=32)}
B: ABS=600000 0
I: Bus=0018 Vendor=27c6 Product=01ea Version=0100
N: Name="Button-only MT touchpad"
H: Handlers=mouse0 event3
B: KEY={_proc_key_bitmap(ecodes.BTN_TOUCH, word_bits=32)}
B: ABS=600000 0
"""
with patch("vocalinux.ui.keyboard_backends.evdev_backend._WORD_BITS", 32):
with patch("builtins.open", mock_open(read_data=mock_proc_content)):
with patch("os.path.exists", return_value=True):
result = find_keyboard_devices()
assert result == ["/dev/input/event2"]
class TestFindKeyboardDevicesFromEvdev:
"""Test _find_keyboard_devices_from_evdev() discovery and filtering."""
def _devices(self, caps_by_path: dict[str, dict]) -> dict[str, MagicMock]:
devices = {}
for path, caps in caps_by_path.items():
device = MagicMock()
device.capabilities.return_value = caps
device.name = f"fake-{path}"
devices[path] = device
return devices
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.evdev")
def test_discovers_keyboards(self, mock_evdev: Mock, mock_input_device: Mock) -> None:
"""A plain keyboard device is reported."""
devices = self._devices({"/dev/input/event0": {ecodes.EV_KEY: [ecodes.KEY_A]}})
mock_evdev.list_devices.return_value = list(devices)
mock_input_device.side_effect = lambda path: devices[path]
assert _find_keyboard_devices_from_evdev() == ["/dev/input/event0"]
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.evdev")
def test_keeps_keyboard_pointer_devices_and_skips_button_only_pointers(
self, mock_evdev: Mock, mock_input_device: Mock
) -> None:
"""Keyboard combos are kept; pointer nodes with only buttons are excluded."""
caps = {
"/dev/input/event0": {ecodes.EV_KEY: [ecodes.KEY_A]},
# Keyboard-plus-pointer combo.
"/dev/input/event1": {
ecodes.EV_KEY: [ecodes.KEY_LEFTCTRL],
ecodes.EV_REL: [ecodes.REL_X, ecodes.REL_Y, ecodes.REL_WHEEL],
},
# Keyboard-plus-absolute-pointer combo (AbsInfo tuples).
"/dev/input/event2": {
ecodes.EV_KEY: [ecodes.KEY_A],
ecodes.EV_ABS: [(ecodes.ABS_X, MagicMock()), (ecodes.ABS_Y, MagicMock())],
},
# Keyboard with a wheel: relative axis but no X/Y motion.
"/dev/input/event3": {
ecodes.EV_KEY: [ecodes.KEY_A],
ecodes.EV_REL: [ecodes.REL_WHEEL],
},
# Keyboard-plus-MT combo: no ABS_X/Y.
"/dev/input/event4": {
ecodes.EV_KEY: [ecodes.KEY_LEFTCTRL, ecodes.BTN_LEFT],
ecodes.EV_ABS: [
(ecodes.ABS_MT_POSITION_X, MagicMock()),
(ecodes.ABS_MT_POSITION_Y, MagicMock()),
],
},
# Button-only pointer nodes remain excluded.
"/dev/input/event5": {
ecodes.EV_KEY: [ecodes.BTN_LEFT, ecodes.BTN_RIGHT],
ecodes.EV_REL: [ecodes.REL_X, ecodes.REL_Y],
},
"/dev/input/event6": {
ecodes.EV_KEY: [ecodes.BTN_TOUCH],
ecodes.EV_ABS: [
(ecodes.ABS_MT_POSITION_X, MagicMock()),
(ecodes.ABS_MT_POSITION_Y, MagicMock()),
],
},
}
devices = self._devices(caps)
mock_evdev.list_devices.return_value = list(caps)
mock_input_device.side_effect = lambda path: devices[path]
assert _find_keyboard_devices_from_evdev() == [
"/dev/input/event0",
"/dev/input/event1",
"/dev/input/event2",
"/dev/input/event3",
"/dev/input/event4",
]
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.evdev")
def test_keeps_hidpp_keyboard_pointer_combos(
self, mock_evdev: Mock, mock_input_device: Mock
) -> None:
"""All three HID++ nodes have keyboard keys and pointer axes."""
hidpp_caps = {
ecodes.EV_KEY: [ecodes.KEY_A, ecodes.KEY_RIGHTALT, ecodes.BTN_LEFT],
ecodes.EV_REL: [
ecodes.REL_X,
ecodes.REL_Y,
ecodes.REL_HWHEEL,
ecodes.REL_WHEEL,
ecodes.REL_WHEEL_HI_RES,
ecodes.REL_HWHEEL_HI_RES,
],
}
names = {
"/dev/input/event3": "Logitech MX Keys",
"/dev/input/event4": "Logitech ERGO K860",
"/dev/input/event5": "Logitech MX Master 3",
}
devices = self._devices({path: hidpp_caps for path in names})
for path, name in names.items():
devices[path].name = name
mock_evdev.list_devices.return_value = list(names)
mock_input_device.side_effect = lambda path: devices[path]
assert _find_keyboard_devices_from_evdev() == [
"/dev/input/event3",
"/dev/input/event4",
"/dev/input/event5",
]
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/voculinux/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/voculinux/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
@pytest.mark.parametrize(
"axes",
[
pytest.param(
{ecodes.EV_REL: [ecodes.REL_X, ecodes.REL_Y]},
id="relative",
),
pytest.param(
{
ecodes.EV_ABS: [
(ecodes.ABS_MT_POSITION_X, MagicMock()),
(ecodes.ABS_MT_POSITION_Y, MagicMock()),
]
},
id="multitouch",
),
],
)
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend._is_virtual_keyboard")
def test_open_monitors_keyboard_pointer_ungrabbed(
self,
mock_is_virtual: Mock,
mock_uinput: Mock,
mock_input_device: Mock,
axes: dict[int, list[object]],
) -> None:
"""Pointer-axis keyboard combos register without grabbing or cloning."""
mock_is_virtual.return_value = False
device = MagicMock()
device.fileno.return_value = 10
device.name = "MX Mechanical"
device.capabilities.return_value = {
ecodes.EV_KEY: [ecodes.KEY_A, ecodes.KEY_RIGHTCTRL],
**axes,
}
mock_input_device.return_value = device
backend = EvdevKeyboardBackend()
create_forwarder = Mock()
backend._create_forwarder = create_forwarder
assert backend._open_keyboard_device("/dev/input/event5") is True
mock_is_virtual.assert_called_once_with("/dev/input/event5", "MX Mechanical")
create_forwarder.assert_not_called()
mock_uinput.assert_not_called()
device.grab.assert_not_called()
assert backend.device_paths == {"/dev/input/event5"}
assert backend.device_fds == [10]
assert backend._forwarders == {}
assert device in backend.devices
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend._is_virtual_keyboard")
def test_open_monitors_keyboard_with_unknown_capabilities_ungrabbed(
self,
mock_is_virtual: Mock,
mock_uinput: Mock,
mock_input_device: Mock,
) -> None:
"""Capability read errors choose the safe ungrabbed path."""
mock_is_virtual.return_value = False
device = MagicMock()
device.fileno.return_value = 10
device.name = "Unknown keyboard"
device.capabilities.side_effect = OSError("capabilities unavailable")
mock_input_device.return_value = device
backend = EvdevKeyboardBackend()
create_forwarder = Mock()
backend._create_forwarder = create_forwarder
assert backend._open_keyboard_device("/dev/input/event5") is True
create_forwarder.assert_not_called()
mock_uinput.assert_not_called()
device.grab.assert_not_called()
assert backend.device_paths == {"/dev/input/event5"}
assert backend.device_fds == [10]
assert backend._forwarders == {}
@pytest.mark.parametrize("capabilities", [None, []])
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend._is_virtual_keyboard")
def test_open_monitors_keyboard_with_non_mapping_capabilities_ungrabbed(
self,
mock_is_virtual: Mock,
mock_uinput: Mock,
mock_input_device: Mock,
capabilities: object,
) -> None:
"""Non-mapping capability results are unknown and remain ungrabbed."""
mock_is_virtual.return_value = False
device = MagicMock()
device.fileno.return_value = 10
device.name = "Unknown keyboard"
device.capabilities.return_value = capabilities
mock_input_device.return_value = device
backend = EvdevKeyboardBackend()
create_forwarder = Mock()
backend._create_forwarder = create_forwarder
assert backend._open_keyboard_device("/dev/input/event5") is True
create_forwarder.assert_not_called()
mock_uinput.assert_not_called()
device.grab.assert_not_called()
assert backend.device_paths == {"/dev/input/event5"}
assert backend.device_fds == [10]
assert backend._forwarders == {}
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend._is_virtual_keyboard")
def test_ungrabbed_keyboard_pointer_dispatches_shortcut_without_forwarder(
self,
mock_is_virtual: Mock,
mock_uinput: Mock,
mock_input_device: Mock,
) -> None:
"""RIGHTCTRL events reach the registered engine without a clone."""
mock_is_virtual.return_value = False
device = MagicMock()
device.fileno.return_value = 10
device.name = "MX Mechanical"
device.capabilities.return_value = {
ecodes.EV_KEY: [ecodes.KEY_A, ecodes.KEY_RIGHTCTRL],
ecodes.EV_REL: [ecodes.REL_X, ecodes.REL_Y],
}
mock_input_device.return_value = device
backend = EvdevKeyboardBackend(shortcut="right_ctrl+right_ctrl", mode="push_to_talk")
activated = Mock()
released = Mock()
backend.register_press_callback(activated)
backend.register_release_callback(released)
assert backend._open_keyboard_device("/dev/input/event5") is True
hub = backend._hub
press = MagicMock(type=ecodes.EV_KEY, code=ecodes.KEY_RIGHTCTRL, value=1)
release = MagicMock(type=ecodes.EV_KEY, code=ecodes.KEY_RIGHTCTRL, value=0)
hub._dispatch_event(10, device, press, [backend])
hub._dispatch_event(10, device, release, [backend])
time.sleep(0.1)
activated.assert_called_once()
released.assert_called_once()
mock_uinput.assert_not_called()
assert hub._forwarders == {}
@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, mock_uinput: Mock
) -> None:
"""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
device.active_keys.return_value = []
device.capabilities.return_value = _keyboard_only_capabilities()
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_clone_drops_rfkill_switch_keeps_other_switches(
self, mock_input_device: Mock, mock_uinput: Mock
) -> None:
"""The clone must not declare SW_RFKILL_ALL (#946).
"ThinkPad Extra Buttons" declares SW_RFKILL_ALL. A fresh uinput
device reports every switch as 0, and the kernel's rfkill-input
handler reads a 0 killswitch as radios-off — cloning the bit
soft-blocked all radios on every startup and resume. Other
switches stay declared so real switch events still forward.
"""
device = MagicMock()
device.fileno.return_value = 10
device.active_keys.return_value = []
device.capabilities.return_value = {
ecodes.EV_KEY: [30], # KEY_A
ecodes.EV_SW: [ecodes.SW_RFKILL_ALL, ecodes.SW_TABLET_MODE],
}
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/event0")
assert opened is True
events = mock_uinput.call_args.kwargs["events"]
assert ecodes.SW_RFKILL_ALL not in events[ecodes.EV_SW]
assert ecodes.SW_TABLET_MODE in events[ecodes.EV_SW]
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
def test_clone_drops_ev_sw_when_only_switch_is_rfkill(
self, mock_input_device: Mock, mock_uinput: Mock
) -> None:
"""A device declaring only SW_RFKILL_ALL gets no EV_SW at all (#946).
ThinkPad Extra Buttons reports SW=8: bit 3 alone. Declaring an
empty EV_SW capability would still register a switch device, so
the type is dropped entirely.
"""
device = MagicMock()
device.fileno.return_value = 10
device.active_keys.return_value = []
device.capabilities.return_value = {
ecodes.EV_KEY: [30], # KEY_A
ecodes.EV_SW: [ecodes.SW_RFKILL_ALL],
}
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/event0")
assert opened is True
events = mock_uinput.call_args.kwargs["events"]
assert ecodes.EV_SW not in events
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.os.path.exists")
def test_open_monitors_remapper_virtual_device_ungrabbed(
self, mock_exists: Mock, mock_input_device: Mock, mock_uinput: Mock
) -> None:
"""Remapper virtual keyboards are monitored without a grab or clone.
keyd reports BUS_USB to look like a real keyboard; with sysfs
unreadable the name check still catches it (#930). Grabbing the
remapper's output device starves the compositor, and the clone is
another uinput device the remapper grabs — so no clone is made.
The listener still sees its events, so shortcuts bound to it work.
"""
mock_exists.return_value = False # no sysfs entry: name fallback
device = MagicMock()
device.fileno.return_value = 10
device.info.bustype = 0x03 # BUS_USB — keyd deliberately looks real
device.name = "keyd virtual keyboard"
mock_input_device.return_value = device
backend = EvdevKeyboardBackend()
opened = backend._open_keyboard_device("/dev/input/event5")
assert opened is True
device.grab.assert_not_called()
mock_uinput.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")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.os.path.realpath")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.os.path.exists")
def test_open_detects_virtual_keyboard_by_sysfs_path(
self,
mock_exists: Mock,
mock_realpath: Mock,
mock_input_device: Mock,
mock_uinput: Mock,
) -> None:
"""A uinput-spawned keyboard is ungrabbed regardless of its name.
keyd names and clones its device however it likes; the reliable
signal is the sysfs device path landing under /devices/virtual.
"""
mock_exists.return_value = True
mock_realpath.return_value = "/sys/devices/virtual/input/input30"
device = MagicMock()
device.fileno.return_value = 10
device.info.bustype = 0x03 # BUS_USB
device.name = "Plausible Keyboard" # name a remapper could choose
mock_input_device.return_value = device
backend = EvdevKeyboardBackend()
opened = backend._open_keyboard_device("/dev/input/event5")
assert opened is True
device.grab.assert_not_called()
mock_uinput.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")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.os.path.realpath")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.os.path.exists")
def test_open_grabs_keyboard_on_real_sysfs_path(
self,
mock_exists: Mock,
mock_realpath: Mock,
mock_input_device: Mock,
mock_uinput: Mock,
) -> None:
"""A real sysfs path keeps suppression even for a remapper-ish name.
sysfs is ground truth for virtualness; the name list is only a
fallback for confined installs where sysfs cannot be read.
"""
mock_exists.return_value = True
mock_realpath.return_value = "/sys/devices/pci0000:00/input/input0"
device = MagicMock()
device.fileno.return_value = 10
device.info.bustype = 0x03
device.name = "keyd virtual keyboard"
device.active_keys.return_value = []
device.capabilities.return_value = _keyboard_only_capabilities()
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 backend._forwarders[10] is forwarder
@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 (voculinux)"
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._WriteOnlyUInput")
@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, mock_uinput: Mock, mock_write_only: Mock
) -> None:
"""Without /dev/uinput the device still works ungrabbed (keys leak)."""
device = MagicMock()
device.fileno.return_value = 10
device.info.bustype = 0x03
device.capabilities.return_value = _keyboard_only_capabilities()
mock_input_device.return_value = device
mock_uinput.side_effect = OSError("uinput denied")
mock_write_only.side_effect = OSError(errno.EACCES, "uinput denied")
backend = EvdevKeyboardBackend()
opened = backend._open_keyboard_device("/dev/input/event0")
assert opened is True
mock_write_only.assert_called_once()
device.grab.assert_not_called()
assert backend._forwarders == {}
assert device in backend.devices
@patch("vocalinux.ui.keyboard_backends.evdev_backend._WriteOnlyUInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
def test_open_retries_uinput_write_only_on_eacces(
self, mock_input_device: Mock, mock_uinput: Mock, mock_write_only: Mock
) -> None:
"""An O_RDWR refusal retries O_WRONLY so the device still gets grabbed.
Installs carrying the original write-only udev rule (MODE=0620)
denied evdev's O_RDWR open of /dev/uinput; a write-only descriptor
satisfies the kernel and keeps key suppression working.
"""
device = MagicMock()
device.fileno.return_value = 10
device.info.bustype = 0x03
device.active_keys.return_value = []
device.capabilities.return_value = _keyboard_only_capabilities()
mock_input_device.return_value = device
mock_uinput.side_effect = OSError("could not open uinput device in write mode")
forwarder = MagicMock()
forwarder.device.path = "/dev/input/event20"
mock_write_only.return_value = forwarder
backend = EvdevKeyboardBackend()
opened = backend._open_keyboard_device("/dev/input/event0")
assert opened is True
mock_write_only.assert_called_once()
device.grab.assert_called_once()
assert backend._forwarders[10] is forwarder
@patch("vocalinux.ui.keyboard_backends.evdev_backend._WriteOnlyUInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
def test_open_does_not_retry_write_only_on_other_oserror(
self, mock_input_device: Mock, mock_uinput: Mock, mock_write_only: Mock
) -> None:
"""A UInput failure with a real errno is not a permission retry."""
device = MagicMock()
device.fileno.return_value = 10
device.info.bustype = 0x03
device.capabilities.return_value = _keyboard_only_capabilities()
mock_input_device.return_value = device
mock_uinput.side_effect = OSError(errno.EIO, "i/o error")
backend = EvdevKeyboardBackend()
opened = backend._open_keyboard_device("/dev/input/event0")
assert opened is True
mock_write_only.assert_not_called()
device.grab.assert_not_called()
assert backend._forwarders == {}
assert device in backend.devices
def _queued_key_event(self, code: int, value: int, ts: float) -> MagicMock:
"""A queued EV_KEY event timestamped pre- or post-grab."""
event = MagicMock()
event.type = ecodes.EV_KEY
event.code = code
event.value = value
event.timestamp.return_value = ts
return event
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
def test_open_leaves_keyboard_with_held_key_ungrabbed(
self, mock_input_device: Mock, mock_uinput: Mock
) -> None:
"""A key held before the grab keeps the device ungrabbed."""
device = MagicMock()
device.fileno.return_value = 10
device.active_keys.return_value = [54]
device.capabilities.return_value = _keyboard_only_capabilities()
mock_input_device.return_value = device
forwarder = MagicMock()
mock_uinput.return_value = forwarder
backend = EvdevKeyboardBackend()
assert backend._open_keyboard_device("/dev/input/event0") is True
device.grab.assert_not_called()
device.ungrab.assert_not_called()
forwarder.close.assert_called_once()
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_refuses_grab_when_key_transition_already_queued(
self, mock_input_device: Mock, mock_uinput: Mock
) -> None:
"""A key transition queued at open refuses the grab entirely.
Everything drained before the grab was delivered to the
compositor's fd too, so a key event there means input is in
flight: grabbing would route a release through the clone for a
press the desktop saw on the physical device. The device stays
monitored ungrabbed and the rescan retries once quiet.
"""
press = self._queued_key_event(54, 1, time.time())
device = MagicMock()
device.fileno.return_value = 10
device.active_keys.return_value = []
device.read_one.side_effect = [press, None]
device.capabilities.return_value = _keyboard_only_capabilities()
mock_input_device.return_value = device
forwarder = MagicMock()
mock_uinput.return_value = forwarder
backend = EvdevKeyboardBackend()
assert backend._open_keyboard_device("/dev/input/event0") is True
device.grab.assert_not_called()
device.ungrab.assert_not_called()
forwarder.write_event.assert_not_called()
forwarder.close.assert_called_once()
assert backend._forwarders == {}
assert backend._pending_events == {}
assert device in backend.devices
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
def test_open_ungrabs_on_pre_grab_key_transition(
self, mock_input_device: Mock, mock_uinput: Mock
) -> None:
"""A press arriving between the pre-grab drain and the grab refuses it.
The event slipped past the empty pre-drain but is timestamped
before the grab, so it belongs to the compositor's view too — the
handover case. Its post-grab release still reaches the desktop
through the clone before the clone is closed, and engines get it
via pending events.
"""
pre = time.time() - 10
post = time.time() + 10
press = self._queued_key_event(54, 1, pre)
release = self._queued_key_event(54, 0, post)
device = MagicMock()
device.fileno.return_value = 10
device.active_keys.return_value = []
device.capabilities.return_value = _keyboard_only_capabilities()
# The pre-grab drain comes up empty; the straddle drain sees both.
device.read_one.side_effect = [None, press, release, None]
mock_input_device.return_value = device
forwarder = MagicMock()
mock_uinput.return_value = forwarder
backend = EvdevKeyboardBackend()
assert backend._open_keyboard_device("/dev/input/event0") is True
device.grab.assert_called_once()
device.ungrab.assert_called_once()
forwarder.write_event.assert_called_once_with(release)
forwarder.close.assert_called_once()
assert backend._forwarders == {}
assert backend._pending_events == {10: [release]}
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_replaying_post_grab_held_press(
self, mock_input_device: Mock, mock_uinput: Mock
) -> None:
"""A post-grab press still held is not replayed onto a dying clone.
Refusing the grab over a pre-grab transition, the held press would
leave the key logically down forever if replayed before close, so
only events for released keys are emitted.
"""
pre = time.time() - 10
post = time.time() + 10
stale_press = self._queued_key_event(54, 1, pre)
held_press = self._queued_key_event(100, 1, post)
device = MagicMock()
device.fileno.return_value = 10
device.active_keys.return_value = []
device.capabilities.return_value = _keyboard_only_capabilities()
device.read_one.side_effect = [None, stale_press, held_press, None]
mock_input_device.return_value = device
forwarder = MagicMock()
mock_uinput.return_value = forwarder
backend = EvdevKeyboardBackend()
assert backend._open_keyboard_device("/dev/input/event0") is True
device.ungrab.assert_called_once()
forwarder.write_event.assert_not_called()
assert backend._pending_events == {10: [held_press]}
@pytest.mark.parametrize(
"queued, expected_pending",
[
("none", "none"), # quiet handover: grab stays, nothing pending
("post_tap", "post_tap"), # post-grab tap replays, grab stays
("post_press", "post_press"), # held post-grab press keeps the grab
],
)
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
def test_open_keeps_grab_when_only_post_grab_events(
self,
mock_input_device: Mock,
mock_uinput: Mock,
queued: str,
expected_pending: str,
) -> None:
"""Events entirely after the grab are input, not handover evidence.
A tap delivered after the grab reached only our fd, so the grab is
kept and the events are queued for dispatch — dropping them would
lose the keystroke the compositor never saw.
"""
post = time.time() + 10
events = {
"post_tap": [
self._queued_key_event(54, 1, post),
self._queued_key_event(54, 0, post),
],
"post_press": [self._queued_key_event(100, 1, post)],
}
queue = events.get(queued, [])
device = MagicMock()
device.fileno.return_value = 10
device.active_keys.return_value = []
device.capabilities.return_value = _keyboard_only_capabilities()
# First None empties the pre-grab drain; the straddle drain follows.
device.read_one.side_effect = [None, *queue, None]
mock_input_device.return_value = device
forwarder = MagicMock()
mock_uinput.return_value = forwarder
backend = EvdevKeyboardBackend()
assert backend._open_keyboard_device("/dev/input/event0") is True
device.grab.assert_called_once()
device.ungrab.assert_not_called()
forwarder.close.assert_not_called()
forwarder.write_event.assert_not_called()
assert backend._forwarders == {10: forwarder}
expected = events.get(expected_pending, [])
if expected:
assert backend._pending_events == {10: expected}
else:
assert backend._pending_events == {}
@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, mock_uinput: Mock
) -> None:
"""A failed grab drops the forwarder but keeps the listener running."""
device = MagicMock()
device.fileno.return_value = 10
device.info.bustype = 0x03
device.capabilities.return_value = _keyboard_only_capabilities()
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_passthrough_skips_replay_when_nothing_is_withheld(self) -> None:
"""No synthetic presses or syncs are emitted without a pending modifier."""
backend = EvdevKeyboardBackend(shortcut="ctrl+ctrl")
fd = 10
clone = MagicMock()
backend._forwarders = {fd: clone}
backend._forwarded_held = {fd: set()}
backend._withheld_modifier = {fd: {}}
backend.set_pass_shortcut_to_apps(True)
backend.set_pass_shortcut_to_apps(True)
assert backend.pass_shortcut_to_apps is True
clone.write_event.assert_not_called()
clone.syn.assert_not_called()
backend.set_pass_shortcut_to_apps(False)
assert backend.pass_shortcut_to_apps is False
def test_passthrough_without_forwarder_handles_held_ctrl(self) -> None:
"""An ungrabbed keyboard safely transitions while Ctrl is held."""
backend = EvdevKeyboardBackend(shortcut="ctrl+ctrl", mode="toggle")
fd = 10
press = self._key_event(29, 1)
release = self._key_event(29, 0)
assert backend._event_is_shortcut(fd, press) is True
assert backend._withheld_modifier[fd][29][1] is False
assert fd not in backend._forwarders
backend.set_pass_shortcut_to_apps(True)
assert backend.pass_shortcut_to_apps is True
assert backend._withheld_modifier[fd][29][1] is True
assert backend._event_is_shortcut(fd, release) is False
assert backend._withheld_modifier[fd] == {}
def test_passthrough_sync_failure_releases_failed_forwarder(self) -> None:
"""A failed uinput frame sync must release the broken virtual keyboard."""
backend = EvdevKeyboardBackend(shortcut="ctrl+ctrl")
fd = 10
clone = MagicMock()
clone.syn.side_effect = OSError(errno.EIO, "uinput unavailable")
backend._forwarders = {fd: clone}
backend._forwarded_held = {fd: set()}
press = self._key_event(29, 1)
assert backend._event_is_shortcut(fd, press) is True
with patch.object(backend._hub, "_release_failed_forwarder") as release_failed:
backend.set_pass_shortcut_to_apps(True)
release_failed.assert_called_once_with(fd, clone)
clone.write_event.assert_called_once_with(press)
clone.syn.assert_called_once_with()
assert backend.pass_shortcut_to_apps is True
def test_passthrough_combo_keeps_swallowed_press_and_release_paired(self) -> None:
"""Enabling pass-through must not leak a release without its press."""
backend = EvdevKeyboardBackend(shortcut="ctrl+r", mode="toggle")
fd = 10
backend._combo_pressed = {29}
key_r = backend._combo_main_code
assert key_r is not None
assert backend._event_is_shortcut(fd, self._key_event(key_r, 1)) is True
backend.set_pass_shortcut_to_apps(True)
assert backend._event_is_shortcut(fd, self._key_event(key_r, 2)) is True
assert backend._event_is_shortcut(fd, self._key_event(key_r, 0)) is True
assert fd not in backend._combo_swallowed
assert backend._event_is_shortcut(fd, self._key_event(key_r, 1)) is False
assert backend._event_is_shortcut(fd, self._key_event(key_r, 0)) is False
def test_disabling_combo_passthrough_preserves_forwarded_key_up(self) -> None:
"""Switching off pass-through cannot strand a previously forwarded key."""
backend = EvdevKeyboardBackend(shortcut="ctrl+r", mode="toggle")
fd = 10
backend._forwarded_held = {fd: {19}}
backend.set_pass_shortcut_to_apps(True)
backend.set_pass_shortcut_to_apps(False)
assert backend._event_is_shortcut(fd, self._key_event(19, 0)) is False
def test_pass_through_double_tap_still_fires_and_forwards_ctrl(self) -> None:
"""Opted-in Ctrl taps must reach the app without losing detection (#950)."""
backend = EvdevKeyboardBackend(shortcut="ctrl+ctrl", mode="toggle")
backend.set_pass_shortcut_to_apps(True)
fired = threading.Event()
backend.register_toggle_callback(fired.set)
fd = 10
device = MagicMock()
forwarder = MagicMock()
backend._forwarders = {fd: forwarder}
backend._forwarded_held = {fd: set()}
events = [
self._key_event(29, 1),
self._key_event(29, 0),
self._key_event(29, 1),
self._key_event(29, 0),
]
with patch(
"vocalinux.ui.keyboard_backends.evdev_backend.time.time",
side_effect=[100.0, 100.15],
):
for event in events:
backend._hub._dispatch_event(fd, device, event, (backend,))
assert fired.wait(1.0)
assert [c.args[0] for c in forwarder.write_event.call_args_list] == events
assert backend._forwarded_held[fd] == set()
def test_pass_through_preserves_ctrl_state_for_mouse_wheel(self) -> None:
"""Ctrl reaches the virtual keyboard without waiting for a chord."""
backend = EvdevKeyboardBackend(shortcut="ctrl+ctrl", mode="toggle")
backend.set_pass_shortcut_to_apps(True)
fd = 10
device = MagicMock()
forwarder = MagicMock()
backend._forwarders = {fd: forwarder}
backend._forwarded_held = {fd: set()}
press = self._key_event(29, 1)
backend._hub._dispatch_event(fd, device, press, (backend,))
assert backend._forwarded_held[fd] == {29}
forwarder.write_event.assert_called_once_with(press)
# Mouse wheel can arrive on a completely separate evdev device.
release = self._key_event(29, 0)
backend._hub._dispatch_event(fd, device, release, (backend,))
assert backend._forwarded_held[fd] == set()
forwarder.write_event.assert_called_with(release)
def test_enabling_pass_through_replays_held_modifier_and_syncs(self) -> None:
backend = EvdevKeyboardBackend(shortcut="ctrl+ctrl", mode="toggle")
fd = 10
forwarder = MagicMock()
backend._forwarders = {fd: forwarder}
backend._forwarded_held = {fd: set()}
press = self._key_event(29, 1)
assert backend._event_is_shortcut(fd, press) is True
backend.set_pass_shortcut_to_apps(True)
forwarder.write_event.assert_called_once_with(press)
forwarder.syn.assert_called_once_with()
assert backend._event_is_shortcut(fd, self._key_event(29, 0)) is False
assert backend._withheld_modifier[fd] == {}
def test_concurrent_pass_through_enable_does_not_stick_ctrl(self) -> None:
"""A release cannot overtake replayed Ctrl-down on the virtual keyboard."""
backend = EvdevKeyboardBackend(shortcut="ctrl+ctrl", mode="toggle")
fd = 10
press, release = self._key_event(29, 1), self._key_event(29, 0)
device, forwarder = MagicMock(), MagicMock()
backend._forwarders = {fd: forwarder}
backend._forwarded_held = {fd: set()}
backend._hub._dispatch_event(fd, device, press, (backend,))
replay_started, allow_replay = threading.Event(), threading.Event()
release_started, release_done = threading.Event(), threading.Event()
written, errors = [], []
def write_event(event: InputEvent) -> None:
if event is press:
replay_started.set()
assert allow_replay.wait(3), "Replay timed out"
written.append(event)
def run(callback: Callable[[], None]) -> None:
try:
callback()
except BaseException as exc:
errors.append(exc)
def release_ctrl() -> None:
release_started.set()
try:
backend._hub._dispatch_event(fd, device, release, (backend,))
finally:
release_done.set()
forwarder.write_event.side_effect = write_event
enabler = threading.Thread(
target=run,
args=(lambda: backend.set_pass_shortcut_to_apps(True),),
)
releaser = threading.Thread(target=run, args=(release_ctrl,))
try:
enabler.start()
assert replay_started.wait(2), "Ctrl-down was not replayed"
releaser.start()
assert release_started.wait(2)
assert not release_done.wait(0.1), "Release slipped through during replay"
finally:
allow_replay.set()
if enabler.ident is not None:
enabler.join(timeout=3)
if releaser.ident is not None:
releaser.join(timeout=3)
assert not errors, errors
assert not enabler.is_alive() and not releaser.is_alive()
assert written == [press, release]
assert backend.pass_shortcut_to_apps is True
assert backend._withheld_modifier[fd] == {}
assert backend._forwarded_held[fd] == set()
forwarder.syn.assert_called_once_with()
def test_disabling_pass_through_keeps_existing_key_release(self) -> None:
backend = EvdevKeyboardBackend(shortcut="ctrl+ctrl", mode="toggle")
fd = 10
backend._forwarders = {fd: MagicMock()}
backend._forwarded_held = {fd: set()}
backend.set_pass_shortcut_to_apps(True)
backend._hub._forward_event(fd, self._key_event(29, 1))
backend.set_pass_shortcut_to_apps(False)
assert backend._event_is_shortcut(fd, self._key_event(29, 0)) is False
def test_pass_through_main_key_in_combo(self) -> None:
backend = EvdevKeyboardBackend(shortcut="ctrl+r", mode="toggle")
backend.set_pass_shortcut_to_apps(True)
assert backend._event_is_shortcut(10, self._key_event(29, 1)) is False
assert backend._event_is_shortcut(10, self._key_event(19, 1)) is False
assert backend._event_is_shortcut(10, self._key_event(19, 0)) 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, mock_find_devices: Mock, mock_uinput: Mock
) -> None:
"""stop() releases grabs (via close) and closes every uinput clone."""
device = MagicMock()
device.fileno.return_value = 10
device.info.bustype = 0x03
device.capabilities.return_value = _keyboard_only_capabilities()
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, mock_uinput: Mock
) -> None:
"""A device that loses the registration race is cleaned up fully."""
device = MagicMock()
device.fileno.return_value = 10
device.info.bustype = 0x03
device.capabilities.return_value = _keyboard_only_capabilities()
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: a withheld modifier press replays ahead of other keys.
RIGHTALT surviving with KEY_A reads as AltGr composition, not a
gesture, so the withheld press is replayed before the letter —
the app sees exactly what the user typed.
"""
backend = EvdevKeyboardBackend()
device = MagicMock()
device.fileno.return_value = 10
forwarder = MagicMock()
backend._forwarders = {10: forwarder}
backend._forwarded_held = {10: set()}
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()
assert forwarder.write_event.call_args_list == [
call(shortcut_event),
call(other_event),
]
def test_pure_modifier_press_withheld_lazily(self) -> None:
"""Production path: the withheld map is created on first modifier event.
The AltGr-replay machinery only works if the per-fd map exists; it
used to be populated by nothing in production, so withheld presses
were silently dropped and composition lost its modifier.
"""
backend = EvdevKeyboardBackend(shortcut="right_alt+right_alt")
fd = 10
forwarder = MagicMock()
backend._forwarders[fd] = forwarder
backend._forwarded_held[fd] = set()
press = self._key_event(100, 1)
key_e = self._key_event(18, 1)
assert backend._event_is_shortcut(fd, press) is True
assert 100 in backend._withheld_modifier[fd]
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]
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 TestForwardingWatchdog:
"""A stalled monitor must not hold grabs while nothing forwards (#1021).
Suspend/resume can leave the reader wedged with every EVIOCGRAB held:
keystrokes queue on hardware only a dead reader can see, and the lock
screen goes dark. The watchdog stamps the monitor's liveness through
``_monitor_heartbeat``; a heartbeat past ``_MONITOR_STALL_SECONDS``
releases every grab, and a recovering monitor re-arms the suspended
devices through the standard open->clone->grab path.
"""
def _hub_with_grabbed_device(
self, heartbeat: float, generation: int = 3
) -> tuple[EvdevDeviceHub, MagicMock, MagicMock]:
"""Hub state as the monitor left it: one grabbed device + clone."""
backend = EvdevKeyboardBackend()
hub = backend._hub
hub.reset()
hub.running = True
hub._generation = generation
hub._monitor_heartbeat = heartbeat
device = MagicMock()
device.fileno.return_value = 10
device.name = "Keyboard"
forwarder = MagicMock()
hub.devices = [device]
hub.device_fds = [10]
hub.device_paths = {"/dev/input/event0"}
hub._device_paths_by_fd = {10: "/dev/input/event0"}
hub._fd_generation = {10: generation}
hub._forwarders = {10: forwarder}
hub._forwarded_held = {10: set()}
return hub, device, forwarder
def test_stalled_monitor_releases_every_grab(self) -> None:
"""A stale heartbeat ungrabs the device, closes its clone, suspends it."""
hub, device, forwarder = self._hub_with_grabbed_device(
heartbeat=time.monotonic() - _MONITOR_STALL_SECONDS - 1.0
)
try:
hub._check_forwarding_health(hub._generation)
device.ungrab.assert_called_once()
forwarder.close.assert_called_once()
assert hub._forwarders == {}
assert hub._grab_suspended_devices == {10: device}
# Still monitored, ungrabbed: keys reach the lock screen and the
# shortcut keeps detecting, just unsuppressed.
assert device in hub.devices
assert hub.device_fds == [10]
finally:
hub.reset()
def test_fresh_heartbeat_keeps_grabs(self) -> None:
"""A live monitor's grabs are never touched by the watchdog."""
hub, device, forwarder = self._hub_with_grabbed_device(heartbeat=time.monotonic())
try:
hub._check_forwarding_health(hub._generation)
device.ungrab.assert_not_called()
forwarder.close.assert_not_called()
assert hub._forwarders == {10: forwarder}
assert hub._grab_suspended_devices == {}
finally:
hub.reset()
def test_unstarted_monitor_is_ignored(self) -> None:
"""Heartbeat 0.0 means the first iteration has not run: not a stall."""
hub, device, forwarder = self._hub_with_grabbed_device(heartbeat=0.0)
try:
hub._check_forwarding_health(hub._generation)
device.ungrab.assert_not_called()
forwarder.close.assert_not_called()
assert hub._forwarders == {10: forwarder}
finally:
hub.reset()
def test_stall_skips_fds_owned_by_a_newer_generation(self) -> None:
"""An fd recycled for a newer generation is never released under it."""
hub, device, forwarder = self._hub_with_grabbed_device(
heartbeat=time.monotonic() - _MONITOR_STALL_SECONDS - 1.0, generation=3
)
hub._fd_generation[10] = 4 # reopened by the newer generation
try:
hub._check_forwarding_health(3)
device.ungrab.assert_not_called()
forwarder.close.assert_not_called()
assert hub._forwarders == {10: forwarder}
assert hub._grab_suspended_devices == {}
finally:
hub.reset()
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
@patch("vocalinux.ui.keyboard_backends.evdev_backend._is_virtual_keyboard")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
def test_recovery_reopens_suspended_devices_through_grab_path(
self,
mock_uinput: Mock,
mock_input_device: Mock,
mock_is_virtual: Mock,
mock_find: Mock,
) -> None:
"""A live heartbeat re-arms suspended devices via rescan and re-grab."""
hub, old_device, _ = self._hub_with_grabbed_device(heartbeat=time.monotonic())
hub._grab_suspended_devices = {10: old_device}
new_device = MagicMock()
new_device.fileno.return_value = 20
new_device.info.bustype = 0x03
new_device.name = "Keyboard"
new_device.capabilities.return_value = _keyboard_only_capabilities()
new_device.active_keys.return_value = []
mock_input_device.return_value = new_device
new_forwarder = MagicMock()
new_forwarder.device.path = "/dev/input/event9"
mock_uinput.return_value = new_forwarder
mock_is_virtual.return_value = False
mock_find.return_value = ["/dev/input/event0"]
try:
# The monitor re-arms suspended devices at its loop top.
hub._rearm_suspended_devices(hub._generation)
old_device.close.assert_called_once()
assert hub._grab_suspended_devices == {}
# Reopened through the normal path: grabbed, cloned, registered.
new_device.grab.assert_called_once()
assert hub._forwarders[20] is new_forwarder
assert new_device in hub.devices
assert hub.device_paths == {"/dev/input/event0"}
finally:
hub.reset()
@patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices")
def test_recovery_skips_suspended_device_replaced_by_another(self, mock_find: Mock) -> None:
"""A suspended fd recycled for a different device tears nothing down."""
mock_find.return_value = []
hub, old_device, _ = self._hub_with_grabbed_device(heartbeat=time.monotonic())
hub._grab_suspended_devices = {10: old_device}
other_device = MagicMock()
other_device.fileno.return_value = 10
hub.devices = [other_device] # fd 10 now belongs to a newer device
try:
hub._rearm_suspended_devices(hub._generation)
old_device.close.assert_not_called()
other_device.close.assert_not_called()
assert other_device in hub.devices
assert hub._grab_suspended_devices == {}
finally:
hub.reset()
def test_monitor_rearms_suspended_devices_in_loop(self) -> None:
"""The monitor retires watchdog-released devices between selects.
Recovery must run on the monitor thread: closing an fd from the
watchdog would leave a dead descriptor in the select set and the
monitor's error handler would tear down every device.
"""
hub, old_device, _ = self._hub_with_grabbed_device(heartbeat=0.0)
hub._grab_suspended_devices = {10: old_device}
rearmed: list[int] = []
def _fake_rearm(generation: int) -> None:
rearmed.append(generation)
hub.running = False
try:
with (
patch.object(hub, "_rearm_suspended_devices", side_effect=_fake_rearm),
patch(
"vocalinux.ui.keyboard_backends.evdev_backend.select.select",
return_value=([], [], []),
),
patch(
"vocalinux.ui.keyboard_backends.evdev_backend.time.monotonic",
side_effect=[0.0, 0.0],
),
):
hub._monitor_devices(generation=hub._generation)
assert rearmed == [3]
finally:
hub.reset()
def test_watchdog_exits_when_running_stops(self) -> None:
"""The watchdog returns immediately once the layer is not running."""
hub, _, _ = self._hub_with_grabbed_device(heartbeat=time.monotonic())
hub.running = False
try:
# Returns instead of looping: running is already False.
hub._forwarding_watchdog(generation=hub._generation)
finally:
hub.reset()
def test_unregister_stops_and_joins_watchdog(self) -> None:
"""Teardown clears the watchdog slot once the last engine leaves."""
hub, _, _ = self._hub_with_grabbed_device(heartbeat=time.monotonic())
engine = EvdevKeyboardBackend(shortcut="ctrl+ctrl")
engine.active = True
watchdog = MagicMock(spec=threading.Thread)
hub._watchdog_thread = watchdog
try:
with hub._engines_lock:
hub._engines.add(engine)
hub.unregister(engine)
watchdog.join.assert_called_once()
assert hub._watchdog_thread is None
assert hub.running is False
finally:
hub.reset()
class TestWriteOnlyUInput:
"""The write-only clone drives the real _uinput setup on an O_WRONLY fd."""
@contextlib.contextmanager
def _built_uinput(self, fd: int = 42):
"""Instantiate the real _WriteOnlyUInput with os/evdev calls mocked."""
from vocalinux.ui.keyboard_backends import evdev_backend as eb
device = MagicMock()
with (
patch.object(eb._WriteOnlyUInput, "_verify"),
patch.object(eb.os, "open", return_value=fd) as mock_open,
patch.object(eb._uinput, "set_phys") as set_phys,
patch.object(eb._uinput, "enable") as enable,
patch.object(eb._uinput, "setup") as setup,
patch.object(eb._uinput, "create") as create,
patch.object(eb._WriteOnlyUInput, "_find_device", return_value=device),
patch.object(eb._uinput, "write") as write,
):
uinput = eb._WriteOnlyUInput(events={1: [30]}, name="kbd (voculinux)", bustype=0x06)
yield uinput, mock_open, set_phys, enable, setup, create, write, device
def test_opens_wronly_and_runs_setup(self) -> None:
"""Constructor opens O_WRONLY and performs the full uinput setup."""
with self._built_uinput(fd=42) as (
uinput,
mock_open,
set_phys,
enable,
setup,
create,
_,
device,
):
mock_open.assert_called_once_with("/dev/uinput", os.O_WRONLY | os.O_NONBLOCK)
set_phys.assert_called_once_with(42, "py-evdev-uinput")
enable.assert_called_once_with(42, 1, 30)
assert setup.call_args[0][0] == 42
create.assert_called_once_with(42)
assert uinput.fd == 42
assert uinput.device is device
def test_emits_through_uinput_write(self) -> None:
"""write/write_event/syn bypass need_write's O_RDWR check."""
import evdev as real_evdev
with self._built_uinput(fd=42) as (uinput, _, _, _, _, _, write, _):
uinput.write(1, 30, 1)
uinput.write_event(real_evdev.InputEvent(0, 0, 1, 30, 0))
uinput.syn()
assert write.call_args_list == [
call(42, 1, 30, 1),
call(42, 1, 30, 0),
call(42, 0, 0, 0), # EV_SYN / SYN_REPORT
]
def test_failed_event_prep_closes_descriptor(self) -> None:
"""A failure before the _uinput calls still releases our fd."""
from vocalinux.ui.keyboard_backends import evdev_backend as eb
with (
patch.object(eb._WriteOnlyUInput, "_verify"),
patch.object(eb.os, "open", return_value=42),
patch.object(eb.os, "close") as mock_close,
patch.object(
eb._WriteOnlyUInput, "_prepare_events", side_effect=TypeError("bad events")
),
):
with pytest.raises(TypeError):
eb._WriteOnlyUInput(events=None, name="kbd (voculinux)", bustype=0x06)
mock_close.assert_called_once_with(42)
def test_clone_device_name_caps_encoded_bytes(self) -> None:
"""A long multibyte device name truncates by bytes, keeping the suffix."""
device = MagicMock()
device.name = "\u2328" * 40 # 3 bytes each, 120 bytes total
name = _clone_device_name(device)
assert name.endswith(" (voculinux)")
assert len(name.encode("utf-8")) <= 79
def test_clone_device_name_keeps_short_names(self) -> None:
"""Ordinary names keep the full base plus suffix."""
device = MagicMock()
device.name = "AT Translated Set 2 keyboard"
assert _clone_device_name(device) == "AT Translated Set 2 keyboard (voculinux)"
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 = {}
device.active_keys.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_dispatch_pending_events_tolerates_empty_read(self) -> None:
"""A pending-only fd replays stored events without disconnecting.
Regression test for "replay drops a working keyboard": when stored
startup events mark a device readable but no new input arrived,
device.read() raises BlockingIOError. That error must not escape to
the monitor, which treats OSError as a disconnect and would close a
healthy keyboard and its clone.
"""
engine = EvdevKeyboardBackend(shortcut="ctrl+ctrl")
hub = self._registered(engine)
engine._handle_key_event = MagicMock()
release = self._key_event(29, 0)
hub._pending_events[10] = [release]
device = MagicMock()
device.read.side_effect = BlockingIOError(11, "Resource temporarily unavailable")
hub._dispatch_events(10, device)
engine._handle_key_event.assert_called_once_with(release, device)
assert hub._pending_events == {}
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()
Reference in New Issue View Git Blame Copy Permalink