diff --git a/src/vocalinux/text_injection/text_injector.py b/src/vocalinux/text_injection/text_injector.py index f0ccee1..8def2ed 100644 --- a/src/vocalinux/text_injection/text_injector.py +++ b/src/vocalinux/text_injection/text_injector.py @@ -6,6 +6,7 @@ application, supporting both X11 and Wayland environments. """ import logging +import math import os import shutil import subprocess @@ -969,6 +970,96 @@ class TextInjector: logger.warning(f"Paste simulation failed: {e}") return False + # evdev keycodes for modifier keys. If any of these is still physically held + # when a Wayland injection fires, the injected keystrokes are modified: a + # held Alt turns the Ctrl+V paste into Ctrl+Alt+V (nothing pastes), and a + # held modifier turns typed letters into shortcuts. Because toggle/PTT + # shortcuts are themselves modifiers (e.g. Alt+R) and transcription can + # finish in tens of milliseconds, the user is often still holding the key + # when injection starts, causing intermittent "nothing pasted" failures. + _MODIFIER_KEYCODES = frozenset({29, 97, 56, 100, 42, 54, 125, 126}) + + def _held_modifier_keycodes(self) -> set: + """Return modifier keycodes currently held on any physical keyboard.""" + try: + import evdev + from evdev import ecodes + except ImportError: + return set() + + held: set = set() + try: + for path in evdev.list_devices(): + try: + device = evdev.InputDevice(path) + except (OSError, PermissionError): + continue + try: + if ecodes.EV_KEY not in device.capabilities(): + continue + held |= set(device.active_keys()) & self._MODIFIER_KEYCODES + except OSError: + continue + finally: + device.close() + except Exception as e: + logger.debug(f"Could not read modifier key state: {e}") + return held + + _DEFAULT_INJECT_MODIFIER_WAIT = 1.0 + + def _injection_modifier_wait_seconds(self) -> float: + """Return the sanitized max seconds to wait for modifiers to release. + + Reads VOCALINUX_INJECT_MODIFIER_WAIT and falls back to the default for + anything unusable: a non-numeric value, or a non-finite one. In + particular ``inf`` parses fine and is positive, so without this guard it + would make the wait deadline infinite and block injection forever while a + modifier stays held. + """ + raw = os.environ.get("VOCALINUX_INJECT_MODIFIER_WAIT") + if raw is None: + return self._DEFAULT_INJECT_MODIFIER_WAIT + try: + value = float(raw) + except ValueError: + return self._DEFAULT_INJECT_MODIFIER_WAIT + if not math.isfinite(value): + logger.warning( + "VOCALINUX_INJECT_MODIFIER_WAIT=%r is not a finite number; " "using default %.1fs", + raw, + self._DEFAULT_INJECT_MODIFIER_WAIT, + ) + return self._DEFAULT_INJECT_MODIFIER_WAIT + return value + + def _wait_for_modifiers_released(self) -> None: + """Wait briefly for shortcut modifier keys to be released before injecting. + + Returns immediately if no modifier is held (the common case), so this + adds no latency unless the user is still holding their toggle/PTT + shortcut. Bounded by VOCALINUX_INJECT_MODIFIER_WAIT seconds (default 1.0; + set to 0 to disable). Best-effort: if evdev/permissions are unavailable + it simply proceeds. + """ + max_wait = self._injection_modifier_wait_seconds() + if max_wait <= 0: + return + + deadline = time.monotonic() + max_wait + waited = False + while time.monotonic() < deadline: + if not self._held_modifier_keycodes(): + if waited: + logger.debug("Modifier keys released; proceeding with injection") + return + waited = True + time.sleep(0.015) + logger.debug( + f"Modifier keys still held after {max_wait:.2f}s; injecting anyway " + "(paste/typing may be affected)" + ) + def _inject_with_wayland_tool(self, text: str): """ Inject text using a Wayland-compatible tool (wtype or ydotool). @@ -984,6 +1075,11 @@ class TextInjector: Raises: subprocess.CalledProcessError: If the tool fails, with stderr captured """ + # Wait for the shortcut modifier(s) to be released first, so a held Alt + # doesn't turn the Ctrl+V paste into Ctrl+Alt+V (nothing pastes) or + # modify typed keys. + self._wait_for_modifiers_released() + # ydotool emits *positional* evdev keycodes, so `ydotool type` is # re-interpreted through the active keyboard layout, which it assumes is # US QWERTY. On any other layout (AZERTY, QWERTZ, Dvorak, ...) the text diff --git a/tests/test_injection_modifier_wait.py b/tests/test_injection_modifier_wait.py new file mode 100644 index 0000000..99f3a96 --- /dev/null +++ b/tests/test_injection_modifier_wait.py @@ -0,0 +1,258 @@ +"""Tests for deferring Wayland injection until shortcut modifiers are released. + +Regression coverage for intermittent "nothing pasted" failures: when a +modifier-based toggle/PTT shortcut (e.g. Alt+R) is still held as injection +fires, the simulated Ctrl+V becomes Ctrl+Alt+V and nothing pastes. +""" + +import types +from unittest.mock import MagicMock, patch + +import pytest + +from vocalinux.text_injection.text_injector import TextInjector + + +def _bare_injector(): + """A TextInjector instance without running the heavy __init__.""" + return TextInjector.__new__(TextInjector) + + +def _fake_evdev(devices): + """Build a fake ``evdev`` module for _held_modifier_keycodes tests. + + devices: list of ``(active_keys, has_ev_key, raises)`` describing each + device that list_devices() reports. + """ + ecodes = types.SimpleNamespace(EV_KEY=1) + + class FakeDevice: + def __init__(self, active, has_ev_key): + self._active = active + self._caps = {ecodes.EV_KEY: []} if has_ev_key else {} + self.closed = False + + def capabilities(self): + return self._caps + + def active_keys(self): + return self._active + + def close(self): + self.closed = True + + paths = [f"/dev/input/event{i}" for i in range(len(devices))] + built = {} + for path, (active, has_ev_key, raises) in zip(paths, devices): + built[path] = (FakeDevice(active, has_ev_key), raises) + + mod = types.ModuleType("evdev") + mod.ecodes = ecodes + mod.list_devices = lambda: list(paths) + + def input_device(path): + device, raises = built[path] + if raises: + raise raises + return device + + mod.InputDevice = input_device + return mod + + +class TestWaitForModifiersReleased: + def test_returns_immediately_when_no_modifier_held(self): + inj = _bare_injector() + with ( + patch.object(inj, "_held_modifier_keycodes", return_value=set()) as held, + patch("time.sleep") as sleep, + ): + inj._wait_for_modifiers_released() + assert held.call_count == 1 # checked once, cleared, no polling loop + sleep.assert_not_called() + + def test_waits_until_modifier_released(self): + inj = _bare_injector() + # Held for the first two checks, then released. + states = [{56}, {56}, set()] + with ( + patch.object(inj, "_held_modifier_keycodes", side_effect=states) as held, + patch("time.sleep"), + ): + inj._wait_for_modifiers_released() + assert held.call_count == 3 + + def test_disabled_via_env(self, monkeypatch): + inj = _bare_injector() + monkeypatch.setenv("VOCALINUX_INJECT_MODIFIER_WAIT", "0") + with patch.object(inj, "_held_modifier_keycodes") as held: + inj._wait_for_modifiers_released() + held.assert_not_called() + + def test_gives_up_after_timeout(self, monkeypatch): + inj = _bare_injector() + monkeypatch.setenv("VOCALINUX_INJECT_MODIFIER_WAIT", "0.05") + # Always held -> must still return (best-effort) rather than hang. + with patch.object(inj, "_held_modifier_keycodes", return_value={56}), patch("time.sleep"): + inj._wait_for_modifiers_released() # should return within the timeout + + +class TestWaylandInjectionWaitsFirst: + def test_clipboard_paste_path_waits_before_pasting(self): + inj = _bare_injector() + inj.wayland_tool = "ydotool" + calls = [] + with ( + patch.object( + inj, "_wait_for_modifiers_released", side_effect=lambda: calls.append("wait") + ), + patch.object( + inj, + "_inject_via_clipboard_paste", + side_effect=lambda text: calls.append("paste") or True, + ), + ): + inj._inject_with_wayland_tool("hello") + # The modifier wait must happen before the paste chord is sent. + assert calls == ["wait", "paste"] + + def test_wtype_path_waits_before_typing(self): + inj = _bare_injector() + inj.wayland_tool = "wtype" + calls = [] + with ( + patch.object( + inj, "_wait_for_modifiers_released", side_effect=lambda: calls.append("wait") + ), + patch("subprocess.run", side_effect=lambda *a, **k: calls.append("run") or MagicMock()), + ): + inj._inject_with_wayland_tool("hello") + assert calls[0] == "wait" + assert "run" in calls + + +class TestHeldModifierKeycodes: + def test_returns_only_modifier_codes(self): + inj = _bare_injector() + # event0: Alt (56) + 'a' (30) held; event1: nothing. + fake = _fake_evdev([([56, 30], True, None), ([], True, None)]) + with patch.dict("sys.modules", {"evdev": fake}): + held = inj._held_modifier_keycodes() + assert held == {56} # 'a' (30) is not a modifier and is excluded + + def test_unions_across_devices_split_keyboard(self): + inj = _bare_injector() + # Left half holds Ctrl (29), right half holds Shift (42). + fake = _fake_evdev([([29], True, None), ([42], True, None)]) + with patch.dict("sys.modules", {"evdev": fake}): + held = inj._held_modifier_keycodes() + assert held == {29, 42} + + def test_skips_non_keyboard_devices(self): + inj = _bare_injector() + # A device without EV_KEY (e.g. a mouse) must be ignored. + fake = _fake_evdev([([56], False, None)]) + with patch.dict("sys.modules", {"evdev": fake}): + held = inj._held_modifier_keycodes() + assert held == set() + + def test_skips_unopenable_devices(self): + inj = _bare_injector() + fake = _fake_evdev([([56], True, PermissionError("denied"))]) + with patch.dict("sys.modules", {"evdev": fake}): + held = inj._held_modifier_keycodes() + assert held == set() + + def test_no_evdev_returns_empty(self): + inj = _bare_injector() + # Simulate evdev not installed. + with patch.dict("sys.modules", {"evdev": None}): + held = inj._held_modifier_keycodes() + assert held == set() + + def test_wait_polls_real_reader(self, monkeypatch): + # Exercise _wait_for_modifiers_released against the real reader: held + # once, then clear. + inj = _bare_injector() + states = [_fake_evdev([([56], True, None)]), _fake_evdev([([], True, None)])] + calls = {"n": 0} + + def held(): + mod = states[min(calls["n"], len(states) - 1)] + calls["n"] += 1 + with patch.dict("sys.modules", {"evdev": mod}): + return TextInjector._held_modifier_keycodes(inj) + + monkeypatch.setattr(inj, "_held_modifier_keycodes", held) + with patch("time.sleep"): + inj._wait_for_modifiers_released() + assert calls["n"] >= 2 + + +class TestHeldModifierKeycodesErrorPaths: + def test_list_devices_failure_returns_empty(self): + inj = _bare_injector() + mod = types.ModuleType("evdev") + mod.ecodes = types.SimpleNamespace(EV_KEY=1) + + def boom(): + raise RuntimeError("enumeration failed") + + mod.list_devices = boom + with patch.dict("sys.modules", {"evdev": mod}): + assert inj._held_modifier_keycodes() == set() + + def test_device_read_oserror_is_skipped(self): + inj = _bare_injector() + mod = types.ModuleType("evdev") + mod.ecodes = types.SimpleNamespace(EV_KEY=1) + + class RaisingDevice: + def capabilities(self): + return {1: []} + + def active_keys(self): + raise OSError("device went away mid-read") + + def close(self): + pass + + mod.list_devices = lambda: ["/dev/input/event0"] + mod.InputDevice = lambda path: RaisingDevice() + with patch.dict("sys.modules", {"evdev": mod}): + assert inj._held_modifier_keycodes() == set() + + +class TestInjectionModifierWaitSeconds: + @pytest.mark.parametrize("raw,expected", [("2.5", 2.5), ("0", 0.0), ("0.5", 0.5)]) + def test_valid_values(self, raw, expected, monkeypatch): + inj = _bare_injector() + monkeypatch.setenv("VOCALINUX_INJECT_MODIFIER_WAIT", raw) + assert inj._injection_modifier_wait_seconds() == expected + + @pytest.mark.parametrize("raw", ["inf", "-inf", "nan", "Infinity", "NaN"]) + def test_non_finite_falls_back_to_default(self, raw, monkeypatch): + inj = _bare_injector() + monkeypatch.setenv("VOCALINUX_INJECT_MODIFIER_WAIT", raw) + assert inj._injection_modifier_wait_seconds() == 1.0 + + def test_non_numeric_falls_back_to_default(self, monkeypatch): + inj = _bare_injector() + monkeypatch.setenv("VOCALINUX_INJECT_MODIFIER_WAIT", "abc") + assert inj._injection_modifier_wait_seconds() == 1.0 + + def test_unset_uses_default(self, monkeypatch): + inj = _bare_injector() + monkeypatch.delenv("VOCALINUX_INJECT_MODIFIER_WAIT", raising=False) + assert inj._injection_modifier_wait_seconds() == 1.0 + + def test_inf_does_not_block_forever(self, monkeypatch): + # Regression: with 'inf', the wait must still terminate (bounded by the + # finite default) instead of looping forever while a modifier is held. + inj = _bare_injector() + monkeypatch.setenv("VOCALINUX_INJECT_MODIFIER_WAIT", "inf") + # Simulate the clock crossing the finite default deadline in a few steps. + clock = iter([0.0, 0.5, 1.5]) + monkeypatch.setattr("time.monotonic", lambda: next(clock, 2.0)) + with patch.object(inj, "_held_modifier_keycodes", return_value={56}), patch("time.sleep"): + inj._wait_for_modifiers_released() # must return, not hang