Merged changes from testing.

This commit is contained in:
Storm Dragon
2026-07-19 15:01:58 -04:00
11 changed files with 1827 additions and 36 deletions
+2 -2
View File
@@ -254,6 +254,8 @@ class AXUtilitiesEvent:
reason = TextEventReason.SELECT_ALL
elif mgr.last_event_was_primary_click_or_release():
reason = TextEventReason.MOUSE_PRIMARY_BUTTON
elif mgr.last_event_was_tab_navigation():
reason = TextEventReason.FOCUS_CHANGE
elif AXUtilitiesState.is_editable(obj) or AXUtilitiesRole.is_terminal(obj):
if mgr.last_event_was_backspace():
reason = TextEventReason.BACKSPACE
@@ -273,8 +275,6 @@ class AXUtilitiesEvent:
reason = TextEventReason.UNSPECIFIED_COMMAND
elif mgr.last_event_was_printable_key():
reason = TextEventReason.TYPING
elif mgr.last_event_was_tab_navigation():
reason = TextEventReason.FOCUS_CHANGE
elif AXObject.find_ancestor(obj, AXUtilitiesRole.children_are_presentational):
reason = TextEventReason.UI_UPDATE
return reason
+1 -1
View File
@@ -23,5 +23,5 @@
# Forked from Orca screen reader.
# Cthulhu project: https://git.stormux.org/storm/cthulhu
version = "2026.07.18"
version = "2026.07.19"
codeName = "wine-access"
+152 -4
View File
@@ -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
+5 -7
View File
@@ -60,6 +60,7 @@ from cthulhu.scripts import default
from cthulhu.ax_object import AXObject
from cthulhu.ax_text import AXText
from cthulhu.ax_utilities import AXUtilities
from cthulhu.ax_utilities_event import AXUtilitiesEvent, TextEventReason
from .bookmarks import Bookmarks
from .braille_generator import BrailleGenerator
@@ -2104,6 +2105,7 @@ class Script(default.Script):
debug.printMessage(debug.LEVEL_INFO, msg, True)
return False
reason = AXUtilitiesEvent.get_text_event_reason(event)
obj, offset = self.utilities.getCaretContext(document, False, False)
tokens = ["WEB: Context: ", obj, ", ", offset, "(focus: ", cthulhu_state.locusOfFocus, ")"]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
@@ -2156,14 +2158,10 @@ class Script(default.Script):
self.updateBraille(event.source)
return True
if self.utilities.lastInputEventWasTab():
msg = "WEB: Last input event was Tab."
if reason == TextEventReason.FOCUS_CHANGE:
msg = "WEB: Event ignored: Caret moved due to focus change."
debug.printMessage(debug.LEVEL_INFO, msg, True)
if self.utilities.isDocument(event.source):
msg = "WEB: Event ignored: Caret moved in document due to Tab."
debug.printMessage(debug.LEVEL_INFO, msg, True)
return True
return True
if self.utilities.inFindContainer():
msg = "WEB: Event handled: Presenting find results"
+55 -11
View File
@@ -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):
@@ -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()
@@ -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 = """
+140 -2
View File
@@ -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()
+32
View File
@@ -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):
File diff suppressed because it is too large Load Diff
+82 -5
View File
@@ -24,6 +24,7 @@ if soundGeneratorModule is not None and not hasattr(soundGeneratorModule, "Sound
soundGeneratorModule.SoundGenerator = _StubSoundGenerator
from cthulhu import messages
from cthulhu.ax_utilities_event import AXUtilitiesEvent, TextEventReason
from cthulhu.scripts.web import script as web_script
from cthulhu.scripts.web import script_utilities as web_script_utilities
from cthulhu.scripts.toolkits.Gecko import script_utilities as gecko_script_utilities
@@ -308,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):
@@ -336,6 +336,7 @@ class WebFocusSpeechInterruptionRegressionTests(unittest.TestCase):
class WebCaretMovedRegressionTests(unittest.TestCase):
def _make_script(self):
testScript = web_script.Script.__new__(web_script.Script)
testScript._inFocusMode = False
testScript._lastCommandWasCaretNav = False
testScript._lastCommandWasStructNav = False
testScript._lastCommandWasMouseButton = False
@@ -351,15 +352,87 @@ class WebCaretMovedRegressionTests(unittest.TestCase):
def test_matching_char_nav_caret_event_is_consumed_without_duplicate_presentation(self):
testScript = self._make_script()
source = "source"
event = mock.Mock(source=source, detail1=4)
event = mock.Mock(type="object:text-caret-moved", source=source, detail1=4)
with mock.patch.object(web_script.cthulhu, "setLocusOfFocus") as setLocusOfFocus:
with (
mock.patch.object(
AXUtilitiesEvent,
"get_text_event_reason",
return_value=TextEventReason.NAVIGATION_BY_CHARACTER,
),
mock.patch.object(web_script.cthulhu, "setLocusOfFocus") as setLocusOfFocus,
):
result = web_script.Script.onCaretMoved(testScript, event)
self.assertTrue(result)
testScript.utilities.setCaretContext.assert_not_called()
setLocusOfFocus.assert_not_called()
def test_focus_change_caret_event_does_not_replace_real_focus_event(self):
testScript = self._make_script()
source = object()
event = mock.Mock(
type="object:text-caret-moved",
source=source,
detail1=0,
)
testScript.utilities.getCaretContext.return_value = ("old-focus", 0)
testScript.utilities.lastInputEventWasCharNav.return_value = False
with (
mock.patch.object(
AXUtilitiesEvent,
"get_text_event_reason",
return_value=TextEventReason.FOCUS_CHANGE,
) as getReason,
mock.patch.object(web_script.cthulhu, "setLocusOfFocus") as setLocusOfFocus,
):
result = web_script.Script.onCaretMoved(testScript, event)
self.assertTrue(result)
getReason.assert_called_once_with(event)
testScript.utilities.setCaretContext.assert_not_called()
setLocusOfFocus.assert_not_called()
class TextEventReasonRegressionTests(unittest.TestCase):
def test_tab_is_classified_as_focus_change_before_editable_state(self):
oldFocus = object()
source = object()
event = mock.Mock(type="object:text-caret-moved", source=source)
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
with (
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",
return_value=True,
) as isEditable,
):
getFocusManager.return_value.get_active_mode_and_object_of_interest.return_value = (
None,
oldFocus,
)
reason = AXUtilitiesEvent._get_caret_moved_event_reason(event)
self.assertEqual(reason, TextEventReason.FOCUS_CHANGE)
isEditable.assert_not_called()
class WebDescriptionChangeRegressionTests(unittest.TestCase):
def _make_script(self):
@@ -419,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):
@@ -512,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()
@@ -610,7 +683,11 @@ class WebSelectionRegressionTests(unittest.TestCase):
testScript = self._make_script()
document = object()
section = object()
event = mock.Mock(source=section, detail1=-1)
event = mock.Mock(
type="object:text-caret-moved",
source=section,
detail1=-1,
)
manager = mock.Mock()
manager.last_event_was_forward_caret_selection.return_value = True
manager.last_event_was_backward_caret_selection.return_value = False