diff --git a/distro-packages/Slint/README b/distro-packages/Slint/README index a275a04..241befc 100644 --- a/distro-packages/Slint/README +++ b/distro-packages/Slint/README @@ -23,7 +23,6 @@ This package requires the following packages, all available from SlackBuilds.org - gtk3 - liblouis - libwnck3 -- python3-atspi - python3-cairo - python3-dasbus - python3-gobject diff --git a/distro-packages/Slint/cthulhu-info b/distro-packages/Slint/cthulhu-info index caf48f9..17d9444 100644 --- a/distro-packages/Slint/cthulhu-info +++ b/distro-packages/Slint/cthulhu-info @@ -5,6 +5,6 @@ DOWNLOAD="https://git.stormux.org/storm/cthulhu.git" MD5SUM="SKIP" DOWNLOAD_x86_64="" MD5SUM_x86_64="" -REQUIRES="at-spi2-core brltty gobject-introspection gstreamer gst-plugins-base gst-plugins-good gtk3 liblouis libwnck3 python3-atspi python3-cairo python3-dasbus python3-gobject python3-pluggy python3-pywayland python3-setproctitle python3-tomlkit speech-dispatcher" +REQUIRES="at-spi2-core brltty gobject-introspection gstreamer gst-plugins-base gst-plugins-good gtk3 liblouis libwnck3 python3-cairo python3-dasbus python3-gobject python3-pluggy python3-pywayland python3-setproctitle python3-tomlkit speech-dispatcher" MAINTAINER="Storm Dragon" EMAIL="storm_dragon@stormux.org" diff --git a/src/cthulhu/input_event_manager.py b/src/cthulhu/input_event_manager.py index ffd81c1..6830299 100644 --- a/src/cthulhu/input_event_manager.py +++ b/src/cthulhu/input_event_manager.py @@ -46,7 +46,6 @@ import gi gi.require_version("Atspi", "2.0") from gi.repository import Atspi from gi.repository import GLib -import pyatspi from . import debug from . import focus_manager @@ -75,6 +74,14 @@ _X11_CONSUMABLE_MODIFIER_MASKS = ( _X11_SHIFT_LOCK_MASK | _X11_NUM_LOCK_MASK, _X11_SHIFT_MASK | _X11_SHIFT_LOCK_MASK | _X11_NUM_LOCK_MASK, ) +_X11_CONSUMABLE_EVENT_TYPES = ( + (1 << int(Atspi.EventType.KEY_PRESSED_EVENT)) + | (1 << int(Atspi.EventType.KEY_RELEASED_EVENT)) +) +_X11_CONSUMABLE_SYNC_TYPE = ( + Atspi.KeyListenerSyncType.SYNCHRONOUS + | Atspi.KeyListenerSyncType.CANCONSUME +) _X11_CONSUMABLE_KEYSYMS = ( "a", "b", "c", "d", "e", "f", "g", "h", "i", "k", "l", "m", "o", "p", "q", "r", "s", "t", "u", "v", "x", "y", @@ -110,6 +117,8 @@ class InputEventManager: self._key_pressed_id: int = 0 self._key_released_id: int = 0 self._selective_legacy_listener_active: bool = False + self._selective_legacy_listener: Optional[Atspi.DeviceListener] = None + self._selective_legacy_registered_masks: List[int] = [] self._selective_legacy_key_set: List[Atspi.KeyDefinition] = [] self._selective_legacy_keycodes: set[int] = set() self._pointer_moved_id: int = 0 @@ -179,15 +188,7 @@ class InputEventManager: } msg = "INPUT EVENT MANAGER: Adding selective consumable X11 key listener." debug.print_message(debug.LEVEL_INFO, msg, True) - pyatspi.Registry.registerKeystrokeListener( - self._process_selective_legacy_keyboard_event, - key_set=self._selective_legacy_key_set, - mask=list(_X11_CONSUMABLE_MODIFIER_MASKS), - kind=(Atspi.EventType.KEY_PRESSED_EVENT, Atspi.EventType.KEY_RELEASED_EVENT), - synchronous=True, - preemptive=True, - ) - self._selective_legacy_listener_active = True + self._register_selective_legacy_key_listener() else: msg = "INPUT EVENT MANAGER: Using legacy AT-SPI key watcher." debug.print_message(debug.LEVEL_INFO, msg, True) @@ -201,16 +202,7 @@ class InputEventManager: msg = "INPUT EVENT MANAGER: Stopping key watcher." debug.print_message(debug.LEVEL_INFO, msg, True) - if self._selective_legacy_listener_active: - pyatspi.Registry.deregisterKeystrokeListener( - self._process_selective_legacy_keyboard_event, - key_set=self._selective_legacy_key_set, - mask=list(_X11_CONSUMABLE_MODIFIER_MASKS), - kind=(Atspi.EventType.KEY_PRESSED_EVENT, Atspi.EventType.KEY_RELEASED_EVENT), - ) - self._selective_legacy_listener_active = False - self._selective_legacy_key_set = [] - self._selective_legacy_keycodes.clear() + self._deregister_selective_legacy_key_listener() if self._device is not None: for handlerId in (self._key_pressed_id, self._key_released_id): if handlerId: @@ -1284,7 +1276,95 @@ class InputEventManager: # pylint: disable=too-many-arguments # pylint: disable=too-many-positional-arguments - def _process_selective_legacy_keyboard_event(self, event: Atspi.DeviceEvent) -> bool: + def _register_selective_legacy_key_listener(self) -> bool: + """Registers the consumable X11 listener directly through libatspi.""" + + self._selective_legacy_listener = Atspi.DeviceListener.new( + self._process_selective_legacy_keyboard_event, + None, + ) + if self._selective_legacy_listener is None: + msg = "INPUT EVENT MANAGER: Could not create selective consumable X11 listener." + debug.print_message(debug.LEVEL_WARNING, msg, True) + self._deregister_selective_legacy_key_listener() + return False + + for modifierMask in _X11_CONSUMABLE_MODIFIER_MASKS: + try: + registered = Atspi.register_keystroke_listener( + self._selective_legacy_listener, + self._selective_legacy_key_set, + modifierMask, + _X11_CONSUMABLE_EVENT_TYPES, + _X11_CONSUMABLE_SYNC_TYPE, + ) + except GLib.GError as error: + msg = ( + "INPUT EVENT MANAGER: Could not register selective consumable " + f"X11 listener for modifier mask {modifierMask}: {error}" + ) + debug.print_message(debug.LEVEL_WARNING, msg, True) + registered = False + + if not registered: + msg = ( + "INPUT EVENT MANAGER: Selective consumable X11 listener registration " + f"failed for modifier mask {modifierMask}." + ) + debug.print_message(debug.LEVEL_WARNING, msg, True) + self._deregister_selective_legacy_key_listener() + return False + + self._selective_legacy_registered_masks.append(modifierMask) + + self._selective_legacy_listener_active = True + return True + + def _deregister_selective_legacy_key_listener(self) -> None: + """Deregisters all masks associated with the consumable X11 listener.""" + + listener = self._selective_legacy_listener + failedMasks = [] + if listener is not None: + for modifierMask in self._selective_legacy_registered_masks: + try: + deregistered = Atspi.deregister_keystroke_listener( + listener, + self._selective_legacy_key_set, + modifierMask, + _X11_CONSUMABLE_EVENT_TYPES, + ) + except GLib.GError as error: + msg = ( + "INPUT EVENT MANAGER: Could not deregister selective consumable " + f"X11 listener for modifier mask {modifierMask}: {error}" + ) + debug.print_message(debug.LEVEL_WARNING, msg, True) + deregistered = False + + if not deregistered: + failedMasks.append(modifierMask) + + self._selective_legacy_registered_masks = failedMasks + if failedMasks: + self._selective_legacy_listener_active = True + msg = ( + "INPUT EVENT MANAGER: Preserving selective consumable X11 listener " + f"for {len(failedMasks)} failed deregistration(s)." + ) + debug.print_message(debug.LEVEL_WARNING, msg, True) + return + + self._selective_legacy_listener_active = False + self._selective_legacy_listener = None + self._selective_legacy_key_set = [] + self._selective_legacy_keycodes.clear() + + def _process_selective_legacy_keyboard_event( + self, + event: Atspi.DeviceEvent, + _userData: object = None, + ) -> bool: """Returns an X11 consume decision for unmodified browse-navigation keys.""" pressed = event.type == Atspi.EventType.KEY_PRESSED_EVENT diff --git a/tests/test_input_event_manager_key_watcher_regressions.py b/tests/test_input_event_manager_key_watcher_regressions.py index e9f0791..b959f26 100644 --- a/tests/test_input_event_manager_key_watcher_regressions.py +++ b/tests/test_input_event_manager_key_watcher_regressions.py @@ -118,6 +118,9 @@ class InputEventManagerKeyWatcherTests(unittest.TestCase): device = FakeDevice() fakeAtspi = self._fake_atspi((2, 60, 5), device) fakeAtspi.EventType = types.SimpleNamespace(KEY_PRESSED_EVENT=0, KEY_RELEASED_EVENT=1) + fakeAtspi.DeviceListener = types.SimpleNamespace(new=mock.Mock(return_value=object())) + fakeAtspi.register_keystroke_listener = mock.Mock(return_value=True) + fakeAtspi.deregister_keystroke_listener = mock.Mock(return_value=True) fakeKeySet = [types.SimpleNamespace(keycode=38), types.SimpleNamespace(keycode=56)] with ( @@ -128,8 +131,6 @@ class InputEventManagerKeyWatcherTests(unittest.TestCase): "_get_x11_consumable_key_set", return_value=fakeKeySet, ), - mock.patch.object(input_event_manager.pyatspi.Registry, "registerKeystrokeListener") as register, - mock.patch.object(input_event_manager.pyatspi.Registry, "deregisterKeystrokeListener") as deregister, mock.patch.object(manager, "_initialize_xterm_handoff") as initializeXtermHandoff, ): manager.start_key_watcher() @@ -148,19 +149,18 @@ class InputEventManagerKeyWatcherTests(unittest.TestCase): shiftLockMask | numLockMask, shiftMask | shiftLockMask | numLockMask, ] - register.assert_called_once_with( + listener = fakeAtspi.DeviceListener.new.return_value + fakeAtspi.DeviceListener.new.assert_called_once_with( manager._process_selective_legacy_keyboard_event, - key_set=fakeKeySet, - mask=expectedMasks, - kind=(0, 1), - synchronous=True, - preemptive=True, + None, ) - deregister.assert_called_once_with( - manager._process_selective_legacy_keyboard_event, - key_set=fakeKeySet, - mask=expectedMasks, - kind=(0, 1), + self.assertEqual( + fakeAtspi.register_keystroke_listener.call_args_list, + [mock.call(listener, fakeKeySet, mask, 3, 3) for mask in expectedMasks], + ) + self.assertEqual( + fakeAtspi.deregister_keystroke_listener.call_args_list, + [mock.call(listener, fakeKeySet, mask, 3) for mask in expectedMasks], ) initializeXtermHandoff.assert_called_once_with() self.assertEqual( @@ -171,6 +171,83 @@ class InputEventManagerKeyWatcherTests(unittest.TestCase): ], ) + def test_x11_consumable_listener_rolls_back_partial_registration(self) -> None: + manager = input_event_manager.InputEventManager() + device = FakeDevice() + fakeAtspi = self._fake_atspi((2, 60, 5), device) + fakeAtspi.EventType = types.SimpleNamespace(KEY_PRESSED_EVENT=0, KEY_RELEASED_EVENT=1) + fakeAtspi.DeviceListener = types.SimpleNamespace(new=mock.Mock(return_value=object())) + fakeAtspi.register_keystroke_listener = mock.Mock( + side_effect=[True, True, False], + ) + fakeAtspi.deregister_keystroke_listener = mock.Mock(return_value=True) + fakeKeySet = [types.SimpleNamespace(keycode=56)] + + with ( + mock.patch.object(input_event_manager, "Atspi", fakeAtspi), + mock.patch.dict(input_event_manager.os.environ, {"XDG_SESSION_TYPE": "x11"}), + mock.patch.object( + input_event_manager, + "_get_x11_consumable_key_set", + return_value=fakeKeySet, + ), + mock.patch.object(manager, "_initialize_xterm_handoff"), + ): + manager.start_key_watcher() + + expectedMasks = list(input_event_manager._X11_CONSUMABLE_MODIFIER_MASKS) + listener = fakeAtspi.DeviceListener.new.return_value + self.assertEqual( + fakeAtspi.register_keystroke_listener.call_args_list, + [ + mock.call(listener, fakeKeySet, expectedMasks[0], 3, 3), + mock.call(listener, fakeKeySet, expectedMasks[1], 3, 3), + mock.call(listener, fakeKeySet, expectedMasks[2], 3, 3), + ], + ) + self.assertEqual( + fakeAtspi.deregister_keystroke_listener.call_args_list, + [ + mock.call(listener, fakeKeySet, expectedMasks[0], 3), + mock.call(listener, fakeKeySet, expectedMasks[1], 3), + ], + ) + self.assertFalse(manager._selective_legacy_listener_active) + self.assertEqual(manager._selective_legacy_key_set, []) + self.assertEqual(manager._selective_legacy_keycodes, set()) + + def test_x11_consumable_listener_preserves_failed_deregistration_for_retry(self) -> None: + manager = input_event_manager.InputEventManager() + listener = object() + keySet = [types.SimpleNamespace(keycode=56)] + manager._selective_legacy_listener_active = True + manager._selective_legacy_listener = listener + manager._selective_legacy_registered_masks = [0] + manager._selective_legacy_key_set = keySet + manager._selective_legacy_keycodes = {56} + + with mock.patch.object( + input_event_manager.Atspi, + "deregister_keystroke_listener", + side_effect=[False, True], + ) as deregister: + manager._deregister_selective_legacy_key_listener() + + self.assertTrue(manager._selective_legacy_listener_active) + self.assertIs(manager._selective_legacy_listener, listener) + self.assertEqual(manager._selective_legacy_registered_masks, [0]) + self.assertEqual(manager._selective_legacy_key_set, keySet) + self.assertEqual(manager._selective_legacy_keycodes, {56}) + + manager._deregister_selective_legacy_key_listener() + + self.assertEqual(deregister.call_count, 2) + self.assertFalse(manager._selective_legacy_listener_active) + self.assertIsNone(manager._selective_legacy_listener) + self.assertEqual(manager._selective_legacy_registered_masks, []) + self.assertEqual(manager._selective_legacy_key_set, []) + self.assertEqual(manager._selective_legacy_keycodes, set()) + def test_x11_consumable_keycodes_include_structural_navigation_but_not_tab(self) -> None: keycodes = { keysym: index + 20