Remove python-atspi dependency
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user