Storm Dragon
333 lines
14 KiB
Python
333 lines
14 KiB
Python
import sys
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
|
|
|
from cthulhu import input_event_manager
|
|
|
|
|
|
class FakeDevice:
|
|
def __init__(self) -> None:
|
|
self.add_key_watcher_calls = []
|
|
self.connect_calls = []
|
|
self.disconnect_calls = []
|
|
self.next_handler_id = 17
|
|
self.add_key_grab_calls = []
|
|
self.grab_ids = []
|
|
|
|
def add_key_watcher(self, callback: object, user_data: object = None) -> None:
|
|
self.add_key_watcher_calls.append((callback, user_data))
|
|
|
|
def connect(self, signalName: str, callback: object) -> int:
|
|
self.connect_calls.append((signalName, callback))
|
|
handlerId = self.next_handler_id
|
|
self.next_handler_id += 1
|
|
return handlerId
|
|
|
|
def disconnect(self, handlerId: int) -> None:
|
|
self.disconnect_calls.append(handlerId)
|
|
|
|
def add_key_grab(self, definition: object, callback: object = None) -> int:
|
|
self.add_key_grab_calls.append((definition, callback))
|
|
return self.grab_ids.pop(0)
|
|
|
|
|
|
class FakeDeviceFactory:
|
|
def __init__(self, device: FakeDevice) -> None:
|
|
self.device = device
|
|
|
|
def new(self) -> FakeDevice:
|
|
return self.device
|
|
|
|
def new_full(self, _appId: str) -> FakeDevice:
|
|
return self.device
|
|
|
|
|
|
class InputEventManagerKeyWatcherTests(unittest.TestCase):
|
|
@staticmethod
|
|
def _fake_atspi(
|
|
version: tuple[int, int, int],
|
|
device: FakeDevice,
|
|
) -> types.SimpleNamespace:
|
|
return types.SimpleNamespace(
|
|
get_version=lambda: version,
|
|
Device=FakeDeviceFactory(device),
|
|
)
|
|
|
|
def test_atspi_260_uses_key_signals_and_disconnects_them(self) -> None:
|
|
manager = input_event_manager.InputEventManager()
|
|
device = FakeDevice()
|
|
fakeAtspi = self._fake_atspi((2, 60, 5), device)
|
|
|
|
with (
|
|
mock.patch.object(input_event_manager, "Atspi", fakeAtspi),
|
|
mock.patch.dict(input_event_manager.os.environ, {"XDG_SESSION_TYPE": "wayland"}),
|
|
):
|
|
manager.start_key_watcher()
|
|
manager.stop_key_watcher()
|
|
|
|
self.assertEqual(
|
|
device.connect_calls,
|
|
[
|
|
("key-pressed", manager._on_key_pressed),
|
|
("key-released", manager._on_key_released),
|
|
],
|
|
)
|
|
self.assertEqual(device.add_key_watcher_calls, [])
|
|
self.assertEqual(device.disconnect_calls, [17, 18])
|
|
self.assertIsNone(manager.get_device())
|
|
|
|
def test_pre_atspi_260_keeps_legacy_key_watcher(self) -> None:
|
|
manager = input_event_manager.InputEventManager()
|
|
device = FakeDevice()
|
|
fakeAtspi = self._fake_atspi((2, 58, 4), device)
|
|
|
|
with (
|
|
mock.patch.object(input_event_manager, "Atspi", fakeAtspi),
|
|
mock.patch.dict(input_event_manager.os.environ, {"XDG_SESSION_TYPE": "wayland"}),
|
|
):
|
|
manager.start_key_watcher()
|
|
|
|
self.assertEqual(
|
|
device.add_key_watcher_calls,
|
|
[(manager.process_keyboard_event, None)],
|
|
)
|
|
self.assertEqual(device.connect_calls, [])
|
|
|
|
def test_key_signal_callbacks_preserve_pressed_state(self) -> None:
|
|
manager = input_event_manager.InputEventManager()
|
|
device = mock.Mock()
|
|
|
|
with mock.patch.object(manager, "process_keyboard_event") as processEvent:
|
|
manager._on_key_pressed(device, 56, 98, 0, "b")
|
|
manager._on_key_released(device, 56, 98, 0, "b")
|
|
|
|
self.assertEqual(
|
|
processEvent.call_args_list,
|
|
[
|
|
mock.call(device, True, 56, 98, 0, "b"),
|
|
mock.call(device, False, 56, 98, 0, "b"),
|
|
],
|
|
)
|
|
|
|
def test_x11_consumable_listener_includes_locking_modifier_combinations(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)
|
|
fakeKeySet = [types.SimpleNamespace(keycode=38), 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(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()
|
|
manager.stop_key_watcher()
|
|
|
|
shiftMask = 1 << int(input_event_manager.Atspi.ModifierType.SHIFT)
|
|
shiftLockMask = 1 << int(input_event_manager.Atspi.ModifierType.SHIFTLOCK)
|
|
numLockMask = 1 << int(input_event_manager.Atspi.ModifierType.NUMLOCK)
|
|
expectedMasks = [
|
|
0,
|
|
shiftMask,
|
|
shiftLockMask,
|
|
shiftMask | shiftLockMask,
|
|
numLockMask,
|
|
shiftMask | numLockMask,
|
|
shiftLockMask | numLockMask,
|
|
shiftMask | shiftLockMask | numLockMask,
|
|
]
|
|
register.assert_called_once_with(
|
|
manager._process_selective_legacy_keyboard_event,
|
|
key_set=fakeKeySet,
|
|
mask=expectedMasks,
|
|
kind=(0, 1),
|
|
synchronous=True,
|
|
preemptive=True,
|
|
)
|
|
deregister.assert_called_once_with(
|
|
manager._process_selective_legacy_keyboard_event,
|
|
key_set=fakeKeySet,
|
|
mask=expectedMasks,
|
|
kind=(0, 1),
|
|
)
|
|
initializeXtermHandoff.assert_called_once_with()
|
|
self.assertEqual(
|
|
device.connect_calls,
|
|
[
|
|
("key-pressed", manager._on_key_pressed),
|
|
("key-released", manager._on_key_released),
|
|
],
|
|
)
|
|
|
|
def test_x11_consumable_keycodes_include_structural_navigation_but_not_tab(self) -> None:
|
|
keycodes = {
|
|
keysym: index + 20
|
|
for index, keysym in enumerate(input_event_manager._X11_CONSUMABLE_KEYSYMS)
|
|
}
|
|
|
|
with mock.patch("cthulhu.keybindings.get_keycodes") as getKeycodes:
|
|
getKeycodes.side_effect = lambda keysym: (0, keycodes[keysym])
|
|
result = input_event_manager._get_x11_consumable_key_set()
|
|
|
|
resultKeycodes = {definition.keycode for definition in result}
|
|
self.assertIn(keycodes["b"], resultKeycodes)
|
|
self.assertIn(keycodes["h"], resultKeycodes)
|
|
requestedKeysyms = {call.args[0] for call in getKeycodes.call_args_list}
|
|
self.assertNotIn("Tab", requestedKeysyms)
|
|
|
|
def test_x11_consumable_key_set_uses_atspi_key_definitions(self) -> None:
|
|
with mock.patch("cthulhu.keybindings.get_keycodes", return_value=(0, 38)):
|
|
keySet = input_event_manager._get_x11_consumable_key_set()
|
|
|
|
self.assertTrue(keySet)
|
|
self.assertTrue(all(isinstance(item, input_event_manager.Atspi.KeyDefinition) for item in keySet))
|
|
|
|
def test_modern_signal_defers_registered_x11_key_to_consumable_listener(self) -> None:
|
|
manager = input_event_manager.InputEventManager()
|
|
manager._selective_legacy_listener_active = True
|
|
manager._selective_legacy_keycodes = {56}
|
|
device = mock.Mock()
|
|
|
|
with mock.patch.object(manager, "process_keyboard_event") as processEvent:
|
|
manager._on_key_pressed(device, 56, 98, 0, "b")
|
|
manager._on_key_released(device, 56, 98, 0, "b")
|
|
|
|
processEvent.assert_not_called()
|
|
|
|
def test_modern_signal_defers_x11_key_with_locking_modifiers(self) -> None:
|
|
manager = input_event_manager.InputEventManager()
|
|
manager._selective_legacy_listener_active = True
|
|
manager._selective_legacy_keycodes = {56}
|
|
device = mock.Mock()
|
|
shiftMask = 1 << int(input_event_manager.Atspi.ModifierType.SHIFT)
|
|
shiftLockMask = 1 << int(input_event_manager.Atspi.ModifierType.SHIFTLOCK)
|
|
numLockMask = 1 << int(input_event_manager.Atspi.ModifierType.NUMLOCK)
|
|
lockingMasks = (
|
|
shiftLockMask,
|
|
shiftMask | shiftLockMask,
|
|
numLockMask,
|
|
shiftMask | numLockMask,
|
|
shiftLockMask | numLockMask,
|
|
shiftMask | shiftLockMask | numLockMask,
|
|
)
|
|
|
|
with mock.patch.object(manager, "process_keyboard_event") as processEvent:
|
|
for modifiers in lockingMasks:
|
|
manager._on_key_pressed(device, 56, 98, modifiers, "b")
|
|
manager._on_key_released(device, 56, 98, modifiers, "b")
|
|
|
|
processEvent.assert_not_called()
|
|
|
|
def test_modern_signal_still_processes_keys_not_owned_by_selective_listener(self) -> None:
|
|
manager = input_event_manager.InputEventManager()
|
|
manager._selective_legacy_listener_active = True
|
|
manager._selective_legacy_keycodes = {56}
|
|
device = mock.Mock()
|
|
|
|
with mock.patch.object(manager, "process_keyboard_event") as processEvent:
|
|
manager._on_key_pressed(device, 23, 65289, 0, "Tab")
|
|
manager._on_key_released(device, 23, 65289, 0, "Tab")
|
|
|
|
self.assertEqual(
|
|
processEvent.call_args_list,
|
|
[
|
|
mock.call(device, True, 23, 65289, 0, "Tab"),
|
|
mock.call(device, False, 23, 65289, 0, "Tab"),
|
|
],
|
|
)
|
|
|
|
def test_keyboard_event_returns_actual_consumption_decision(self) -> None:
|
|
manager = input_event_manager.InputEventManager()
|
|
focusManager = mock.Mock()
|
|
focusManager.get_active_window.return_value = object()
|
|
focusManager.get_locus_of_focus.return_value = object()
|
|
scriptManager = mock.Mock()
|
|
scriptManager.get_active_script.return_value = mock.Mock()
|
|
keyboardEvent = mock.Mock()
|
|
keyboardEvent.is_modifier_key.return_value = False
|
|
keyboardEvent.process.return_value = False
|
|
|
|
with (
|
|
mock.patch.object(input_event_manager.cthulhu_state, "capturingKeys", False),
|
|
mock.patch.object(input_event_manager.cthulhu_state, "pendingSelfHostedFocus", None),
|
|
mock.patch.object(input_event_manager.focus_manager, "get_manager", return_value=focusManager),
|
|
mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager),
|
|
mock.patch.object(input_event_manager.input_event, "KeyboardEvent", return_value=keyboardEvent),
|
|
mock.patch.object(manager, "_should_pass_through_for_active_xterm", return_value=False),
|
|
mock.patch.object(input_event_manager.AXUtilities, "can_be_active_window", return_value=True),
|
|
mock.patch.object(manager, "last_event_was_keyboard", return_value=False),
|
|
):
|
|
result = manager.process_keyboard_event(mock.Mock(), True, 56, 98, 0, "b")
|
|
|
|
self.assertFalse(result)
|
|
keyboardEvent.process.assert_called_once_with()
|
|
|
|
def test_x11_consumable_listener_is_only_owner_of_matching_key_grabs(self) -> None:
|
|
manager = input_event_manager.InputEventManager()
|
|
device = FakeDevice()
|
|
device.grab_ids = [101, 102, 103]
|
|
manager._device = device
|
|
manager._selective_legacy_listener_active = True
|
|
manager._selective_legacy_keycodes = {56}
|
|
legacyOwned = types.SimpleNamespace(keycode=0, keysym=98, modifiers=0)
|
|
modified = types.SimpleNamespace(keycode=0, keysym=98, modifiers=4)
|
|
unrelated = types.SimpleNamespace(keycode=0, keysym=106, modifiers=0)
|
|
binding = mock.Mock()
|
|
binding.is_enabled.return_value = True
|
|
binding.is_bound.return_value = True
|
|
binding.has_grabs.return_value = False
|
|
binding.keycode = 56
|
|
binding.key_definitions.return_value = [legacyOwned, modified]
|
|
|
|
unrelatedBinding = mock.Mock()
|
|
unrelatedBinding.is_enabled.return_value = True
|
|
unrelatedBinding.is_bound.return_value = True
|
|
unrelatedBinding.has_grabs.return_value = False
|
|
unrelatedBinding.keycode = 44
|
|
unrelatedBinding.key_definitions.return_value = [unrelated]
|
|
|
|
result = manager.add_grabs_for_keybinding(binding)
|
|
unrelatedResult = manager.add_grabs_for_keybinding(unrelatedBinding)
|
|
|
|
self.assertEqual(result, [101])
|
|
self.assertEqual(unrelatedResult, [102])
|
|
self.assertEqual(
|
|
device.add_key_grab_calls,
|
|
[(modified, None), (unrelated, None)],
|
|
)
|
|
|
|
def test_keybinding_grab_failure_rolls_back_partial_additions(self) -> None:
|
|
manager = input_event_manager.InputEventManager()
|
|
device = mock.Mock()
|
|
device.add_key_grab.side_effect = [101, RuntimeError("grab failed")]
|
|
manager._device = device
|
|
firstDefinition = types.SimpleNamespace(keycode=44, modifiers=0)
|
|
secondDefinition = types.SimpleNamespace(keycode=45, modifiers=0)
|
|
binding = mock.Mock()
|
|
binding.is_enabled.return_value = True
|
|
binding.is_bound.return_value = True
|
|
binding.has_grabs.return_value = False
|
|
binding.keycode = 44
|
|
binding.key_definitions.return_value = [firstDefinition, secondDefinition]
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "grab failed"):
|
|
manager.add_grabs_for_keybinding(binding)
|
|
|
|
device.remove_key_grab.assert_called_once_with(101)
|
|
self.assertEqual(manager._grabbed_bindings, {})
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|