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()