diff --git a/src/cthulhu/input_event_manager.py b/src/cthulhu/input_event_manager.py index 15d5874..f4c3e35 100644 --- a/src/cthulhu/input_event_manager.py +++ b/src/cthulhu/input_event_manager.py @@ -46,6 +46,7 @@ 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 @@ -61,6 +62,44 @@ from .ax_utilities_application import AXUtilitiesApplication if TYPE_CHECKING: from . import keybindings +_X11_SHIFT_MASK = 1 << int(Atspi.ModifierType.SHIFT) +_X11_SHIFT_LOCK_MASK = 1 << int(Atspi.ModifierType.SHIFTLOCK) +_X11_NUM_LOCK_MASK = 1 << int(Atspi.ModifierType.NUMLOCK) +_X11_CONSUMABLE_MODIFIER_MASKS = ( + 0, + _X11_SHIFT_MASK, + _X11_SHIFT_LOCK_MASK, + _X11_SHIFT_MASK | _X11_SHIFT_LOCK_MASK, + _X11_NUM_LOCK_MASK, + _X11_SHIFT_MASK | _X11_NUM_LOCK_MASK, + _X11_SHIFT_LOCK_MASK | _X11_NUM_LOCK_MASK, + _X11_SHIFT_MASK | _X11_SHIFT_LOCK_MASK | _X11_NUM_LOCK_MASK, +) +_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", + "1", "2", "3", "4", "5", "6", "comma", +) +def _get_x11_consumable_key_set() -> List[Atspi.KeyDefinition]: + """Returns AT-SPI definitions for unmodified structural-navigation keys.""" + + # Importing here avoids a module-level cycle: keybindings obtains this + # manager lazily when bindings install their AT-SPI grabs. + from . import keybindings # pylint: disable=import-outside-toplevel + + keycodes = sorted({ + keycode + for keysym in _X11_CONSUMABLE_KEYSYMS + if (keycode := keybindings.get_keycodes(keysym)[1]) + }) + keySet = [] + for keycode in keycodes: + definition = Atspi.KeyDefinition() + definition.keycode = keycode + keySet.append(definition) + return keySet + + class InputEventManager: """Provides utilities for managing input events.""" @@ -68,6 +107,11 @@ class InputEventManager: self._last_input_event: Optional[input_event.InputEvent] = None self._last_non_modifier_key_event: Optional[input_event.KeyboardEvent] = None self._device: Optional[Atspi.Device] = None + self._key_pressed_id: int = 0 + self._key_released_id: int = 0 + self._selective_legacy_listener_active: bool = False + self._selective_legacy_key_set: List[Atspi.KeyDefinition] = [] + self._selective_legacy_keycodes: set[int] = set() self._pointer_moved_id: int = 0 self._mapped_keycodes: List[int] = [] self._mapped_keysyms: List[int] = [] @@ -101,13 +145,56 @@ class InputEventManager: msg = "INPUT EVENT MANAGER: Starting key watcher." debug.print_message(debug.LEVEL_INFO, msg, True) - self.activate_device().add_key_watcher(self.process_keyboard_event) + device = self.activate_device() + atspiVersion = Atspi.get_version() + if atspiVersion[0] > 2 or atspiVersion[1] >= 60: + msg = "INPUT EVENT MANAGER: Using AT-SPI key signals." + debug.print_message(debug.LEVEL_INFO, msg, True) + self._key_pressed_id = device.connect("key-pressed", self._on_key_pressed) + self._key_released_id = device.connect("key-released", self._on_key_released) + if os.environ.get("XDG_SESSION_TYPE", "").lower() == "x11": + self._selective_legacy_key_set = _get_x11_consumable_key_set() + if self._selective_legacy_key_set: + self._selective_legacy_keycodes = { + definition.keycode for definition in self._selective_legacy_key_set + } + 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 + else: + msg = "INPUT EVENT MANAGER: Using legacy AT-SPI key watcher." + debug.print_message(debug.LEVEL_INFO, msg, True) + device.add_key_watcher(self.process_keyboard_event) def stop_key_watcher(self) -> None: - """Starts the watcher for keyboard input events.""" + """Stops the watcher for keyboard input events.""" 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() + if self._device is not None: + for handlerId in (self._key_pressed_id, self._key_released_id): + if handlerId: + self._device.disconnect(handlerId) + self._key_pressed_id = 0 + self._key_released_id = 0 self._device = None def get_debug_snapshot(self) -> Dict[str, object]: @@ -229,6 +316,11 @@ class InputEventManager: grab_ids = [] for kd in binding.key_definitions(): + definitionKeycode = kd.keycode or binding.keycode + if self._signal_event_uses_selective_legacy_listener( + definitionKeycode, kd.modifiers + ): + continue grab_id = self._device.add_key_grab(kd, None) if grab_id == 0: continue @@ -756,6 +848,62 @@ class InputEventManager: # pylint: disable=too-many-arguments # pylint: disable=too-many-positional-arguments + def _process_selective_legacy_keyboard_event(self, event: Atspi.DeviceEvent) -> bool: + """Returns an X11 consume decision for unmodified browse-navigation keys.""" + + pressed = event.type == Atspi.EventType.KEY_PRESSED_EVENT + return self.process_keyboard_event( + self._device, + pressed, + event.hw_code, + event.id, + event.modifiers, + event.event_string or "", + ) + + def _signal_event_uses_selective_legacy_listener(self, keycode: int, modifiers: int) -> bool: + """Returns True if the X11 compatibility listener owns this event.""" + + return ( + self._selective_legacy_listener_active + and keycode in self._selective_legacy_keycodes + and modifiers in _X11_CONSUMABLE_MODIFIER_MASKS + ) + + def _on_key_pressed( + self, + device: Atspi.Device, + keycode: int, + keysym: int, + modifiers: int, + text: str, + ) -> None: + """Processes a key-pressed signal from AT-SPI 2.60 and later.""" + + if self._signal_event_uses_selective_legacy_listener(keycode, modifiers): + msg = "INPUT EVENT MANAGER: Deferring key-pressed signal to X11 consumable listener." + debug.print_message(debug.LEVEL_INFO, msg, True) + return + + self.process_keyboard_event(device, True, keycode, keysym, modifiers, text) + + def _on_key_released( + self, + device: Atspi.Device, + keycode: int, + keysym: int, + modifiers: int, + text: str, + ) -> None: + """Processes a key-released signal from AT-SPI 2.60 and later.""" + + if self._signal_event_uses_selective_legacy_listener(keycode, modifiers): + msg = "INPUT EVENT MANAGER: Deferring key-released signal to X11 consumable listener." + debug.print_message(debug.LEVEL_INFO, msg, True) + return + + self.process_keyboard_event(device, False, keycode, keysym, modifiers, text) + def process_keyboard_event( self, _device: Atspi.Device, @@ -849,7 +997,7 @@ class InputEventManager: event._finalize_initialization() event.set_click_count(self._determine_keyboard_event_click_count(event)) - event.process() + consumed = event.process() if event.is_modifier_key(): if self.is_release_for(event, self._last_input_event): @@ -859,7 +1007,7 @@ class InputEventManager: else: self._last_non_modifier_key_event = event self._last_input_event = event - return True + return consumed # pylint: enable=too-many-arguments # pylint: enable=too-many-positional-arguments diff --git a/tests/test_cthulhu_remote_plugin.py b/tests/test_cthulhu_remote_plugin.py index 6e01677..17ffab2 100644 --- a/tests/test_cthulhu_remote_plugin.py +++ b/tests/test_cthulhu_remote_plugin.py @@ -6,18 +6,62 @@ from unittest import mock sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) -input_event_manager_stub = types.ModuleType("cthulhu.input_event_manager") -input_event_manager_stub.get_manager = mock.Mock(return_value=mock.Mock()) -sys.modules["cthulhu.input_event_manager"] = input_event_manager_stub +import cthulhu as cthulhuPackage -from cthulhu.plugin_system_manager import PluginSystemManager -from cthulhu import cthulhu_state -from cthulhu import speech -from cthulhu.plugins.CthulhuRemote.connection_info import ConnectionInfo, ConnectionMode -from cthulhu.plugins.CthulhuRemote.local_machine import LocalMachine -from cthulhu.plugins.CthulhuRemote.plugin import CthulhuRemote -from cthulhu.plugins.CthulhuRemote.protocol import RemoteMessageType -from cthulhu.plugins.CthulhuRemote.serializer import JSONSerializer +missingModule = object() +previousInputEventManagerModule = sys.modules.get( + "cthulhu.input_event_manager", + missingModule, +) +previousInputEventManagerAttribute = getattr( + cthulhuPackage, + "input_event_manager", + missingModule, +) +previousPluginSystemManagerModule = sys.modules.get( + "cthulhu.plugin_system_manager", + missingModule, +) +previousPluginSystemManagerAttribute = getattr( + cthulhuPackage, + "plugin_system_manager", + missingModule, +) +inputEventManagerStub = types.ModuleType("cthulhu.input_event_manager") +inputEventManagerStub.get_manager = mock.Mock(return_value=mock.Mock()) +sys.modules["cthulhu.input_event_manager"] = inputEventManagerStub +cthulhuPackage.input_event_manager = inputEventManagerStub + +try: + from cthulhu import plugin_system_manager + from cthulhu import cthulhu_state + from cthulhu import speech + from cthulhu.plugin_system_manager import PluginSystemManager + from cthulhu.plugins.CthulhuRemote.connection_info import ConnectionInfo, ConnectionMode + from cthulhu.plugins.CthulhuRemote.local_machine import LocalMachine + from cthulhu.plugins.CthulhuRemote.plugin import CthulhuRemote + from cthulhu.plugins.CthulhuRemote.protocol import RemoteMessageType + from cthulhu.plugins.CthulhuRemote.serializer import JSONSerializer +finally: + if previousInputEventManagerModule is missingModule: + sys.modules.pop("cthulhu.input_event_manager", None) + else: + sys.modules["cthulhu.input_event_manager"] = previousInputEventManagerModule + + if previousInputEventManagerAttribute is missingModule: + delattr(cthulhuPackage, "input_event_manager") + else: + cthulhuPackage.input_event_manager = previousInputEventManagerAttribute + + if previousPluginSystemManagerModule is missingModule: + sys.modules.pop("cthulhu.plugin_system_manager", None) + else: + sys.modules["cthulhu.plugin_system_manager"] = previousPluginSystemManagerModule + + if previousPluginSystemManagerAttribute is missingModule: + delattr(cthulhuPackage, "plugin_system_manager") + else: + cthulhuPackage.plugin_system_manager = previousPluginSystemManagerAttribute class CthulhuRemotePluginTests(unittest.TestCase): diff --git a/tests/test_input_event_manager_key_watcher_regressions.py b/tests/test_input_event_manager_key_watcher_regressions.py new file mode 100644 index 0000000..5233464 --- /dev/null +++ b/tests/test_input_event_manager_key_watcher_regressions.py @@ -0,0 +1,310 @@ +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, + ): + 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), + ) + 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)], + ) + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_plugin_system_manager_regressions.py b/tests/test_plugin_system_manager_regressions.py index fa65cd0..b415670 100644 --- a/tests/test_plugin_system_manager_regressions.py +++ b/tests/test_plugin_system_manager_regressions.py @@ -8,11 +8,55 @@ from unittest import mock sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) -input_event_manager_stub = types.ModuleType("cthulhu.input_event_manager") -input_event_manager_stub.get_manager = mock.Mock(return_value=mock.Mock()) -sys.modules["cthulhu.input_event_manager"] = input_event_manager_stub +import cthulhu as cthulhuPackage -from cthulhu.plugin_system_manager import PluginInfo, PluginSystemManager +missingModule = object() +previousInputEventManagerModule = sys.modules.get( + "cthulhu.input_event_manager", + missingModule, +) +previousInputEventManagerAttribute = getattr( + cthulhuPackage, + "input_event_manager", + missingModule, +) +previousPluginSystemManagerModule = sys.modules.get( + "cthulhu.plugin_system_manager", + missingModule, +) +previousPluginSystemManagerAttribute = getattr( + cthulhuPackage, + "plugin_system_manager", + missingModule, +) +inputEventManagerStub = types.ModuleType("cthulhu.input_event_manager") +inputEventManagerStub.get_manager = mock.Mock(return_value=mock.Mock()) +sys.modules["cthulhu.input_event_manager"] = inputEventManagerStub +cthulhuPackage.input_event_manager = inputEventManagerStub + +try: + from cthulhu import plugin_system_manager + from cthulhu.plugin_system_manager import PluginInfo, PluginSystemManager +finally: + if previousInputEventManagerModule is missingModule: + sys.modules.pop("cthulhu.input_event_manager", None) + else: + sys.modules["cthulhu.input_event_manager"] = previousInputEventManagerModule + + if previousInputEventManagerAttribute is missingModule: + delattr(cthulhuPackage, "input_event_manager") + else: + cthulhuPackage.input_event_manager = previousInputEventManagerAttribute + + if previousPluginSystemManagerModule is missingModule: + sys.modules.pop("cthulhu.plugin_system_manager", None) + else: + sys.modules["cthulhu.plugin_system_manager"] = previousPluginSystemManagerModule + + if previousPluginSystemManagerAttribute is missingModule: + delattr(cthulhuPackage, "plugin_system_manager") + else: + cthulhuPackage.plugin_system_manager = previousPluginSystemManagerAttribute PLUGIN_TEMPLATE = """ diff --git a/tests/test_steam_notification_regressions.py b/tests/test_steam_notification_regressions.py index b732aeb..8efc244 100644 --- a/tests/test_steam_notification_regressions.py +++ b/tests/test_steam_notification_regressions.py @@ -8,6 +8,9 @@ import gi gi.require_version("Gdk", "3.0") gi.require_version("Gtk", "3.0") +gi.require_version("Atspi", "2.0") + +from gi.repository import Atspi sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) @@ -20,9 +23,62 @@ if soundGeneratorModule is not None and not hasattr(soundGeneratorModule, "Sound from cthulhu.ax_object import AXObject from cthulhu.ax_utilities import AXUtilities +from cthulhu import cthulhu_state from cthulhu.scripts.apps.steamwebhelper import script as steam_script +class SteamInactiveNotificationContractTests(unittest.TestCase): + @staticmethod + def _prepare_event_script(testScript, presentIfInactive): + handler = mock.Mock() + testScript.presentIfInactive = presentIfInactive + testScript.listeners = {"object:state-changed:showing": handler} + testScript.generatorCache = {"stale": object()} + testScript.skipObjectEvent = mock.Mock(return_value=False) + return handler + + def test_steam_notification_is_processed_while_script_is_inactive(self): + app = object() + + def initialize_chromium(testScript, initializedApp): + testScript.app = initializedApp + testScript.presentIfInactive = False + + with mock.patch.object( + steam_script.Chromium.Script, + "__init__", + new=initialize_chromium, + ): + testScript = steam_script.Script(app) + + handler = self._prepare_event_script(testScript, testScript.presentIfInactive) + event = mock.Mock(type="object:state-changed:showing", source=object()) + + with ( + mock.patch.object(cthulhu_state, "activeScript", object()), + mock.patch.object(AXObject, "get_role", return_value=Atspi.Role.NOTIFICATION), + ): + testScript.processObjectEvent(event) + + self.assertTrue(testScript.presentIfInactive) + handler.assert_called_once_with(event) + + def test_generic_chromium_notification_is_deferred_while_script_is_inactive(self): + testScript = steam_script.Chromium.Script.__new__( + steam_script.Chromium.Script + ) + handler = self._prepare_event_script(testScript, presentIfInactive=False) + event = mock.Mock(type="object:state-changed:showing", source=object()) + + with ( + mock.patch.object(cthulhu_state, "activeScript", object()), + mock.patch.object(AXObject, "get_role", return_value=Atspi.Role.NOTIFICATION), + ): + testScript.processObjectEvent(event) + + handler.assert_not_called() + + class SteamNotificationRootTests(unittest.TestCase): def test_prefers_notification_ancestor_over_live_region_descendant(self): testScript = steam_script.Script.__new__(steam_script.Script) @@ -50,11 +106,55 @@ class SteamNotificationRootTests(unittest.TestCase): self.assertIs(result, notification) -class SteamNotificationQueueTests(unittest.TestCase): - def test_merges_non_overlapping_fragments_for_same_notification(self): +class SteamNotificationRoutingContractTests(unittest.TestCase): + def test_steam_notification_override_handles_showing_before_chromium(self): testScript = steam_script.Script.__new__(steam_script.Script) notification = object() + event = mock.Mock(source=notification, detail1=1) + testScript._presentSteamNotification = mock.Mock() + with ( + mock.patch.object( + testScript, + "_isSteamNotification", + return_value=True, + ), + mock.patch.object( + steam_script.Chromium.Script, + "onShowingChanged", + ) as chromiumHandler, + ): + testScript.onShowingChanged(event) + + testScript._presentSteamNotification.assert_called_once_with(notification) + chromiumHandler.assert_not_called() + + def test_nonsteam_showing_event_defers_to_chromium(self): + testScript = steam_script.Script.__new__(steam_script.Script) + event = mock.Mock(source=object(), detail1=1) + testScript._presentSteamNotification = mock.Mock() + + with ( + mock.patch.object( + testScript, + "_isSteamNotification", + return_value=False, + ), + mock.patch.object( + steam_script.Chromium.Script, + "onShowingChanged", + ) as chromiumHandler, + ): + testScript.onShowingChanged(event) + + testScript._presentSteamNotification.assert_not_called() + chromiumHandler.assert_called_once_with(event) + + +class SteamNotificationQueueTests(unittest.TestCase): + @staticmethod + def _make_script(): + testScript = steam_script.Script.__new__(steam_script.Script) testScript._lastSteamNotification = ("", 0.0) testScript._steamPendingNotification = None testScript._steamRelativeTimePattern = re.compile( @@ -63,6 +163,11 @@ class SteamNotificationQueueTests(unittest.TestCase): ) testScript._resetSteamPendingTimer = mock.Mock() testScript._presentSteamNotificationTextNow = mock.Mock() + return testScript + + def test_merges_non_overlapping_fragments_for_same_notification(self): + testScript = self._make_script() + notification = object() testScript._queueSteamNotification("username", notification) testScript._queueSteamNotification("Playing: Borderlands 2", notification) @@ -80,6 +185,39 @@ class SteamNotificationQueueTests(unittest.TestCase): notification, ) + def test_merges_relative_timestamp_without_repeated_presentation(self): + testScript = self._make_script() + notification = object() + + testScript._queueSteamNotification("Username is now online", notification) + testScript._queueSteamNotification("2 minutes ago", notification) + + testScript._presentSteamNotificationTextNow.assert_not_called() + self.assertEqual( + testScript._steamPendingNotification["text"], + "Username is now online. 2 minutes ago", + ) + + testScript._flushSteamPendingNotification(fromTimer=True) + + testScript._presentSteamNotificationTextNow.assert_called_once_with( + "Username is now online. 2 minutes ago", + notification, + ) + + def test_duplicate_complete_notification_is_presented_once(self): + testScript = self._make_script() + notification = object() + + testScript._queueSteamNotification("Username is now online", notification) + testScript._queueSteamNotification("Username is now online", notification) + testScript._flushSteamPendingNotification(fromTimer=True) + + testScript._presentSteamNotificationTextNow.assert_called_once_with( + "Username is now online", + notification, + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_steam_selection_regressions.py b/tests/test_steam_selection_regressions.py index ab30ae3..5414757 100644 --- a/tests/test_steam_selection_regressions.py +++ b/tests/test_steam_selection_regressions.py @@ -569,6 +569,38 @@ class SteamReturnActivationTests(unittest.TestCase): self.assertEqual(testScript._steamFriendsPanelActivationTarget, (None, "", 0.0)) + def test_return_rejects_friends_target_after_nonsteam_x11_handoff(self): + testScript = steam_script.Script.__new__(steam_script.Script) + button = object() + keyboardEvent = mock.Mock(event_string="Return", modifiers=0) + keyboardEvent.is_pressed_key.return_value = True + + testScript.utilities = mock.Mock() + testScript._steamActiveQuickAccessTab = "Friends" + testScript._steamFriendsPanelActivationTarget = ( + button, + "Username Playing Terraria", + 100.0, + ) + testScript._steamFriendsPanelActivationTargetMaxAge = 10.0 + + with ( + mock.patch.object(steam_script.time, "monotonic", return_value=105.0), + mock.patch.object( + steam_script.Script, + "_activeX11WindowIsSteamSelectionContext", + return_value=False, + ), + mock.patch.object(testScript, "_getClickableActivationTarget", return_value=None), + mock.patch.object( + steam_script.Script, + "_performClickableAction", + ) as performAction, + ): + self.assertFalse(testScript.shouldConsumeKeyboardEvent(keyboardEvent, None)) + + performAction.assert_not_called() + class SteamVirtualizedListMutationTests(unittest.TestCase): def test_name_changed_presents_friends_panel_item_after_friends_tab_activation(self): diff --git a/tests/test_web_event_sequences.py b/tests/test_web_event_sequences.py new file mode 100644 index 0000000..3b49b58 --- /dev/null +++ b/tests/test_web_event_sequences.py @@ -0,0 +1,1000 @@ +"""Ordered browser event-sequence regression tests.""" + +import unittest +from dataclasses import dataclass +from typing import Any +from unittest import mock + +import gi + +gi.require_version("Atspi", "2.0") +from gi.repository import Atspi + +from cthulhu import cthulhu_state +from cthulhu.ax_utilities_event import AXUtilitiesEvent +from cthulhu.scripts import default +from cthulhu.scripts.toolkits.Chromium import script as chromium_script +from cthulhu.scripts.toolkits.Gecko import script as gecko_script +from cthulhu.scripts.web import script as web_script + + +@dataclass(eq=False) +class FakeAccessible: + """Minimal accessible metadata needed to describe a sequence target.""" + + role: Atspi.Role + editable: bool = False + + +@dataclass(eq=False) +class FakeEvent: + """Minimal AT-SPI event used by ordered web regression fixtures.""" + + type: str + source: FakeAccessible + detail1: int = 0 + any_data: FakeAccessible | None = None + + +class WebEventSequence: + """Dispatch an ordered event list through Chromium's current routing path.""" + + def __init__(self, testScript: chromium_script.Script) -> None: + self.testScript = testScript + + def run(self, events: list[FakeEvent]) -> None: + handlers = { + "object:active-descendant-changed": chromium_script.Script.onActiveDescendantChanged, + "object:children-changed:add": chromium_script.Script.onChildrenAdded, + "object:children-changed:remove": chromium_script.Script.onChildrenRemoved, + "object:selection-changed": chromium_script.Script.onSelectionChanged, + "object:state-changed:busy": chromium_script.Script.onBusyChanged, + "object:text-caret-moved": chromium_script.Script.onCaretMoved, + "object:text-selection-changed": chromium_script.Script.onTextSelectionChanged, + "object:state-changed:focused": chromium_script.Script.onFocusedChanged, + "document:load-complete": chromium_script.Script.onDocumentLoadComplete, + "document:reload": chromium_script.Script.onDocumentReload, + } + for event in events: + handler = handlers[event.type] + handler(self.testScript, event) + + +class GeckoEventSequence: + """Dispatch an ordered event list through Gecko's current routing path.""" + + def __init__(self, testScript: gecko_script.Script) -> None: + self.testScript = testScript + + def run(self, events: list[FakeEvent]) -> None: + handlers = { + "object:text-caret-moved": gecko_script.Script.onCaretMoved, + "object:state-changed:focused": gecko_script.Script.onFocusedChanged, + } + for event in events: + handler = handlers[event.type] + handler(self.testScript, event) + + +class ChromiumTabFocusSequenceTests(unittest.TestCase): + """Behavior contracts for controls reached by Tab in Chromium.""" + + @staticmethod + def _make_script(document: object, oldFocus: object) -> chromium_script.Script: + testScript = chromium_script.Script.__new__(chromium_script.Script) + testScript._lastCommandWasCaretNav = False + testScript._lastCommandWasStructNav = False + testScript._lastCommandWasMouseButton = False + testScript._clearSyntheticWebSelection = mock.Mock() + testScript.utilities = mock.Mock() + testScript.utilities.isStaticTextLeaf.return_value = False + testScript.utilities.isRedundantAutocompleteEvent.return_value = False + testScript.utilities.isZombie.return_value = False + testScript.utilities.getTopLevelDocumentForObject.return_value = document + testScript.utilities.getCaretContext.return_value = (oldFocus, 0) + testScript.utilities.lastInputEventWasCaretNavWithSelection.return_value = False + testScript.utilities.lastInputEventWasCharNav.return_value = False + testScript.utilities.getDocumentForObject.return_value = document + testScript.utilities.isWebAppDescendant.return_value = False + testScript.utilities.handleEventFromContextReplicant.return_value = False + return testScript + + def _assert_tab_caret_focused_sequence_presents_once( + self, + target: FakeAccessible, + ) -> None: + document = object() + oldFocus = object() + testScript = self._make_script(document, oldFocus) + inputManager = mock.Mock() + inputManager.last_event_was_caret_selection.return_value = False + inputManager.last_event_was_caret_navigation.return_value = False + inputManager.last_event_was_select_all.return_value = False + inputManager.last_event_was_primary_click_or_release.return_value = False + inputManager.last_event_was_tab_navigation.return_value = True + presentations: list[Any] = [] + events = [ + FakeEvent("object:text-caret-moved", target, detail1=0), + FakeEvent("object:state-changed:focused", target, detail1=1), + ] + + with ( + mock.patch.object(cthulhu_state, "locusOfFocus", oldFocus), + mock.patch( + "cthulhu.input_event_manager.get_manager", + return_value=inputManager, + ), + mock.patch( + "cthulhu.ax_utilities_event.focus_manager.get_manager" + ) as getFocusManager, + mock.patch( + "cthulhu.ax_utilities_event.AXUtilitiesRole.is_text_input_search", + return_value=False, + ), + mock.patch( + "cthulhu.ax_utilities_event.AXUtilitiesState.is_editable", + side_effect=lambda obj: obj.editable, + ), + mock.patch.object( + web_script.AXUtilities, + "is_editable", + side_effect=lambda obj: obj.editable, + ), + mock.patch.object(web_script.AXUtilities, "is_dialog_or_alert", return_value=False), + mock.patch.object(web_script.AXUtilities, "is_focusable", return_value=True), + mock.patch.object(web_script.AXUtilities, "is_focused", return_value=True), + mock.patch.object(web_script.AXUtilities, "is_document", return_value=False), + mock.patch.object( + default.Script, + "onCaretMoved", + side_effect=lambda _script, event: presentations.append(event.source), + ) as defaultCaretHandler, + mock.patch.object( + default.Script, + "onFocusedChanged", + side_effect=lambda _script, event: presentations.append(event.source), + ) as defaultFocusHandler, + ): + getFocusManager.return_value.get_active_mode_and_object_of_interest.return_value = ( + None, + oldFocus, + ) + WebEventSequence(testScript).run(events) + + self.assertEqual(presentations, [target]) + defaultCaretHandler.assert_not_called() + defaultFocusHandler.assert_called_once_with(testScript, events[1]) + testScript._clearSyntheticWebSelection.assert_called_once_with() + + def test_tab_caret_focused_sequence_presents_editable_custom_button_once(self) -> None: + customButton = FakeAccessible(Atspi.Role.PUSH_BUTTON, editable=True) + + self._assert_tab_caret_focused_sequence_presents_once(customButton) + + def test_tab_caret_focused_sequence_presents_native_entry_once(self) -> None: + entry = FakeAccessible(Atspi.Role.ENTRY, editable=True) + + self._assert_tab_caret_focused_sequence_presents_once(entry) + + def test_tab_caret_focused_sequence_presents_native_checkbox_once(self) -> None: + checkbox = FakeAccessible(Atspi.Role.CHECK_BOX) + + self._assert_tab_caret_focused_sequence_presents_once(checkbox) + + def test_tab_caret_focused_sequence_presents_native_radio_once(self) -> None: + radio = FakeAccessible(Atspi.Role.RADIO_BUTTON) + + self._assert_tab_caret_focused_sequence_presents_once(radio) + + +class ChromiumNativeContainerSequenceTests(unittest.TestCase): + """Current routing contracts for native containers in Chromium content.""" + + @staticmethod + def _make_script(document: object, caretContext: object) -> chromium_script.Script: + testScript = chromium_script.Script.__new__(chromium_script.Script) + testScript._lastCommandWasCaretNav = False + testScript._lastCommandWasStructNav = False + testScript._browseModeIsSticky = False + testScript.utilities = mock.Mock() + testScript.utilities.isDocument.return_value = False + testScript.utilities.isZombie.return_value = False + testScript.utilities.getDocumentForObject.return_value = document + testScript.utilities.isWebAppDescendant.return_value = False + testScript.utilities.handleEventFromContextReplicant.return_value = False + testScript.utilities.getCaretContext.return_value = (caretContext, 0) + testScript.utilities.inDocumentContent.return_value = True + testScript.utilities.eventIsBrowserUIAutocompleteNoise.return_value = False + testScript.utilities.eventIsBrowserUIPageSwitch.return_value = False + testScript.utilities.eventIsFromLocusOfFocusDocument.return_value = True + testScript.utilities.eventIsIrrelevantSelectionChangedEvent.return_value = False + testScript.utilities.commonAncestor.return_value = None + return testScript + + def _run_container_sequence( + self, + events: list[FakeEvent], + expectedOwner: str, + ) -> None: + document = object() + oldFocus = object() + testScript = self._make_script(document, oldFocus) + presentations: list[tuple[str, FakeAccessible]] = [] + + with ( + mock.patch.object(cthulhu_state, "locusOfFocus", oldFocus), + mock.patch.object(web_script.AXUtilities, "is_editable", return_value=False), + mock.patch.object(web_script.AXUtilities, "is_dialog_or_alert", return_value=False), + mock.patch.object(web_script.AXUtilities, "is_focusable", return_value=True), + mock.patch.object(web_script.AXUtilities, "is_focused", return_value=True), + mock.patch.object(web_script.AXUtilities, "is_document", return_value=False), + mock.patch.object( + default.Script, + "onActiveDescendantChanged", + side_effect=lambda _script, event: presentations.append( + ("active-descendant", event.any_data) + ), + ) as defaultActiveDescendantHandler, + mock.patch.object( + default.Script, + "onSelectionChanged", + side_effect=lambda _script, event: presentations.append( + ("selection", event.source) + ), + ) as defaultSelectionHandler, + mock.patch.object( + default.Script, + "onFocusedChanged", + side_effect=lambda _script, event: presentations.append(("focus", event.source)), + ) as defaultFocusHandler, + ): + WebEventSequence(testScript).run(events) + + self.assertEqual(presentations, [(expectedOwner, events[-1].source)]) + defaultActiveDescendantHandler.assert_not_called() + if expectedOwner == "selection": + defaultSelectionHandler.assert_called_once_with(testScript, events[-1]) + defaultFocusHandler.assert_not_called() + else: + defaultSelectionHandler.assert_not_called() + defaultFocusHandler.assert_called_once_with(testScript, events[-1]) + + def test_combo_box_active_descendant_then_focus_uses_focus_owner(self) -> None: + comboBox = FakeAccessible(Atspi.Role.COMBO_BOX) + option = FakeAccessible(Atspi.Role.LIST_ITEM) + events = [ + FakeEvent("object:active-descendant-changed", comboBox, any_data=option), + FakeEvent("object:state-changed:focused", comboBox, detail1=1), + ] + + self._run_container_sequence(events, expectedOwner="focus") + + def test_listbox_active_descendant_then_selection_uses_selection_owner(self) -> None: + listBox = FakeAccessible(Atspi.Role.LIST_BOX) + option = FakeAccessible(Atspi.Role.LIST_ITEM) + events = [ + FakeEvent("object:active-descendant-changed", listBox, any_data=option), + FakeEvent("object:selection-changed", listBox), + ] + + self._run_container_sequence(events, expectedOwner="selection") + + def test_tree_active_descendant_then_focus_uses_focus_owner(self) -> None: + tree = FakeAccessible(Atspi.Role.TREE) + treeItem = FakeAccessible(Atspi.Role.TREE_ITEM) + events = [ + FakeEvent("object:active-descendant-changed", tree, any_data=treeItem), + FakeEvent("object:state-changed:focused", treeItem, detail1=1), + ] + + self._run_container_sequence(events, expectedOwner="focus") + + def test_grid_active_descendant_then_selection_uses_selection_owner(self) -> None: + grid = FakeAccessible(Atspi.Role.TABLE) + gridCell = FakeAccessible(Atspi.Role.TABLE_CELL) + events = [ + FakeEvent("object:active-descendant-changed", grid, any_data=gridCell), + FakeEvent("object:selection-changed", grid), + ] + + self._run_container_sequence(events, expectedOwner="selection") + + def test_dialog_focus_loss_then_gain_uses_shared_focus_owner(self) -> None: + document = object() + oldFocus = object() + dialog = FakeAccessible(Atspi.Role.DIALOG) + testScript = self._make_script(document, oldFocus) + events = [ + FakeEvent("object:state-changed:focused", dialog, detail1=0), + FakeEvent("object:state-changed:focused", dialog, detail1=1), + ] + presentations: list[FakeAccessible] = [] + + with ( + mock.patch.object(cthulhu_state, "locusOfFocus", oldFocus), + mock.patch.object(web_script.AXUtilities, "is_editable", return_value=False), + mock.patch.object( + web_script.AXUtilities, + "is_dialog_or_alert", + side_effect=lambda obj: obj is dialog, + ), + mock.patch( + "cthulhu.scripts.web.script.cthulhu.setLocusOfFocus", + side_effect=lambda _event, obj: presentations.append(obj), + ) as setLocusOfFocus, + mock.patch.object(default.Script, "onFocusedChanged") as defaultFocusHandler, + ): + WebEventSequence(testScript).run(events) + + self.assertEqual(presentations, [dialog]) + setLocusOfFocus.assert_called_once_with(events[1], dialog) + defaultFocusHandler.assert_not_called() + + +class ChromiumImplementationDetailSequenceTests(unittest.TestCase): + """Routing contracts for Chromium-only accessibility-tree details.""" + + @staticmethod + def _make_script( + staticTextLeaf: FakeAccessible | None = None, + listItemMarker: FakeAccessible | None = None, + ) -> chromium_script.Script: + testScript = chromium_script.Script.__new__(chromium_script.Script) + testScript.utilities = mock.Mock() + testScript.utilities.isStaticTextLeaf.side_effect = ( + lambda obj: obj is staticTextLeaf + ) + testScript.utilities.isListItemMarker.side_effect = ( + lambda obj: obj is listItemMarker + ) + testScript.utilities.isRedundantAutocompleteEvent.return_value = False + return testScript + + def test_static_text_child_caret_then_parent_caret_uses_parent_owner(self) -> None: + staticText = FakeAccessible(Atspi.Role.STATIC) + parentText = FakeAccessible(Atspi.Role.PARAGRAPH) + testScript = self._make_script(staticTextLeaf=staticText) + events = [ + FakeEvent("object:text-caret-moved", staticText, detail1=0), + FakeEvent("object:text-caret-moved", parentText, detail1=0), + ] + presentations: list[FakeAccessible] = [] + + with ( + mock.patch.object(web_script.Script, "onCaretMoved", return_value=False) as webHandler, + mock.patch.object( + default.Script, + "onCaretMoved", + side_effect=lambda _script, event: presentations.append(event.source), + ) as defaultHandler, + ): + WebEventSequence(testScript).run(events) + + self.assertEqual(presentations, [parentText]) + webHandler.assert_called_once_with(events[1]) + defaultHandler.assert_called_once_with(testScript, events[1]) + + + def test_static_text_child_selection_then_parent_selection_uses_parent_owner(self) -> None: + staticText = FakeAccessible(Atspi.Role.STATIC) + parentText = FakeAccessible(Atspi.Role.PARAGRAPH) + testScript = self._make_script(staticTextLeaf=staticText) + events = [ + FakeEvent("object:text-selection-changed", staticText), + FakeEvent("object:text-selection-changed", parentText), + ] + + self._assert_text_selection_owner(testScript, events, parentText) + + def test_list_marker_selection_then_list_item_selection_uses_list_item_owner(self) -> None: + listMarker = FakeAccessible(Atspi.Role.STATIC) + listItem = FakeAccessible(Atspi.Role.LIST_ITEM) + testScript = self._make_script(listItemMarker=listMarker) + events = [ + FakeEvent("object:text-selection-changed", listMarker), + FakeEvent("object:text-selection-changed", listItem), + ] + + self._assert_text_selection_owner(testScript, events, listItem) + + def _assert_text_selection_owner( + self, + testScript: chromium_script.Script, + events: list[FakeEvent], + expectedOwner: FakeAccessible, + ) -> None: + presentations: list[FakeAccessible] = [] + + with ( + mock.patch.object( + web_script.Script, + "onTextSelectionChanged", + return_value=False, + ) as webHandler, + mock.patch.object( + default.Script, + "onTextSelectionChanged", + side_effect=lambda _script, event: presentations.append(event.source), + ) as defaultHandler, + ): + WebEventSequence(testScript).run(events) + + self.assertEqual(presentations, [expectedOwner]) + webHandler.assert_called_once_with(events[1]) + defaultHandler.assert_called_once_with(testScript, events[1]) + + +class ChromiumMutationRecoverySequenceTests(unittest.TestCase): + """Routing contracts for stale context replacement after child removal.""" + + @staticmethod + def _make_script(document: object, recoveredContext: FakeAccessible) -> chromium_script.Script: + testScript = chromium_script.Script.__new__(chromium_script.Script) + testScript._browseModeIsSticky = False + testScript._loadingDocumentContent = False + testScript.lastMouseRoutingTime = 0 + testScript.utilities = mock.Mock() + testScript.utilities.isStaticTextLeaf.return_value = False + testScript.utilities.eventIsBrowserUINoise.return_value = False + testScript.utilities.isLiveRegion.return_value = False + testScript.utilities.getTopLevelDocumentForObject.return_value = document + testScript.utilities.getDocumentForObject.return_value = document + testScript.utilities.inDocumentContent.return_value = True + testScript.utilities.isZombie.return_value = False + testScript.utilities.isWebAppDescendant.return_value = False + testScript.utilities.handleEventForRemovedChild.return_value = False + testScript.utilities.handleEventFromContextReplicant.side_effect = ( + lambda _event, candidate: candidate is recoveredContext + ) + return testScript + + def _assert_removed_context_then_replicant_recovers_once( + self, + recoveryEventType: str, + ) -> None: + document = object() + container = FakeAccessible(Atspi.Role.PANEL) + removedContext = FakeAccessible(Atspi.Role.PUSH_BUTTON) + recoveredContext = FakeAccessible(Atspi.Role.PUSH_BUTTON) + testScript = self._make_script(document, recoveredContext) + removalEvent = FakeEvent( + "object:children-changed:remove", + container, + detail1=0, + any_data=removedContext, + ) + if recoveryEventType == "object:children-changed:add": + recoveryEvent = FakeEvent( + recoveryEventType, + container, + detail1=0, + any_data=recoveredContext, + ) + else: + recoveryEvent = FakeEvent(recoveryEventType, recoveredContext, detail1=1) + + with ( + mock.patch.object(cthulhu_state, "locusOfFocus", removedContext), + mock.patch.object(web_script.AXObject, "clear_cache_now"), + mock.patch.object(web_script.AXObject, "is_dead", return_value=True), + mock.patch.object(web_script.AXUtilities, "is_busy", return_value=False), + mock.patch.object(web_script.AXUtilities, "is_editable", return_value=False), + mock.patch.object(web_script.AXUtilities, "is_dialog_or_alert", return_value=False), + mock.patch.object(default.Script, "onChildrenRemoved") as defaultRemovalHandler, + mock.patch.object(default.Script, "onChildrenAdded") as defaultAddedHandler, + mock.patch.object(default.Script, "onFocusedChanged") as defaultFocusHandler, + ): + WebEventSequence(testScript).run([removalEvent, recoveryEvent]) + + defaultRemovalHandler.assert_called_once_with(testScript, removalEvent) + testScript.utilities.handleEventForRemovedChild.assert_called_once_with(removalEvent) + testScript.utilities.handleEventFromContextReplicant.assert_called_once_with( + recoveryEvent, + recoveredContext, + ) + defaultAddedHandler.assert_not_called() + defaultFocusHandler.assert_not_called() + + def test_removed_context_then_added_replicant_uses_replicant_owner(self) -> None: + self._assert_removed_context_then_replicant_recovers_once( + "object:children-changed:add" + ) + + def test_removed_context_then_focused_replicant_uses_replicant_owner(self) -> None: + self._assert_removed_context_then_replicant_recovers_once( + "object:state-changed:focused" + ) + + +class StickyModeTransitionSequenceTests(unittest.TestCase): + """Ordered document-exit and re-entry contracts for sticky modes.""" + + @staticmethod + def _make_script( + document: object, + documentObjects: set[FakeAccessible], + inFocusMode: bool, + focusModeIsSticky: bool, + browseModeIsSticky: bool, + ) -> chromium_script.Script: + testScript = chromium_script.Script.__new__(chromium_script.Script) + testScript._lastCommandWasCaretNav = False + testScript._lastCommandWasStructNav = False + testScript._lastCommandWasMouseButton = False + testScript._madeFindAnnouncement = False + testScript._inFocusMode = inFocusMode + testScript._focusModeIsSticky = focusModeIsSticky + testScript._browseModeIsSticky = browseModeIsSticky + testScript._navSuspended = False + testScript.refreshKeyGrabs = mock.Mock() + testScript.presentMessage = mock.Mock() + testScript.utilities = mock.Mock() + testScript.utilities.isZombie.return_value = False + testScript.utilities.isDocument.return_value = False + testScript.utilities.getTopLevelDocumentForObject.side_effect = ( + lambda obj: document if obj in documentObjects else None + ) + testScript.utilities.getDocumentForObject.side_effect = ( + lambda obj: document if obj in documentObjects else None + ) + testScript.utilities.inDocumentContent.side_effect = ( + lambda obj: obj in documentObjects + ) + testScript.utilities.inFindContainer.return_value = False + testScript.utilities.queryNonEmptyText.return_value = None + testScript.utilities.isContentEditableWithEmbeddedObjects.return_value = False + testScript.utilities.isAnchor.return_value = False + testScript.utilities.lastInputEventWasPageNav.return_value = False + testScript.utilities.isFocusedWithMathChild.return_value = False + testScript.utilities.caretMovedToSamePageFragment.return_value = False + testScript.utilities.lastInputEventWasLineNav.return_value = False + testScript.utilities.shouldInterruptForLocusOfFocusChange.return_value = False + testScript.utilities.isWebAppDescendant.return_value = False + testScript.flatReviewPresenter = mock.Mock() + testScript.flatReviewPresenter.is_active.return_value = False + testScript.updateBraille = mock.Mock() + testScript.useFocusMode = mock.Mock(return_value=False) + testScript.speechGenerator = mock.Mock() + testScript.speechGenerator.generateSpeech.return_value = [] + testScript._saveFocusedObjectInfo = mock.Mock() + + def set_navigation_suspended(suspended: bool, _reason: str) -> None: + testScript._navSuspended = suspended + + testScript._setNavigationSuspended = mock.Mock( + side_effect=set_navigation_suspended + ) + return testScript + + @staticmethod + def _run_round_trip( + testScript: chromium_script.Script, + documentControl: FakeAccessible, + browserUI: FakeAccessible, + returnedControl: FakeAccessible, + ) -> list[tuple[bool, bool, bool, bool]]: + snapshots: list[tuple[bool, bool, bool, bool]] = [] + transitions = [ + (documentControl, browserUI), + (browserUI, returnedControl), + ] + for oldFocus, newFocus in transitions: + web_script.Script.locus_of_focus_changed( + testScript, + None, + oldFocus, + newFocus, + ) + snapshots.append( + ( + testScript._inFocusMode, + testScript._focusModeIsSticky, + testScript._browseModeIsSticky, + testScript._navSuspended, + ) + ) + return snapshots + + def test_sticky_focus_survives_document_exit_and_reentry(self) -> None: + document = object() + documentControl = FakeAccessible(Atspi.Role.ENTRY, editable=True) + browserUI = FakeAccessible(Atspi.Role.ENTRY, editable=True) + returnedControl = FakeAccessible(Atspi.Role.ENTRY, editable=True) + testScript = self._make_script( + document, + {documentControl, returnedControl}, + inFocusMode=True, + focusModeIsSticky=True, + browseModeIsSticky=False, + ) + + with ( + mock.patch.object(web_script.AXObject, "is_dead", return_value=False), + mock.patch.object(web_script.AXUtilities, "is_unknown_or_redundant", return_value=False), + mock.patch.object(web_script.AXUtilities, "is_heading", return_value=False), + mock.patch.object(web_script.speech, "speak"), + mock.patch.object(web_script.cthulhu, "emitRegionChanged"), + ): + snapshots = self._run_round_trip( + testScript, + documentControl, + browserUI, + returnedControl, + ) + + self.assertEqual( + snapshots, + [ + (True, True, False, True), + (True, True, False, False), + ], + ) + testScript._setNavigationSuspended.assert_has_calls( + [ + mock.call(True, "focus left document content"), + mock.call(False, "focus entered document content"), + ] + ) + + def test_sticky_browse_survives_round_trip_and_blocks_widget_focus(self) -> None: + document = object() + documentControl = FakeAccessible(Atspi.Role.PARAGRAPH) + browserUI = FakeAccessible(Atspi.Role.ENTRY, editable=True) + webAppWidget = FakeAccessible(Atspi.Role.PUSH_BUTTON) + testScript = self._make_script( + document, + {documentControl, webAppWidget}, + inFocusMode=False, + focusModeIsSticky=False, + browseModeIsSticky=True, + ) + testScript.utilities.isWebAppDescendant.side_effect = ( + lambda obj: obj is webAppWidget + ) + focusEvent = FakeEvent( + "object:state-changed:focused", + webAppWidget, + detail1=1, + ) + + with ( + mock.patch.object(cthulhu_state, "locusOfFocus", browserUI), + mock.patch.object(web_script.AXObject, "is_dead", return_value=False), + mock.patch.object(web_script.AXUtilities, "is_unknown_or_redundant", return_value=False), + mock.patch.object(web_script.AXUtilities, "is_heading", return_value=False), + mock.patch.object(web_script.speech, "speak"), + mock.patch.object(web_script.cthulhu, "emitRegionChanged"), + mock.patch.object(web_script.cthulhu, "setLocusOfFocus") as setLocusOfFocus, + mock.patch.object(default.Script, "onFocusedChanged") as defaultFocusHandler, + ): + snapshots = self._run_round_trip( + testScript, + documentControl, + browserUI, + webAppWidget, + ) + WebEventSequence(testScript).run([focusEvent]) + + self.assertEqual( + snapshots, + [ + (False, False, True, True), + (False, False, True, False), + ], + ) + self.assertFalse(testScript._inFocusMode) + self.assertFalse(testScript._focusModeIsSticky) + self.assertTrue(testScript._browseModeIsSticky) + setLocusOfFocus.assert_not_called() + defaultFocusHandler.assert_not_called() + + +class DocumentAndBrowserUIRoundTripSequenceTests(unittest.TestCase): + """Lifecycle and browser-UI round-trip contracts for web document context.""" + + def test_script_deactivate_then_reactivate_restores_document_navigation(self) -> None: + testScript = chromium_script.Script.__new__(chromium_script.Script) + app = object() + document = object() + focus = FakeAccessible(Atspi.Role.PARAGRAPH) + testScript.app = app + testScript._sayAllContents = [object()] + testScript._inSayAll = True + testScript._sayAllIsInterrupted = True + testScript._loadingDocumentContent = True + testScript._madeFindAnnouncement = True + testScript._lastCommandWasCaretNav = True + testScript._lastCommandWasStructNav = True + testScript._lastCommandWasMouseButton = True + testScript._lastMouseButtonContext = (object(), 3) + testScript._lastMouseOverObject = object() + testScript._preMouseOverContext = (object(), 4) + testScript._inMouseOverObject = True + testScript.utilities = mock.Mock() + testScript.utilities.getDocumentForObject.return_value = document + testScript._setNavigationSuspended = mock.Mock() + testScript.removeKeyGrabs = mock.Mock() + + with ( + mock.patch.object(cthulhu_state, "locusOfFocus", focus), + mock.patch.object(web_script.AXObject, "is_dead", return_value=False), + mock.patch.object(web_script.AXObject, "get_application", return_value=app), + mock.patch.object(default.Script, "activate") as defaultActivate, + ): + web_script.Script.deactivate(testScript) + web_script.Script.activate(testScript) + + self.assertEqual(testScript._sayAllContents, []) + self.assertFalse(testScript._inSayAll) + self.assertFalse(testScript._sayAllIsInterrupted) + self.assertFalse(testScript._loadingDocumentContent) + self.assertFalse(testScript._madeFindAnnouncement) + self.assertFalse(testScript._lastCommandWasCaretNav) + self.assertFalse(testScript._lastCommandWasStructNav) + self.assertFalse(testScript._lastCommandWasMouseButton) + self.assertEqual(testScript._lastMouseButtonContext, (None, -1)) + self.assertIsNone(testScript._lastMouseOverObject) + self.assertEqual(testScript._preMouseOverContext, (None, -1)) + self.assertFalse(testScript._inMouseOverObject) + testScript.utilities.clearCachedObjects.assert_called_once_with() + testScript._setNavigationSuspended.assert_has_calls( + [ + mock.call(False, "script deactivation"), + mock.call(False, "activation-in-document"), + ] + ) + testScript.removeKeyGrabs.assert_called_once_with() + defaultActivate.assert_called_once_with() + + def test_address_bar_round_trip_has_one_owner_per_focus_and_recovers_context( + self, + ) -> None: + document = object() + pageControl = FakeAccessible(Atspi.Role.ENTRY, editable=True) + addressBar = FakeAccessible(Atspi.Role.ENTRY, editable=True) + returnedControl = FakeAccessible(Atspi.Role.ENTRY, editable=True) + testScript = StickyModeTransitionSequenceTests._make_script( + document, + {pageControl, returnedControl}, + inFocusMode=False, + focusModeIsSticky=False, + browseModeIsSticky=False, + ) + focusEvents = [ + FakeEvent("object:state-changed:focused", addressBar, detail1=1), + FakeEvent("object:state-changed:focused", returnedControl, detail1=1), + ] + presentations: list[FakeAccessible] = [] + + def present_and_move_focus( + _script: chromium_script.Script, + event: FakeEvent, + ) -> None: + oldFocus = cthulhu_state.locusOfFocus + presentations.append(event.source) + web_script.Script.locus_of_focus_changed( + testScript, + event, + oldFocus, + event.source, + ) + cthulhu_state.locusOfFocus = event.source + + with ( + mock.patch.object(cthulhu_state, "locusOfFocus", pageControl), + mock.patch.object(web_script.AXObject, "is_dead", return_value=False), + mock.patch.object(web_script.AXUtilities, "is_editable", side_effect=lambda obj: obj.editable), + mock.patch.object(web_script.AXUtilities, "is_dialog_or_alert", return_value=False), + mock.patch.object(web_script.AXUtilities, "is_unknown_or_redundant", return_value=False), + mock.patch.object(web_script.AXUtilities, "is_heading", return_value=False), + mock.patch.object(web_script.speech, "speak"), + mock.patch.object(web_script.cthulhu, "emitRegionChanged"), + mock.patch.object( + default.Script, + "onFocusedChanged", + side_effect=present_and_move_focus, + ) as defaultFocusHandler, + ): + WebEventSequence(testScript).run(focusEvents) + + self.assertEqual(presentations, [addressBar, returnedControl]) + self.assertEqual( + defaultFocusHandler.call_args_list, + [ + mock.call(testScript, focusEvents[0]), + mock.call(testScript, focusEvents[1]), + ], + ) + testScript._setNavigationSuspended.assert_has_calls( + [ + mock.call(True, "focus left document content"), + mock.call(False, "focus entered document content"), + ] + ) + self.assertFalse(testScript._navSuspended) + testScript.utilities.setCaretContext.assert_called_once_with( + returnedControl, + 0, + document, + ) + testScript.updateBraille.assert_called_once_with( + returnedControl, + documentFrame=document, + ) + + +class ChromiumDocumentLifecycleSequenceTests(unittest.TestCase): + """Routing contracts for transient and authoritative Chromium documents.""" + + @staticmethod + def _make_script( + transientDocument: FakeAccessible, + realDocument: FakeAccessible, + ) -> chromium_script.Script: + testScript = chromium_script.Script.__new__(chromium_script.Script) + testScript.utilities = mock.Mock() + testScript.utilities.hasNoSize.return_value = False + testScript.utilities.documentFrameURI.side_effect = ( + lambda obj: "" if obj is transientDocument else "https://example.test/" + ) + return testScript + + def _assert_real_document_owns_sequence( + self, + eventType: str, + webHandlerName: str, + defaultHandlerName: str, + detail1: int = 0, + ) -> None: + transientDocument = FakeAccessible(Atspi.Role.DOCUMENT_WEB) + realDocument = FakeAccessible(Atspi.Role.DOCUMENT_WEB) + testScript = self._make_script(transientDocument, realDocument) + events = [ + FakeEvent(eventType, transientDocument, detail1=detail1), + FakeEvent(eventType, realDocument, detail1=detail1), + ] + handledDocuments: list[FakeAccessible] = [] + + def handle_real_document(event: FakeEvent) -> bool: + handledDocuments.append(event.source) + return True + + with ( + mock.patch.object( + web_script.Script, + webHandlerName, + side_effect=handle_real_document, + ) as webHandler, + mock.patch.object(default.Script, defaultHandlerName) as defaultHandler, + ): + WebEventSequence(testScript).run(events) + + self.assertEqual(handledDocuments, [realDocument]) + webHandler.assert_called_once_with(events[1]) + defaultHandler.assert_not_called() + + def test_uri_less_busy_then_real_busy_uses_real_document_owner(self) -> None: + self._assert_real_document_owns_sequence( + "object:state-changed:busy", + "onBusyChanged", + "onBusyChanged", + detail1=1, + ) + + def test_uri_less_load_complete_then_real_load_complete_uses_real_document_owner( + self, + ) -> None: + self._assert_real_document_owns_sequence( + "document:load-complete", + "onDocumentLoadComplete", + "onDocumentLoadComplete", + ) + + def test_uri_less_reload_then_real_reload_uses_real_document_owner(self) -> None: + self._assert_real_document_owns_sequence( + "document:reload", + "onDocumentReload", + "onDocumentReload", + ) + + +class GeckoFocusAndFindSequenceTests(unittest.TestCase): + """Routing contracts for Gecko focus noise and find-result events.""" + + @staticmethod + def _make_script() -> gecko_script.Script: + testScript = gecko_script.Script.__new__(gecko_script.Script) + testScript._lastCommandWasCaretNav = False + testScript._lastCommandWasStructNav = False + testScript._lastCommandWasMouseButton = False + testScript._clearSyntheticWebSelection = mock.Mock() + testScript._saveFocusedObjectInfo = mock.Mock() + testScript.presentFindResults = mock.Mock() + testScript.utilities = mock.Mock() + testScript.utilities.isZombie.return_value = False + testScript.utilities.getTopLevelDocumentForObject.return_value = object() + testScript.utilities.getCaretContext.return_value = (object(), 0) + testScript.utilities.lastInputEventWasCaretNavWithSelection.return_value = False + testScript.utilities.lastInputEventWasCharNav.return_value = False + testScript.utilities.inFindContainer.return_value = True + return testScript + + def _assert_focus_noise_then_control_uses_control_owner( + self, + noiseRole: Atspi.Role, + ) -> None: + activeWindow = FakeAccessible(Atspi.Role.FRAME) + noise = FakeAccessible(noiseRole) + control = FakeAccessible(Atspi.Role.ENTRY, editable=True) + testScript = self._make_script() + events = [ + FakeEvent("object:state-changed:focused", noise, detail1=1), + FakeEvent("object:state-changed:focused", control, detail1=1), + ] + presentations: list[FakeAccessible] = [] + + with ( + mock.patch.object(cthulhu_state, "activeWindow", activeWindow), + mock.patch.object(cthulhu_state, "locusOfFocus", activeWindow), + mock.patch.object(web_script.Script, "onFocusedChanged", return_value=False) as webHandler, + mock.patch.object( + gecko_script.AXObject, + "get_role", + side_effect=lambda obj: obj.role, + ), + mock.patch.object( + default.Script, + "onFocusedChanged", + side_effect=lambda _script, event: presentations.append(event.source), + ) as defaultHandler, + ): + GeckoEventSequence(testScript).run(events) + + self.assertEqual(presentations, [control]) + self.assertEqual(webHandler.call_args_list, [mock.call(events[0]), mock.call(events[1])]) + defaultHandler.assert_called_once_with(testScript, events[1]) + + def test_panel_focus_noise_then_real_focus_uses_real_control_owner(self) -> None: + self._assert_focus_noise_then_control_uses_control_owner(Atspi.Role.PANEL) + + def test_frame_focus_noise_then_real_focus_uses_real_control_owner(self) -> None: + self._assert_focus_noise_then_control_uses_control_owner(Atspi.Role.FRAME) + + def test_find_entry_focus_then_result_caret_uses_find_presenter(self) -> None: + findEntry = FakeAccessible(Atspi.Role.ENTRY, editable=True) + resultText = FakeAccessible(Atspi.Role.PARAGRAPH) + testScript = self._make_script() + events = [ + FakeEvent("object:state-changed:focused", findEntry, detail1=1), + FakeEvent("object:text-caret-moved", resultText, detail1=7), + ] + focusPresentations: list[FakeAccessible] = [] + + with ( + mock.patch.object(cthulhu_state, "activeWindow", FakeAccessible(Atspi.Role.FRAME)), + mock.patch.object(cthulhu_state, "locusOfFocus", findEntry), + mock.patch.object(web_script.Script, "onFocusedChanged", return_value=False), + mock.patch.object( + gecko_script.AXObject, + "get_role", + side_effect=lambda obj: obj.role, + ), + mock.patch.object( + default.Script, + "onFocusedChanged", + side_effect=lambda _script, event: focusPresentations.append(event.source), + ) as defaultFocusHandler, + mock.patch.object(default.Script, "onCaretMoved") as defaultCaretHandler, + mock.patch.object(AXUtilitiesEvent, "get_text_event_reason", return_value=None), + ): + GeckoEventSequence(testScript).run(events) + + self.assertEqual(focusPresentations, [findEntry]) + defaultFocusHandler.assert_called_once_with(testScript, events[0]) + testScript.presentFindResults.assert_called_once_with(resultText, 7) + testScript._saveFocusedObjectInfo.assert_called_once_with(findEntry) + defaultCaretHandler.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_web_input_regressions.py b/tests/test_web_input_regressions.py index a966589..eb0dac4 100644 --- a/tests/test_web_input_regressions.py +++ b/tests/test_web_input_regressions.py @@ -309,7 +309,6 @@ class WebFocusSpeechInterruptionRegressionTests(unittest.TestCase): testScript.updateBraille = mock.Mock() testScript.presentationInterrupt = mock.Mock() testScript._saveFocusedObjectInfo = mock.Mock() - testScript.refreshKeyGrabs = mock.Mock() return testScript def test_focus_generated_speech_uses_explicit_interrupt_then_appends(self): @@ -493,6 +492,7 @@ class WebDynamicContentRecoveryRegressionTests(unittest.TestCase): testScript.utilities.handleEventFromContextReplicant.return_value = False testScript.utilities.handleEventForRemovedChild.return_value = False testScript.utilities.inDocumentContent.return_value = True + testScript.refreshKeyGrabs = mock.Mock() return testScript def test_children_added_to_live_focus_preserves_caret_context(self): @@ -586,7 +586,6 @@ class WebDynamicContentRecoveryRegressionTests(unittest.TestCase): setLocusOfFocus.assert_called_once_with(event, link, False) testScript.utilities.setCaretContext.assert_called_once_with(link, 0) testScript.utilities.searchForCaretContext.assert_not_called() - def test_browse_mode_sticky_blocks_web_app_descendant_focus_claim(self): testScript = self._make_dynamic_script() source = object()