diff --git a/src/cthulhu/event_manager.py b/src/cthulhu/event_manager.py index ba27685..5903865 100644 --- a/src/cthulhu/event_manager.py +++ b/src/cthulhu/event_manager.py @@ -1715,6 +1715,49 @@ class EventManager: return False + def _event_refreshes_xterm_handoff(self, event: Atspi.Event) -> bool: + """Returns True when an event can represent a real foreground transition.""" + + if event.type.startswith("window:activate"): + return True + + if event.type.startswith("object:state-changed:focused"): + return bool(event.detail1) + + if event.type.startswith("object:state-changed:active") and event.detail1: + return AXUtilities.is_frame(event.source) or AXUtilities.is_window(event.source) + + return False + + def _is_notification_event_during_xterm_handoff(self, event: Atspi.Event) -> bool: + """Returns True when an event should bypass XTerm handoff suppression.""" + + for obj in (event.source, event.any_data): + if obj is None or isinstance(obj, (str, bytes, int, float, bool)): + continue + try: + if AXUtilities.is_notification(obj) or AXUtilities.is_alert(obj): + return True + except Exception: + pass + + notificationEventTypes = ( + "window:create", + "object:property-change:accessible-name", + "object:property-change:accessible-value", + ) + if not event.type.startswith(notificationEventTypes): + return False + + app = AXObject.get_application(event.source) + appName = (AXObject.get_name(app) or "").casefold() + return appName in ( + "notification-daemon", + "notify-osd", + "xfce4-notifyd", + "mate-notification-daemon", + ) + def _processObjectEvent(self, event: Atspi.Event) -> None: """Handles all object events destined for scripts. @@ -1728,6 +1771,21 @@ class EventManager: debug.printObjectEvent(debug.LEVEL_INFO, event, timestamp=True) eType = event.type + refreshXtermHandoff = self._event_refreshes_xterm_handoff(event) + isNotification = self._is_notification_event_during_xterm_handoff(event) + suppressForXterm = not isNotification and self._inputEventManager is not None and ( + self._inputEventManager.should_suppress_atspi_events_for_xterm( + refresh=refreshXtermHandoff, + ) + ) + if suppressForXterm: + msg = ( + "EVENT MANAGER: Ignoring AT-SPI event while XTerm owns the " + "foreground handoff." + ) + debug.printMessage(debug.LEVEL_INFO, msg, True) + return + if eType.startswith("object:children-changed:remove") \ and event.source == AXUtilities.get_desktop(): cthulhu.cthulhuApp.scriptManager.reclaim_scripts() diff --git a/src/cthulhu/input_event_manager.py b/src/cthulhu/input_event_manager.py index f04753c..7119485 100644 --- a/src/cthulhu/input_event_manager.py +++ b/src/cthulhu/input_event_manager.py @@ -120,6 +120,13 @@ class InputEventManager: self._secure_input_active: bool = False self._wnck = None self._did_attempt_wnck_load: bool = False + self._x11Display = None + self._x11Root = None + self._netActiveWindowAtom = None + self._x11AnyPropertyType = None + self._didAttemptX11Display: bool = False + self._lastEwmhActiveWindowId: Optional[int] = None + self._lastEwmhXtermMatch: Optional[bool] = None self._scriptWithSuspendedGrabsForXterm = None self._xtermFocusScreen = None self._xtermFocusHandlerId: int = 0 @@ -177,6 +184,9 @@ class InputEventManager: debug.print_message(debug.LEVEL_INFO, msg, True) device.add_key_watcher(self.process_keyboard_event) + if os.environ.get("XDG_SESSION_TYPE", "").lower() == "x11": + self._initialize_xterm_handoff() + def stop_key_watcher(self) -> None: """Stops the watcher for keyboard input events.""" @@ -201,6 +211,8 @@ class InputEventManager: self._device = None self._stop_xterm_focus_monitor() self._stop_xterm_recovery_timer() + self._close_x11_display() + self._didAttemptX11Display = False self._scriptWithSuspendedGrabsForXterm = None def get_debug_snapshot(self) -> Dict[str, object]: @@ -215,6 +227,14 @@ class InputEventManager: "paused": self._paused, "secure_input_active": self._secure_input_active, "suspended_xterm_script_present": self._scriptWithSuspendedGrabsForXterm is not None, + "xterm_focus_monitor_active": self._xtermFocusHandlerId != 0, + "xterm_recovery_timer_active": self._xtermRecoverySourceId != 0, + "last_ewmh_active_window_id": ( + hex(self._lastEwmhActiveWindowId) + if self._lastEwmhActiveWindowId is not None + else "" + ), + "last_ewmh_xterm_match": self._lastEwmhXtermMatch, "last_input_event_present": self._last_input_event is not None, "last_non_modifier_key_event_present": self._last_non_modifier_key_event is not None, } @@ -524,8 +544,8 @@ class InputEventManager: return self._wnck - def _get_active_x11_window(self) -> Optional[Any]: - """Returns the active X11 window if Wnck can provide it.""" + def _get_active_x11_window(self, ewmhWindowId: Optional[int] = None) -> Optional[Any]: + """Returns the EWMH active X11 window, using Wnck for its metadata.""" wnck = self._get_wnck() if wnck is None: @@ -537,12 +557,78 @@ class InputEventManager: return None self._ensure_xterm_focus_monitor(screen) screen.force_update() + windowId = ewmhWindowId + if windowId is None: + windowId = self._get_ewmh_active_x11_window_id() + if windowId is not None: + window = wnck.Window.get(windowId) + if window is None: + tokens = [ + "INPUT EVENT MANAGER: Wnck has no window for EWMH active X11 id", + hex(windowId), + ] + debug.print_tokens(debug.LEVEL_INFO, tokens, True) + return window return screen.get_active_window() except Exception as error: msg = f"INPUT EVENT MANAGER: Could not obtain active X11 window: {error}" debug.print_message(debug.LEVEL_INFO, msg, True) return None + def _get_ewmh_active_x11_window_id(self) -> Optional[int]: + """Returns _NET_ACTIVE_WINDOW from the X11 root window when available.""" + + if not self._didAttemptX11Display: + self._didAttemptX11Display = True + try: + from Xlib import X # pylint: disable=import-outside-toplevel + from Xlib import display as xDisplay # pylint: disable=import-outside-toplevel + + self._x11Display = xDisplay.Display() + self._x11Root = self._x11Display.screen().root + self._netActiveWindowAtom = self._x11Display.intern_atom("_NET_ACTIVE_WINDOW") + self._x11AnyPropertyType = X.AnyPropertyType + except Exception as error: + msg = f"INPUT EVENT MANAGER: EWMH active-window lookup unavailable: {error}" + debug.print_message(debug.LEVEL_INFO, msg, True) + self._close_x11_display() + + if self._x11Root is None or self._netActiveWindowAtom is None: + return None + + try: + propertyData = self._x11Root.get_full_property( + self._netActiveWindowAtom, + self._x11AnyPropertyType, + ) + if propertyData is None or not propertyData.value: + return None + windowId = int(propertyData.value[0]) + return windowId if windowId > 0 else None + except Exception as error: + msg = f"INPUT EVENT MANAGER: Could not read EWMH active window: {error}" + debug.print_message(debug.LEVEL_INFO, msg, True) + return None + + def _close_x11_display(self) -> None: + """Closes the optional X11 display used for EWMH focus lookup.""" + + x11Display = self._x11Display + self._x11Display = None + self._x11Root = None + self._netActiveWindowAtom = None + self._x11AnyPropertyType = None + self._lastEwmhActiveWindowId = None + self._lastEwmhXtermMatch = None + if x11Display is None: + return + + try: + x11Display.close() + except Exception as error: + msg = f"INPUT EVENT MANAGER: Could not close EWMH display: {error}" + debug.print_message(debug.LEVEL_INFO, msg, True) + def _get_active_x11_window_pid(self) -> int: """Returns the PID of the active X11 window if Wnck can provide it.""" @@ -641,6 +727,16 @@ class InputEventManager: self._xtermFocusScreen = screen self._xtermFocusHandlerId = handlerId + def _initialize_xterm_handoff(self) -> None: + """Initializes X11 focus monitoring and suspends immediately for XTerm.""" + + match = self._active_x11_window_xterm_match() + tokens = ["INPUT EVENT MANAGER: Initial XTerm handoff matcher returned", match] + debug.print_tokens(debug.LEVEL_INFO, tokens, True) + if match is True: + self._suspend_key_grabs_for_xterm() + self._ensure_xterm_recovery_timer() + def _stop_xterm_recovery_timer(self) -> None: """Stops the fallback timer used to recover XTerm-suspended grabs.""" @@ -656,7 +752,7 @@ class InputEventManager: debug.print_message(debug.LEVEL_INFO, msg, True) def _ensure_xterm_recovery_timer(self) -> None: - """Starts fallback recovery when X11 focus notifications are insufficient.""" + """Starts generic X11 polling when focus notifications are insufficient.""" if self._device is None or self._xtermRecoverySourceId: return @@ -667,52 +763,50 @@ class InputEventManager: ) def _poll_xterm_grab_recovery(self) -> bool: - """Checks whether XTerm-suspended grabs can now be restored.""" + """Updates XTerm handoff state from the authoritative EWMH active window.""" - if self._scriptWithSuspendedGrabsForXterm is None: + if self._device is None: self._xtermRecoverySourceId = 0 return False match = self._active_x11_window_xterm_match() - tokens = ["INPUT EVENT MANAGER: XTerm recovery matcher returned", match] - debug.print_tokens(debug.LEVEL_INFO, tokens, True) - if match is False: - self._xtermRecoverySourceId = 0 + if match is True: + self._suspend_key_grabs_for_xterm() + elif match is False: self._restore_key_grabs_after_xterm() - return False - - keepPolling = not self._xtermFocusHandlerId or match is None - if not keepPolling: - self._xtermRecoverySourceId = 0 - return keepPolling + return True def _on_active_x11_window_changed(self, screen: Any, _previousWindow: Any) -> None: - """Restores suspended grabs as soon as X11 focus leaves XTerm.""" + """Updates grab suspension as soon as X11 focus changes.""" - if self._scriptWithSuspendedGrabsForXterm is None: - return - - try: - window = screen.get_active_window() - except Exception as error: - msg = f"INPUT EVENT MANAGER: Could not check changed X11 focus: {error}" - debug.print_message(debug.LEVEL_INFO, msg, True) - self._ensure_xterm_recovery_timer() - return - - match = self._x11_window_xterm_match(window) + del screen + match = self._active_x11_window_xterm_match() tokens = ["INPUT EVENT MANAGER: XTerm focus-change matcher returned", match] debug.print_tokens(debug.LEVEL_INFO, tokens, True) - if match is False: + if match is True: + self._suspend_key_grabs_for_xterm() + elif match is False: self._restore_key_grabs_after_xterm() - elif match is None: + elif self._scriptWithSuspendedGrabsForXterm is not None: self._ensure_xterm_recovery_timer() def _active_x11_window_xterm_match(self) -> Optional[bool]: """Returns whether the active X11 window is XTerm, or None when unknown.""" - window = self._get_active_x11_window() - return self._x11_window_xterm_match(window) + windowId = self._get_ewmh_active_x11_window_id() + if ( + windowId is not None + and windowId == self._lastEwmhActiveWindowId + and self._lastEwmhXtermMatch is not None + ): + return self._lastEwmhXtermMatch + + window = self._get_active_x11_window(ewmhWindowId=windowId) + match = self._x11_window_xterm_match(window) + if windowId is not None and match is not None: + self._lastEwmhActiveWindowId = windowId + self._lastEwmhXtermMatch = match + return match def _x11_window_xterm_match(self, window: Any) -> Optional[bool]: """Returns whether window is XTerm, or None when it cannot be identified.""" @@ -775,6 +869,25 @@ class InputEventManager: return self._active_x11_window_xterm_match() is True + def should_suppress_atspi_events_for_xterm(self, refresh: bool = False) -> bool: + """Returns True while the XTerm handoff should suppress AT-SPI events.""" + + suspendedScript = self._scriptWithSuspendedGrabsForXterm + if suspendedScript is None: + return False + + if not refresh: + return True + + match = self._active_x11_window_xterm_match() + tokens = ["INPUT EVENT MANAGER: XTerm AT-SPI suppression matcher returned", match] + debug.print_tokens(debug.LEVEL_INFO, tokens, True) + if match is False: + self._restore_key_grabs_after_xterm() + return False + + return True + def _find_active_x11_atspi_window(self) -> Optional[Atspi.Accessible]: """Returns the focused AT-SPI window for the active X11 PID, if possible.""" @@ -941,8 +1054,7 @@ class InputEventManager: debug.print_tokens(debug.LEVEL_INFO, tokens, True) removeGrabs() self._scriptWithSuspendedGrabsForXterm = script - if not self._xtermFocusHandlerId: - self._ensure_xterm_recovery_timer() + self._ensure_xterm_recovery_timer() def _restore_key_grabs_after_xterm(self) -> None: """Restores key grabs after leaving XTerm.""" @@ -960,7 +1072,6 @@ class InputEventManager: ] debug.print_tokens(debug.LEVEL_INFO, tokens, True) self._scriptWithSuspendedGrabsForXterm = None - self._stop_xterm_recovery_timer() return addGrabs = getattr(script, "addKeyGrabs", None) @@ -988,13 +1099,11 @@ class InputEventManager: ) debug.print_message(debug.LEVEL_INFO, msg, True) self._scriptWithSuspendedGrabsForXterm = None - self._stop_xterm_recovery_timer() return self._ensure_xterm_recovery_timer() return self._scriptWithSuspendedGrabsForXterm = None - self._stop_xterm_recovery_timer() # pylint: disable=too-many-arguments # pylint: disable=too-many-positional-arguments diff --git a/src/cthulhu/scripts/toolkits/Chromium/script.py b/src/cthulhu/scripts/toolkits/Chromium/script.py index e0da14e..db52607 100644 --- a/src/cthulhu/scripts/toolkits/Chromium/script.py +++ b/src/cthulhu/scripts/toolkits/Chromium/script.py @@ -171,12 +171,20 @@ class Script(web.Script): def onColumnReordered(self, event): """Callback for object:column-reordered accessibility events.""" - if super().onColumnReordered(event): - return + def handle_shared_web(routed_event: object) -> RouteResult: + handled = web.Script.onColumnReordered(self, routed_event) + return RouteResult.HANDLED if handled else RouteResult.CONTINUE - msg = "CHROMIUM: Passing along event to default script" - debug.printMessage(debug.LEVEL_INFO, msg, True) - default.Script.onColumnReordered(self, event) + self._eventRouter.route_event( + event, + application_handler=None, + toolkit_handler=None, + web_handler=handle_shared_web, + default_handler=lambda routed_event: default.Script.onColumnReordered( + self, + routed_event, + ), + ) def onChildrenAdded(self, event): """Callback for object:children-changed:add accessibility events.""" @@ -256,12 +264,20 @@ class Script(web.Script): def onExpandedChanged(self, event): """Callback for object:state-changed:expanded accessibility events.""" - if super().onExpandedChanged(event): - return + def handle_shared_web(routed_event: object) -> RouteResult: + handled = web.Script.onExpandedChanged(self, routed_event) + return RouteResult.HANDLED if handled else RouteResult.CONTINUE - msg = "CHROMIUM: Passing along event to default script" - debug.printMessage(debug.LEVEL_INFO, msg, True) - default.Script.onExpandedChanged(self, event) + self._eventRouter.route_event( + event, + application_handler=None, + toolkit_handler=None, + web_handler=handle_shared_web, + default_handler=lambda routed_event: default.Script.onExpandedChanged( + self, + routed_event, + ), + ) def onFocus(self, event): """Callback for focus: accessibility events.""" @@ -315,12 +331,20 @@ class Script(web.Script): def onRowReordered(self, event): """Callback for object:row-reordered accessibility events.""" - if super().onRowReordered(event): - return + def handle_shared_web(routed_event: object) -> RouteResult: + handled = web.Script.onRowReordered(self, routed_event) + return RouteResult.HANDLED if handled else RouteResult.CONTINUE - msg = "CHROMIUM: Passing along event to default script" - debug.printMessage(debug.LEVEL_INFO, msg, True) - default.Script.onRowReordered(self, event) + self._eventRouter.route_event( + event, + application_handler=None, + toolkit_handler=None, + web_handler=handle_shared_web, + default_handler=lambda routed_event: default.Script.onRowReordered( + self, + routed_event, + ), + ) def onSelectedChanged(self, event): """Callback for object:state-changed:selected accessibility events.""" diff --git a/src/cthulhu/scripts/toolkits/Gecko/script.py b/src/cthulhu/scripts/toolkits/Gecko/script.py index 032d9aa..ba0b1eb 100644 --- a/src/cthulhu/scripts/toolkits/Gecko/script.py +++ b/src/cthulhu/scripts/toolkits/Gecko/script.py @@ -144,12 +144,20 @@ class Script(web.Script): def onColumnReordered(self, event): """Callback for object:column-reordered accessibility events.""" - if super().onColumnReordered(event): - return + def handle_shared_web(routed_event: object) -> RouteResult: + handled = web.Script.onColumnReordered(self, routed_event) + return RouteResult.HANDLED if handled else RouteResult.CONTINUE - msg = "GECKO: Passing along event to default script" - debug.printMessage(debug.LEVEL_INFO, msg, True) - default.Script.onColumnReordered(self, event) + self._eventRouter.route_event( + event, + application_handler=None, + toolkit_handler=None, + web_handler=handle_shared_web, + default_handler=lambda routed_event: default.Script.onColumnReordered( + self, + routed_event, + ), + ) def onChildrenAdded(self, event): """Callback for object:children-changed:add accessibility events.""" @@ -209,12 +217,20 @@ class Script(web.Script): def onExpandedChanged(self, event): """Callback for object:state-changed:expanded accessibility events.""" - if super().onExpandedChanged(event): - return + def handle_shared_web(routed_event: object) -> RouteResult: + handled = web.Script.onExpandedChanged(self, routed_event) + return RouteResult.HANDLED if handled else RouteResult.CONTINUE - msg = "GECKO: Passing along event to default script" - debug.printMessage(debug.LEVEL_INFO, msg, True) - default.Script.onExpandedChanged(self, event) + self._eventRouter.route_event( + event, + application_handler=None, + toolkit_handler=None, + web_handler=handle_shared_web, + default_handler=lambda routed_event: default.Script.onExpandedChanged( + self, + routed_event, + ), + ) def onFocus(self, event): """Callback for focus: accessibility events.""" @@ -285,12 +301,20 @@ class Script(web.Script): def onRowReordered(self, event): """Callback for object:row-reordered accessibility events.""" - if super().onRowReordered(event): - return + def handle_shared_web(routed_event: object) -> RouteResult: + handled = web.Script.onRowReordered(self, routed_event) + return RouteResult.HANDLED if handled else RouteResult.CONTINUE - msg = "GECKO: Passing along event to default script" - debug.printMessage(debug.LEVEL_INFO, msg, True) - default.Script.onRowReordered(self, event) + self._eventRouter.route_event( + event, + application_handler=None, + toolkit_handler=None, + web_handler=handle_shared_web, + default_handler=lambda routed_event: default.Script.onRowReordered( + self, + routed_event, + ), + ) def onSelectedChanged(self, event): """Callback for object:state-changed:selected accessibility events.""" diff --git a/tests/test_event_manager_xterm_handoff_regressions.py b/tests/test_event_manager_xterm_handoff_regressions.py new file mode 100644 index 0000000..01b0a2e --- /dev/null +++ b/tests/test_event_manager_xterm_handoff_regressions.py @@ -0,0 +1,129 @@ +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")) + +stubCthulhu = types.ModuleType("cthulhu.cthulhu") +stubCthulhu.cthulhuApp = mock.Mock() +sys.modules.setdefault("cthulhu.cthulhu", stubCthulhu) + +from cthulhu import event_manager + + +class FakeEvent: + def __init__(self, event_type, source="window", detail1=0): + self.type = event_type + self.source = source + self.detail1 = detail1 + self.detail2 = 0 + self.any_data = None + + +class EventManagerXtermHandoffRegressionTests(unittest.TestCase): + def setUp(self) -> None: + listener = mock.Mock() + fakeAtspi = types.SimpleNamespace( + Accessible=event_manager.Atspi.Accessible, + EventListener=types.SimpleNamespace(new=mock.Mock(return_value=listener)), + Role=event_manager.Atspi.Role, + ) + self.listenerPatch = mock.patch.object(event_manager, "Atspi", fakeAtspi) + self.listenerPatch.start() + self.addCleanup(self.listenerPatch.stop) + + self.manager = event_manager.EventManager(mock.Mock(), asyncMode=False) + self.manager._active = True + self.manager._inputEventManager = mock.Mock() + self.manager._inputEventManager.should_suppress_atspi_events_for_xterm.return_value = True + self.manager._get_scriptForEvent = mock.Mock() + + def test_window_activate_is_dropped_while_xterm_is_foreground(self) -> None: + event = FakeEvent("window:activate") + + with ( + mock.patch.object(event_manager.debug, "printObjectEvent"), + mock.patch.object(event_manager.debug, "printMessage"), + ): + self.manager._processObjectEvent(event) + + self.manager._get_scriptForEvent.assert_not_called() + self.manager._inputEventManager.should_suppress_atspi_events_for_xterm.assert_called_once_with( + refresh=True, + ) + + def test_active_frame_event_is_dropped_while_xterm_is_foreground(self) -> None: + event = FakeEvent("object:state-changed:active", detail1=1) + + with ( + mock.patch.object(event_manager.debug, "printObjectEvent"), + mock.patch.object(event_manager.debug, "printMessage"), + mock.patch.object(event_manager.AXUtilities, "is_frame", return_value=True), + ): + self.manager._processObjectEvent(event) + + self.manager._get_scriptForEvent.assert_not_called() + self.manager._inputEventManager.should_suppress_atspi_events_for_xterm.assert_called_once_with( + refresh=True, + ) + + def test_focused_event_is_dropped_while_xterm_is_foreground(self) -> None: + event = FakeEvent("object:state-changed:focused", detail1=1) + + with ( + mock.patch.object(event_manager.debug, "printObjectEvent"), + mock.patch.object(event_manager.debug, "printMessage"), + ): + self.manager._processObjectEvent(event) + + self.manager._get_scriptForEvent.assert_not_called() + self.manager._inputEventManager.should_suppress_atspi_events_for_xterm.assert_called_once_with( + refresh=True, + ) + + def test_background_event_is_dropped_during_xterm_handoff_without_refresh(self) -> None: + event = FakeEvent("object:text-changed:insert") + + with ( + mock.patch.object(event_manager.debug, "printObjectEvent"), + mock.patch.object(event_manager.debug, "printMessage"), + ): + self.manager._processObjectEvent(event) + + self.manager._get_scriptForEvent.assert_not_called() + self.manager._inputEventManager.should_suppress_atspi_events_for_xterm.assert_called_once_with( + refresh=False, + ) + + def test_notification_daemon_window_create_is_processed_during_xterm_handoff(self) -> None: + event = FakeEvent("window:create", source=object()) + notificationApp = object() + self.manager._isActivatableEvent = mock.Mock(return_value=(False, "test")) + + with ( + mock.patch.object(event_manager.debug, "printObjectEvent"), + mock.patch.object(event_manager.debug, "printMessage"), + mock.patch.object( + event_manager.AXObject, + "get_application", + return_value=notificationApp, + ), + mock.patch.object( + event_manager.AXObject, + "get_name", + return_value="xfce4-notifyd", + ), + mock.patch.object(event_manager.AXObject, "is_dead", return_value=False), + mock.patch.object(event_manager.AXUtilities, "is_defunct", return_value=False), + mock.patch.object(event_manager.AXUtilities, "is_iconified", return_value=False), + ): + self.manager._processObjectEvent(event) + + self.manager._get_scriptForEvent.assert_called_once_with(event) + self.manager._inputEventManager.should_suppress_atspi_events_for_xterm.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_input_event_manager_key_watcher_regressions.py b/tests/test_input_event_manager_key_watcher_regressions.py index 81413fb..e9f0791 100644 --- a/tests/test_input_event_manager_key_watcher_regressions.py +++ b/tests/test_input_event_manager_key_watcher_regressions.py @@ -130,6 +130,7 @@ class InputEventManagerKeyWatcherTests(unittest.TestCase): ), 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() @@ -161,6 +162,7 @@ class InputEventManagerKeyWatcherTests(unittest.TestCase): mask=expectedMasks, kind=(0, 1), ) + initializeXtermHandoff.assert_called_once_with() self.assertEqual( device.connect_calls, [ diff --git a/tests/test_input_event_manager_x11_focus_regressions.py b/tests/test_input_event_manager_x11_focus_regressions.py index e6bff40..df988ae 100644 --- a/tests/test_input_event_manager_x11_focus_regressions.py +++ b/tests/test_input_event_manager_x11_focus_regressions.py @@ -530,19 +530,200 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase): activeScript = mock.Mock() scriptManager = mock.Mock() scriptManager.get_active_script.return_value = activeScript - screen = mock.Mock() - activeWindow = object() - screen.get_active_window.return_value = activeWindow manager._scriptWithSuspendedGrabsForXterm = activeScript with ( mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager), - mock.patch.object(manager, "_x11_window_xterm_match", return_value=False) as matcher, + mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=False) as matcher, mock.patch.object(input_event_manager.debug, "print_tokens"), ): - manager._on_active_x11_window_changed(screen, None) + manager._on_active_x11_window_changed(mock.Mock(), None) + matcher.assert_called_once_with() + activeScript.addKeyGrabs.assert_called_once_with() + self.assertIsNone(manager._scriptWithSuspendedGrabsForXterm) + + def test_active_x11_window_uses_ewmh_id_instead_of_stale_wnck_active_window(self): + manager = input_event_manager.InputEventManager() + wnck = mock.Mock() + screen = mock.Mock() + ewmhWindow = object() + wnck.Screen.get_default.return_value = screen + wnck.Window.get.return_value = ewmhWindow + + with ( + mock.patch.object(manager, "_get_wnck", return_value=wnck), + mock.patch.object(manager, "_ensure_xterm_focus_monitor"), + mock.patch.object(manager, "_get_ewmh_active_x11_window_id", return_value=0x1400010), + ): + result = manager._get_active_x11_window() + + self.assertIs(result, ewmhWindow) + wnck.Window.get.assert_called_once_with(0x1400010) + screen.get_active_window.assert_not_called() + + def test_active_x11_window_does_not_fall_back_to_stale_wnck_window_for_known_ewmh_id(self): + manager = input_event_manager.InputEventManager() + wnck = mock.Mock() + screen = mock.Mock() + wnck.Screen.get_default.return_value = screen + wnck.Window.get.return_value = None + + with ( + mock.patch.object(manager, "_get_wnck", return_value=wnck), + mock.patch.object(manager, "_ensure_xterm_focus_monitor"), + mock.patch.object(manager, "_get_ewmh_active_x11_window_id", return_value=0x1400010), + mock.patch.object(input_event_manager.debug, "print_tokens"), + ): + result = manager._get_active_x11_window() + + self.assertIsNone(result) + screen.get_active_window.assert_not_called() + + def test_active_x11_window_falls_back_to_wnck_when_ewmh_is_unavailable(self): + manager = input_event_manager.InputEventManager() + wnck = mock.Mock() + screen = mock.Mock() + fallbackWindow = object() + wnck.Screen.get_default.return_value = screen + screen.get_active_window.return_value = fallbackWindow + + with ( + mock.patch.object(manager, "_get_wnck", return_value=wnck), + mock.patch.object(manager, "_ensure_xterm_focus_monitor"), + mock.patch.object(manager, "_get_ewmh_active_x11_window_id", return_value=None), + ): + result = manager._get_active_x11_window() + + self.assertIs(result, fallbackWindow) + + def test_xterm_match_reuses_cached_metadata_while_ewmh_window_id_is_unchanged(self): + manager = input_event_manager.InputEventManager() + activeWindow = object() + + with ( + mock.patch.object( + manager, + "_get_ewmh_active_x11_window_id", + return_value=0x1400010, + ), + mock.patch.object( + manager, + "_get_active_x11_window", + return_value=activeWindow, + ) as getActiveWindow, + mock.patch.object( + manager, + "_x11_window_xterm_match", + return_value=False, + ) as matcher, + ): + firstResult = manager._active_x11_window_xterm_match() + secondResult = manager._active_x11_window_xterm_match() + + self.assertFalse(firstResult) + self.assertFalse(secondResult) + getActiveWindow.assert_called_once_with(ewmhWindowId=0x1400010) matcher.assert_called_once_with(activeWindow) + + def test_xterm_focus_change_suspends_grabs_without_keyboard_event(self): + manager = input_event_manager.InputEventManager() + activeScript = mock.Mock() + scriptManager = mock.Mock() + scriptManager.get_active_script.return_value = activeScript + + with ( + mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager), + mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=True) as matcher, + mock.patch.object(input_event_manager.debug, "print_tokens"), + ): + manager._on_active_x11_window_changed(mock.Mock(), None) + + matcher.assert_called_once_with() + activeScript.removeKeyGrabs.assert_called_once_with() + self.assertIs(manager._scriptWithSuspendedGrabsForXterm, activeScript) + + def test_initial_xterm_handoff_suspends_grabs(self): + manager = input_event_manager.InputEventManager() + manager._device = object() + + with ( + mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=True), + mock.patch.object(manager, "_suspend_key_grabs_for_xterm") as suspend, + mock.patch.object(manager, "_ensure_xterm_recovery_timer") as ensureTimer, + mock.patch.object(input_event_manager.debug, "print_tokens"), + ): + manager._initialize_xterm_handoff() + + suspend.assert_called_once_with() + ensureTimer.assert_called_once_with() + + def test_xterm_focus_poll_suspends_grabs_after_entering_xterm(self): + manager = input_event_manager.InputEventManager() + manager._device = object() + + with ( + mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=True), + mock.patch.object(manager, "_suspend_key_grabs_for_xterm") as suspend, + mock.patch.object(manager, "_restore_key_grabs_after_xterm") as restore, + mock.patch.object(input_event_manager.debug, "print_tokens"), + ): + result = manager._poll_xterm_grab_recovery() + + self.assertTrue(result) + suspend.assert_called_once_with() + restore.assert_not_called() + + def test_xterm_focus_poll_restores_grabs_after_leaving_xterm(self): + manager = input_event_manager.InputEventManager() + manager._device = object() + + with ( + mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=False), + mock.patch.object(manager, "_suspend_key_grabs_for_xterm") as suspend, + mock.patch.object(manager, "_restore_key_grabs_after_xterm") as restore, + mock.patch.object(input_event_manager.debug, "print_tokens"), + ): + result = manager._poll_xterm_grab_recovery() + + self.assertTrue(result) + restore.assert_called_once_with() + suspend.assert_not_called() + + def test_xterm_suppression_does_not_check_x11_without_active_handoff(self): + manager = input_event_manager.InputEventManager() + + with mock.patch.object(manager, "_active_x11_window_xterm_match") as matcher: + result = manager.should_suppress_atspi_events_for_xterm(refresh=True) + + self.assertFalse(result) + matcher.assert_not_called() + + def test_xterm_suppression_ignores_background_event_without_x11_refresh(self): + manager = input_event_manager.InputEventManager() + manager._scriptWithSuspendedGrabsForXterm = object() + + with mock.patch.object(manager, "_active_x11_window_xterm_match") as matcher: + result = manager.should_suppress_atspi_events_for_xterm(refresh=False) + + self.assertTrue(result) + matcher.assert_not_called() + + def test_xterm_suppression_releases_handoff_for_real_focus_return(self): + manager = input_event_manager.InputEventManager() + activeScript = mock.Mock() + scriptManager = mock.Mock() + scriptManager.get_active_script.return_value = activeScript + manager._scriptWithSuspendedGrabsForXterm = activeScript + + with ( + mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager), + mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=False), + mock.patch.object(input_event_manager.debug, "print_tokens"), + ): + result = manager.should_suppress_atspi_events_for_xterm(refresh=True) + + self.assertFalse(result) activeScript.addKeyGrabs.assert_called_once_with() self.assertIsNone(manager._scriptWithSuspendedGrabsForXterm) @@ -550,14 +731,13 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase): manager = input_event_manager.InputEventManager() manager._device = object() manager._scriptWithSuspendedGrabsForXterm = object() - screen = mock.Mock() - screen.get_active_window.side_effect = RuntimeError("focus unavailable") with ( + mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=None), mock.patch.object(input_event_manager.GLib, "timeout_add", return_value=23), - mock.patch.object(input_event_manager.debug, "print_message"), + mock.patch.object(input_event_manager.debug, "print_tokens"), ): - manager._on_active_x11_window_changed(screen, None) + manager._on_active_x11_window_changed(mock.Mock(), None) self.assertEqual(manager._xtermRecoverySourceId, 23) @@ -619,6 +799,9 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase): manager._xtermFocusHandlerId = 17 manager._xtermRecoverySourceId = 23 manager._scriptWithSuspendedGrabsForXterm = object() + x11Display = mock.Mock() + manager._x11Display = x11Display + manager._didAttemptX11Display = True with ( mock.patch.object(input_event_manager.GLib, "source_remove") as sourceRemove, @@ -632,6 +815,8 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase): self.assertEqual(manager._xtermFocusHandlerId, 0) self.assertEqual(manager._xtermRecoverySourceId, 0) self.assertIsNone(manager._scriptWithSuspendedGrabsForXterm) + x11Display.close.assert_called_once_with() + self.assertFalse(manager._didAttemptX11Display) def test_identifier_is_xterm_matches_exact_xterm_only(self): self.assertTrue(input_event_manager.InputEventManager._identifier_is_xterm("xterm")) diff --git a/tests/test_runtime_state_snapshot.py b/tests/test_runtime_state_snapshot.py index 75d5003..6596020 100644 --- a/tests/test_runtime_state_snapshot.py +++ b/tests/test_runtime_state_snapshot.py @@ -80,6 +80,10 @@ class RuntimeStateSnapshotTests(unittest.TestCase): manager._grabbed_bindings = {10: object(), 11: object()} manager._paused = True manager._scriptWithSuspendedGrabsForXterm = object() + manager._xtermFocusHandlerId = 17 + manager._xtermRecoverySourceId = 23 + manager._lastEwmhActiveWindowId = 0x1400010 + manager._lastEwmhXtermMatch = False manager._last_input_event = object() manager._last_non_modifier_key_event = None @@ -91,6 +95,10 @@ class RuntimeStateSnapshotTests(unittest.TestCase): self.assertEqual(1, snapshot["mapped_keysyms"]) self.assertEqual(2, snapshot["grabbed_bindings"]) self.assertTrue(snapshot["suspended_xterm_script_present"]) + self.assertTrue(snapshot["xterm_focus_monitor_active"]) + self.assertTrue(snapshot["xterm_recovery_timer_active"]) + self.assertEqual("0x1400010", snapshot["last_ewmh_active_window_id"]) + self.assertFalse(snapshot["last_ewmh_xterm_match"]) self.assertFalse(snapshot["last_non_modifier_key_event_present"]) def test_script_manager_snapshot_reports_cache_counts(self) -> None: diff --git a/tests/test_web_event_router_contract.py b/tests/test_web_event_router_contract.py index 7e47efd..8ef821c 100644 --- a/tests/test_web_event_router_contract.py +++ b/tests/test_web_event_router_contract.py @@ -177,5 +177,137 @@ class CheckedStateRoutingContractTests(unittest.TestCase): ) +class ExpandedStateRoutingContractTests(unittest.TestCase): + def test_shared_web_owns_expanded_state_without_default_fallback(self): + for toolkit_script in (chromium_script, gecko_script): + with self.subTest(toolkit=toolkit_script.__name__): + test_script = toolkit_script.Script.__new__(toolkit_script.Script) + test_script._eventRouter = WebEventRouter() + event = object() + with ( + mock.patch.object( + web.Script, + "onExpandedChanged", + return_value=True, + ) as web_handler, + mock.patch.object( + default.Script, + "onExpandedChanged", + ) as default_handler, + mock.patch( + "cthulhu.scripts.web.event_router.debug.printMessage" + ) as print_message, + ): + toolkit_script.Script.onExpandedChanged(test_script, event) + + web_handler.assert_called_once_with(test_script, event) + default_handler.assert_not_called() + messages = [call.args[1] for call in print_message.call_args_list] + self.assertIn( + "WEB EVENT ROUTER: SHARED_WEB -> HANDLED", + messages, + ) + + def test_unhandled_expanded_state_reaches_default_once(self): + for toolkit_script in (chromium_script, gecko_script): + with self.subTest(toolkit=toolkit_script.__name__): + test_script = toolkit_script.Script.__new__(toolkit_script.Script) + test_script._eventRouter = WebEventRouter() + event = object() + with ( + mock.patch.object( + web.Script, + "onExpandedChanged", + return_value=False, + ) as web_handler, + mock.patch.object( + default.Script, + "onExpandedChanged", + ) as default_handler, + mock.patch( + "cthulhu.scripts.web.event_router.debug.printMessage" + ) as print_message, + ): + toolkit_script.Script.onExpandedChanged(test_script, event) + + web_handler.assert_called_once_with(test_script, event) + default_handler.assert_called_once_with(test_script, event) + messages = [call.args[1] for call in print_message.call_args_list] + self.assertIn( + "WEB EVENT ROUTER: DEFAULT -> HANDLED", + messages, + ) + + +class TableReorderRoutingContractTests(unittest.TestCase): + def test_shared_web_owns_table_reorder_without_default_fallback(self): + for toolkit_script in (chromium_script, gecko_script): + for handler_name in ("onColumnReordered", "onRowReordered"): + with self.subTest( + toolkit=toolkit_script.__name__, + handler=handler_name, + ): + test_script = toolkit_script.Script.__new__(toolkit_script.Script) + test_script._eventRouter = WebEventRouter() + event = object() + with ( + mock.patch.object( + web.Script, + handler_name, + return_value=True, + ) as web_handler, + mock.patch.object( + default.Script, + handler_name, + ) as default_handler, + mock.patch( + "cthulhu.scripts.web.event_router.debug.printMessage" + ) as print_message, + ): + getattr(toolkit_script.Script, handler_name)(test_script, event) + + web_handler.assert_called_once_with(test_script, event) + default_handler.assert_not_called() + messages = [call.args[1] for call in print_message.call_args_list] + self.assertIn( + "WEB EVENT ROUTER: SHARED_WEB -> HANDLED", + messages, + ) + + def test_unhandled_table_reorder_reaches_default_once(self): + for toolkit_script in (chromium_script, gecko_script): + for handler_name in ("onColumnReordered", "onRowReordered"): + with self.subTest( + toolkit=toolkit_script.__name__, + handler=handler_name, + ): + test_script = toolkit_script.Script.__new__(toolkit_script.Script) + test_script._eventRouter = WebEventRouter() + event = object() + with ( + mock.patch.object( + web.Script, + handler_name, + return_value=False, + ) as web_handler, + mock.patch.object( + default.Script, + handler_name, + ) as default_handler, + mock.patch( + "cthulhu.scripts.web.event_router.debug.printMessage" + ) as print_message, + ): + getattr(toolkit_script.Script, handler_name)(test_script, event) + + web_handler.assert_called_once_with(test_script, event) + default_handler.assert_called_once_with(test_script, event) + messages = [call.args[1] for call in print_message.call_args_list] + self.assertIn( + "WEB EVENT ROUTER: DEFAULT -> HANDLED", + messages, + ) + + if __name__ == "__main__": unittest.main()