storm/cthulhu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'testing' into wine-access

Browse Source
This commit is contained in:
Storm Dragon
2026-07-28 23:34:18 -04:00
parent edd606a331 28984449e0
commit 8b90a020c4
9 changed files with 747 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cthulhu/event_manager.py
+58
View File
@@ -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()
src/cthulhu/input_event_manager.py
+146 -37
View File
@@ -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
src/cthulhu/scripts/toolkits/Chromium/script.py
+39 -15
View File
@@ -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."""
src/cthulhu/scripts/toolkits/Gecko/script.py
+39 -15
View File
@@ -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."""
tests/test_event_manager_xterm_handoff_regressions.py
+129
View File
@@ -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()
tests/test_input_event_manager_key_watcher_regressions.py
+2
View File
@@ -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,
[
tests/test_input_event_manager_x11_focus_regressions.py
+194 -9
View File
@@ -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"))
tests/test_runtime_state_snapshot.py
+8
View File
@@ -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:
tests/test_web_event_router_contract.py
+132
View File
@@ -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()