From 323d0a3975f5311177edd225d97e93a6c5becf22 Mon Sep 17 00:00:00 2001 From: Storm Dragon Date: Mon, 6 Jul 2026 18:22:19 -0400 Subject: [PATCH] Some web work. Firefox still being weird. --- src/cthulhu/cthulhu.py | 26 +++ src/cthulhu/dbus_service.py | 11 + src/cthulhu/debug.py | 16 +- src/cthulhu/event_manager.py | 24 ++ src/cthulhu/input_event_manager.py | 15 ++ src/cthulhu/plugin_system_manager.py | 22 ++ src/cthulhu/script_manager.py | 25 +++ src/cthulhu/scripts/web/script.py | 19 ++ src/cthulhu/scripts/web/script_utilities.py | 47 ++++ tests/test_runtime_state_snapshot.py | 148 +++++++++++++ tests/test_web_input_regressions.py | 232 ++++++++++++++++++++ 11 files changed, 584 insertions(+), 1 deletion(-) create mode 100644 tests/test_runtime_state_snapshot.py diff --git a/src/cthulhu/cthulhu.py b/src/cthulhu/cthulhu.py index 7fe541d..9d06531 100644 --- a/src/cthulhu/cthulhu.py +++ b/src/cthulhu/cthulhu.py @@ -1042,6 +1042,32 @@ class Cthulhu(GObject.Object): def getLogger(self) -> logger.Logger: # New getter for the logger return self.logger + def get_debug_snapshot(self) -> Dict[str, Dict[str, object]]: + """Returns read-only runtime counters useful for diagnostics.""" + + speechServer = speech.getSpeechServer() + snapshot: Dict[str, Dict[str, object]] = { + "cthulhu_state": { + "active_window_present": cthulhu_state.activeWindow is not None, + "locus_of_focus_present": cthulhu_state.locusOfFocus is not None, + "active_script_present": cthulhu_state.activeScript is not None, + "pending_self_hosted_focus_present": cthulhu_state.pendingSelfHostedFocus is not None, + "device_present": cthulhu_state.device is not None, + "pause_atspi_churn": cthulhu_state.pauseAtspiChurn, + "prioritized_desktop_context_token_present": + cthulhu_state.prioritizedDesktopContextToken is not None, + }, + "speech": { + "speech_server_present": speechServer is not None, + "speech_server": speechServer.__class__.__name__ if speechServer else "", + }, + "event_manager": self.eventManager.get_debug_snapshot(), + "script_manager": self.scriptManager.get_debug_snapshot(), + "input_event_manager": input_event_manager.get_manager().get_debug_snapshot(), + "plugin_system_manager": self.pluginSystemManager.get_debug_snapshot(), + } + return snapshot + def addKeyGrab(self, binding: Any) -> List[int]: return addKeyGrab(binding) diff --git a/src/cthulhu/dbus_service.py b/src/cthulhu/dbus_service.py index 01c771d..1a38bf8 100644 --- a/src/cthulhu/dbus_service.py +++ b/src/cthulhu/dbus_service.py @@ -698,6 +698,17 @@ class CthulhuDBusServiceInterface(Publishable): debug.print_message(debug.LEVEL_INFO, msg, True) return result + def GetRuntimeStateSnapshot(self) -> str: # pylint: disable=invalid-name + """Returns read-only runtime state counters for diagnostics.""" + + msg = "DBUS SERVICE: GetRuntimeStateSnapshot called." + debug.print_message(debug.LEVEL_INFO, msg, True) + + from . import cthulhu # pylint: disable=import-outside-toplevel + + snapshot = cthulhu.cthulhuApp.get_debug_snapshot() + return debug.format_runtime_state_snapshot(snapshot) + def Quit(self) -> bool: # pylint: disable=invalid-name """Quits Cthulhu.""" diff --git a/src/cthulhu/debug.py b/src/cthulhu/debug.py index 14c9df2..7c689df 100644 --- a/src/cthulhu/debug.py +++ b/src/cthulhu/debug.py @@ -45,7 +45,7 @@ import subprocess import sys import time import types -from typing import Any, Generator, Optional, TextIO, Pattern, TYPE_CHECKING +from typing import Any, Generator, Mapping, Optional, TextIO, Pattern, TYPE_CHECKING from datetime import datetime import gi @@ -291,6 +291,20 @@ def print_log_tokens(level: int, prefix: str, tokens: list[Any], reason: Optiona text = format_log_message(prefix, _format_tokens(tokens), reason) printMessage(level, text, timestamp, stack) +def format_runtime_state_snapshot(snapshot: Mapping[str, Any]) -> str: + """Formats read-only runtime state counters for debug capture.""" + + lines = ["CTHULHU RUNTIME STATE SNAPSHOT"] + for section, values in snapshot.items(): + lines.append(f"[{section}]") + if not isinstance(values, Mapping): + lines.append(f"value: {values}") + continue + for key in sorted(values): + lines.append(f"{key}: {values[key]}") + + return "\n".join(lines) + def reset_startup_timing() -> None: """Reset the startup timing baseline.""" diff --git a/src/cthulhu/event_manager.py b/src/cthulhu/event_manager.py index 4c1b636..d131ce2 100644 --- a/src/cthulhu/event_manager.py +++ b/src/cthulhu/event_manager.py @@ -194,6 +194,30 @@ class EventManager: self._deactivateKeyHandling() debug.printMessage(debug.LEVEL_INFO, 'EVENT MANAGER: Deactivated', True) + def get_debug_snapshot(self) -> Dict[str, object]: + """Returns read-only counters useful for long-uptime diagnostics.""" + + return { + "active": self._active, + "async_mode": self._asyncMode, + "event_queue_size": self._eventQueue.qsize(), + "prioritized_event_present": self._prioritizedEvent is not None, + "idle_source_id": self._gidleId, + "prioritized_idle_source_id": self._prioritizedIdleId, + "events_suspended": self._eventsSuspended, + "suspendable_event_types": len(self._suspendableEvents), + "events_triggering_suspension": len(self._eventsTriggeringSuspension), + "ignored_event_types": len(self._ignoredEvents), + "script_listener_types": len(self._scriptListenerCounts), + "parents_of_defunct_descendants": len(self._parentsOfDefunctDescendants), + "cmdline_cache_size": len(self._cmdlineCache), + "relevance_burst_history_size": len(self._relevanceBurstHistory), + "churn_suppressed": self._churnSuppressed, + "prioritized_context_token_present": self._prioritizedContextToken is not None, + "key_handling_active": self._keyHandlingActive, + "input_event_manager_present": self._inputEventManager is not None, + } + def set_compositor_state_adapter(self, adapter: Any) -> None: """Stores the compositor state adapter used for startup wiring.""" diff --git a/src/cthulhu/input_event_manager.py b/src/cthulhu/input_event_manager.py index 8fe766c..4c78470 100644 --- a/src/cthulhu/input_event_manager.py +++ b/src/cthulhu/input_event_manager.py @@ -109,6 +109,21 @@ class InputEventManager: debug.print_message(debug.LEVEL_INFO, msg, True) self._device = None + def get_debug_snapshot(self) -> Dict[str, object]: + """Returns read-only counters useful for long-uptime diagnostics.""" + + return { + "device_active": self._device is not None, + "pointer_watcher_active": self._pointer_moved_id != 0, + "mapped_keycodes": len(self._mapped_keycodes), + "mapped_keysyms": len(self._mapped_keysyms), + "grabbed_bindings": len(self._grabbed_bindings), + "paused": self._paused, + "suspended_xterm_script_present": self._scriptWithSuspendedGrabsForXterm is not None, + "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, + } + def enable_pointer_monitoring(self) -> bool: """Enables pointer monitoring on the current device, if possible.""" diff --git a/src/cthulhu/plugin_system_manager.py b/src/cthulhu/plugin_system_manager.py index 5d58366..55d5488 100644 --- a/src/cthulhu/plugin_system_manager.py +++ b/src/cthulhu/plugin_system_manager.py @@ -1102,3 +1102,25 @@ class PluginSystemManager: for pluginInfo in self.plugins: if ForceAllPlugins or pluginInfo.loaded: self.unloadPlugin(pluginInfo) + + def get_debug_snapshot(self) -> Dict[str, object]: + """Returns read-only counters useful for long-uptime diagnostics.""" + + loadedPlugins = [info for info in self._plugins.values() if info.loaded] + try: + pluggyRegisteredPlugins = len(self.plugin_manager.get_plugins()) + except Exception: + pluggyRegisteredPlugins = -1 + + return { + "scanned_plugins": len(self._plugins), + "active_plugins": len(self._active_plugins), + "loaded_plugins": len(loadedPlugins), + "builtin_plugins": sum(1 for info in self._plugins.values() if info.builtin), + "hidden_plugins": sum(1 for info in self._plugins.values() if info.hidden), + "plugin_keybinding_contexts": len(self._plugin_keybindings), + "plugin_keybindings": sum(len(bindings) for bindings in self._plugin_keybindings.values()), + "global_keybindings": len(self._global_bindings), + "last_active_script_present": self._last_active_script is not None, + "pluggy_registered_plugins": pluggyRegisteredPlugins, + } diff --git a/src/cthulhu/script_manager.py b/src/cthulhu/script_manager.py index da29416..b393745 100644 --- a/src/cthulhu/script_manager.py +++ b/src/cthulhu/script_manager.py @@ -396,6 +396,31 @@ class ScriptManager: reason ) + def get_debug_snapshot(self) -> Dict[str, object]: + """Returns read-only counters useful for long-uptime diagnostics.""" + + activeScript = cthulhu_state.activeScript + activeApp = getattr(activeScript, "app", None) + activeAppName = "" + if activeApp is not None: + try: + activeAppName = AXObject.get_name(activeApp) or "" + except Exception: + activeAppName = "[unavailable]" + + return { + "active": self._active, + "active_script": activeScript.__class__.__name__ if activeScript else "", + "active_script_app": activeAppName, + "default_script_present": self._defaultScript is not None, + "app_scripts": len(self.appScripts), + "toolkit_script_apps": len(self.toolkitScripts), + "toolkit_scripts": sum(len(scripts) for scripts in self.toolkitScripts.values()), + "custom_script_apps": len(self.customScripts), + "custom_scripts": sum(len(scripts) for scripts in self.customScripts.values()), + "sleep_mode_scripts": len(self._sleepModeScripts), + } + def _get_script_for_app_replicant(self, app: Atspi.Accessible) -> Optional[Script]: if not self._active: return None diff --git a/src/cthulhu/scripts/web/script.py b/src/cthulhu/scripts/web/script.py index 14d47af..c911f03 100644 --- a/src/cthulhu/scripts/web/script.py +++ b/src/cthulhu/scripts/web/script.py @@ -2116,6 +2116,11 @@ class Script(default.Script): else: self._clearSyntheticWebSelection() + if self.utilities.lastInputEventWasCharNav() and (event.source, event.detail1) == (obj, offset): + msg = "WEB: Event ignored: Character navigation already presented this context" + debug.printMessage(debug.LEVEL_INFO, msg, True) + return True + if self._lastCommandWasCaretNav: msg = "WEB: Event ignored: Last command was caret nav" debug.printMessage(debug.LEVEL_INFO, msg, True) @@ -2624,6 +2629,20 @@ class Script(default.Script): self.utilities.setCaretContext(event.source, 0) obj, offset = event.source, 0 + if self._lastCommandWasCaretNav \ + and obj is not event.source \ + and AXUtilities.is_focusable(event.source) \ + and AXUtilities.is_focused(event.source) \ + and self.utilities.inDocumentContent(event.source) \ + and self.utilities.isLink(event.source) \ + and obj \ + and AXUtilities.is_section(obj): + msg = "WEB: Event handled: Replacing stale section context with focused link" + debug.printMessage(debug.LEVEL_INFO, msg, True) + cthulhu.setLocusOfFocus(event, event.source, False) + self.utilities.setCaretContext(event.source, 0) + return True + if self._lastCommandWasCaretNav: msg = "WEB: Event ignored: Last command was caret nav" debug.printMessage(debug.LEVEL_INFO, msg, True) diff --git a/src/cthulhu/scripts/web/script_utilities.py b/src/cthulhu/scripts/web/script_utilities.py index 05012b6..f15c5fc 100644 --- a/src/cthulhu/scripts/web/script_utilities.py +++ b/src/cthulhu/scripts/web/script_utilities.py @@ -230,6 +230,7 @@ class Utilities(script_utilities.Utilities): self._shouldInferLabelFor = {} self._treatAsTextObject = {} self._treatAsDiv = {} + self.clearContentCache() self._paths = {} self._contextPathsRolesAndNames = {} self._canHaveCaretContextDecision = {} @@ -1793,6 +1794,21 @@ class Utilities(script_utilities.Utilities): self._canHaveCaretContextDecision = {} return rv + def _trimContentsBeforeEmbeddedObjectBoundary(self, contents, obj, offset): + if not contents: + return contents + + firstObj, firstStart, firstEnd, firstString = contents[0] + if firstObj != obj or not (firstStart < offset < firstEnd): + return contents + + char = AXText.get_character_at_offset(obj, firstStart)[0] + if char != self.EMBEDDED_OBJECT_CHARACTER: + return contents + + string = AXText.get_substring(obj, offset, firstEnd) + return [[firstObj, offset, firstEnd, string]] + contents[1:] + def _getLineContentsAtOffset(self, obj, offset, layoutMode=None, useCache=True): startTime = time.time() if not obj: @@ -1881,6 +1897,13 @@ class Utilities(script_utilities.Utilities): self._debugContentsInfo(obj, offset, objects, "Line (not layout mode)") return objects + if self.isLink(obj) and objects and objects[0][0] == obj and objects[0][1] > 0: + if useCache: + self._currentLineContents = objects + + self._debugContentsInfo(obj, offset, objects, "Line (wrapped link)") + return objects + firstObj, firstStart, firstEnd, firstString = objects[0] if (extents[2] == 0 and extents[3] == 0) or self.isMath(firstObj): extents = self.getExtents(firstObj, firstStart, firstEnd) @@ -2027,6 +2050,13 @@ class Utilities(script_utilities.Utilities): tokens = ["WEB: Previous context is: ", obj, ", ", offset] debug.printTokens(debug.LEVEL_INFO, tokens, True) + if obj == firstObj and 0 <= offset < firstOffset and self.isLink(obj): + contents = self.getLineContentsAtOffset(obj, offset, layoutMode=False, useCache=False) + if contents and contents != line: + msg = "WEB: Using same-link non-layout previous line contents." + debug.printMessage(debug.LEVEL_INFO, msg, True) + return contents + contents = self.getLineContentsAtOffset(obj, offset, layoutMode, useCache) if not contents: tokens = ["WEB: Could not get line contents for ", obj, ", ", offset] @@ -2092,6 +2122,23 @@ class Utilities(script_utilities.Utilities): tokens = ["WEB: Next context is: ", obj, ", ", offset] debug.printTokens(debug.LEVEL_INFO, tokens, True) + if obj == lastObj and offset > lastOffset and self.isLink(obj): + contents = self.getLineContentsAtOffset(obj, offset, layoutMode=False, useCache=False) + if contents and contents != line: + msg = "WEB: Using same-link non-layout next line contents." + debug.printMessage(debug.LEVEL_INFO, msg, True) + return contents + + if self.isLink(lastObj): + start, end = self.getHyperlinkRange(lastObj) + if obj == AXObject.get_parent(lastObj) and end >= 0 and offset >= end: + contents = self.getLineContentsAtOffset(obj, offset, layoutMode=False, useCache=False) + contents = self._trimContentsBeforeEmbeddedObjectBoundary(contents, obj, offset) + if contents and contents != line: + msg = "WEB: Using non-layout line after split link." + debug.printMessage(debug.LEVEL_INFO, msg, True) + return contents + contents = self.getLineContentsAtOffset(obj, offset, layoutMode, useCache) if line == contents: obj, offset = self.nextContext(obj, offset, True) diff --git a/tests/test_runtime_state_snapshot.py b/tests/test_runtime_state_snapshot.py new file mode 100644 index 0000000..75d5003 --- /dev/null +++ b/tests/test_runtime_state_snapshot.py @@ -0,0 +1,148 @@ +import queue +import sys +import types +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from cthulhu import cthulhu_state +from cthulhu import debug + +stubCthulhu = types.ModuleType("cthulhu.cthulhu") +stubCthulhu.cthulhuApp = mock.Mock() +sys.modules.setdefault("cthulhu.cthulhu", stubCthulhu) + +from cthulhu import event_manager +from cthulhu.input_event_manager import InputEventManager +from cthulhu.plugin_system_manager import PluginSystemManager +from cthulhu.script_manager import ScriptManager + + +class RuntimeStateSnapshotTests(unittest.TestCase): + def test_formatter_sorts_keys_and_groups_sections(self) -> None: + result = debug.format_runtime_state_snapshot({ + "event_manager": {"z_count": 2, "a_count": 1}, + "speech": {"speech_server_present": False}, + }) + + self.assertEqual( + result, + "\n".join([ + "CTHULHU RUNTIME STATE SNAPSHOT", + "[event_manager]", + "a_count: 1", + "z_count: 2", + "[speech]", + "speech_server_present: False", + ]), + ) + + def test_event_manager_snapshot_reports_core_counters(self) -> None: + manager = event_manager.EventManager.__new__(event_manager.EventManager) + manager._active = True + manager._asyncMode = True + manager._eventQueue = queue.Queue() + manager._eventQueue.put(object()) + manager._prioritizedEvent = object() + manager._gidleId = 11 + manager._prioritizedIdleId = 12 + manager._eventsSuspended = True + manager._suspendableEvents = ["object:children-changed:add"] + manager._eventsTriggeringSuspension = [object(), object()] + manager._ignoredEvents = ["object:bounds-changed"] + manager._scriptListenerCounts = {"focus:": 1} + manager._parentsOfDefunctDescendants = [object()] + manager._cmdlineCache = {123: "app"} + manager._relevanceBurstHistory = {("a", "b", "c"): 1.0} + manager._churnSuppressed = True + manager._prioritizedContextToken = "token" + manager._keyHandlingActive = True + manager._inputEventManager = object() + + snapshot = manager.get_debug_snapshot() + + self.assertEqual(1, snapshot["event_queue_size"]) + self.assertTrue(snapshot["prioritized_event_present"]) + self.assertEqual(11, snapshot["idle_source_id"]) + self.assertEqual(2, snapshot["events_triggering_suspension"]) + self.assertEqual(1, snapshot["cmdline_cache_size"]) + self.assertTrue(snapshot["prioritized_context_token_present"]) + + def test_input_event_manager_snapshot_reports_grab_state(self) -> None: + manager = InputEventManager() + manager._device = object() + manager._pointer_moved_id = 42 + manager._mapped_keycodes = [1, 2] + manager._mapped_keysyms = [3] + manager._grabbed_bindings = {10: object(), 11: object()} + manager._paused = True + manager._scriptWithSuspendedGrabsForXterm = object() + manager._last_input_event = object() + manager._last_non_modifier_key_event = None + + snapshot = manager.get_debug_snapshot() + + self.assertTrue(snapshot["device_active"]) + self.assertTrue(snapshot["pointer_watcher_active"]) + self.assertEqual(2, snapshot["mapped_keycodes"]) + self.assertEqual(1, snapshot["mapped_keysyms"]) + self.assertEqual(2, snapshot["grabbed_bindings"]) + self.assertTrue(snapshot["suspended_xterm_script_present"]) + self.assertFalse(snapshot["last_non_modifier_key_event_present"]) + + def test_script_manager_snapshot_reports_cache_counts(self) -> None: + activeScript = SimpleNamespace(app=None) + oldActiveScript = cthulhu_state.activeScript + cthulhu_state.activeScript = activeScript + try: + manager = ScriptManager.__new__(ScriptManager) + manager._active = True + manager._defaultScript = object() + manager.appScripts = {"app": object()} + manager.toolkitScripts = {"app": {"toolkit": object()}} + manager.customScripts = {"app": {"role": object(), "other": object()}} + manager._sleepModeScripts = {"app": object()} + + snapshot = manager.get_debug_snapshot() + finally: + cthulhu_state.activeScript = oldActiveScript + + self.assertTrue(snapshot["active"]) + self.assertEqual("SimpleNamespace", snapshot["active_script"]) + self.assertTrue(snapshot["default_script_present"]) + self.assertEqual(1, snapshot["app_scripts"]) + self.assertEqual(1, snapshot["toolkit_scripts"]) + self.assertEqual(2, snapshot["custom_scripts"]) + self.assertEqual(1, snapshot["sleep_mode_scripts"]) + + def test_plugin_system_manager_snapshot_reports_plugin_counts(self) -> None: + manager = PluginSystemManager.__new__(PluginSystemManager) + manager._plugins = { + "Loaded": SimpleNamespace(loaded=True, builtin=False, hidden=False), + "Builtin": SimpleNamespace(loaded=True, builtin=True, hidden=False), + "Hidden": SimpleNamespace(loaded=False, builtin=False, hidden=True), + } + manager._active_plugins = ["Loaded", "Builtin"] + manager._plugin_keybindings = {"Loaded": [object(), object()]} + manager._global_bindings = [object()] + manager._last_active_script = object() + manager.plugin_manager = mock.Mock() + manager.plugin_manager.get_plugins.return_value = [object(), object()] + + snapshot = manager.get_debug_snapshot() + + self.assertEqual(3, snapshot["scanned_plugins"]) + self.assertEqual(2, snapshot["active_plugins"]) + self.assertEqual(2, snapshot["loaded_plugins"]) + self.assertEqual(1, snapshot["builtin_plugins"]) + self.assertEqual(1, snapshot["hidden_plugins"]) + self.assertEqual(2, snapshot["plugin_keybindings"]) + self.assertEqual(1, snapshot["global_keybindings"]) + self.assertEqual(2, snapshot["pluggy_registered_plugins"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_web_input_regressions.py b/tests/test_web_input_regressions.py index e46bd5c..7ad2b16 100644 --- a/tests/test_web_input_regressions.py +++ b/tests/test_web_input_regressions.py @@ -329,6 +329,34 @@ class WebFocusSpeechInterruptionRegressionTests(unittest.TestCase): speak.assert_called_once_with(["focus speech"], interrupt=False) +class WebCaretMovedRegressionTests(unittest.TestCase): + def _make_script(self): + testScript = web_script.Script.__new__(web_script.Script) + testScript._lastCommandWasCaretNav = False + testScript._lastCommandWasStructNav = False + testScript._lastCommandWasMouseButton = False + testScript._clearSyntheticWebSelection = mock.Mock() + testScript.utilities = mock.Mock() + testScript.utilities.isZombie.return_value = False + testScript.utilities.getTopLevelDocumentForObject.return_value = "document" + testScript.utilities.getCaretContext.return_value = ("source", 4) + testScript.utilities.lastInputEventWasCaretNavWithSelection.return_value = False + testScript.utilities.lastInputEventWasCharNav.return_value = True + return testScript + + 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) + + with 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() + + class WebDescriptionChangeRegressionTests(unittest.TestCase): def _make_script(self): testScript = web_script.Script.__new__(web_script.Script) @@ -452,6 +480,35 @@ class WebDynamicContentRecoveryRegressionTests(unittest.TestCase): setLocusOfFocus.assert_called_once_with(event, source, False) testScript.utilities.setCaretContext.assert_called_once_with(source, 0) + def test_focused_link_replaces_stale_section_context_after_caret_nav(self): + testScript = self._make_dynamic_script() + document = object() + section = object() + link = object() + event = mock.Mock(detail1=1, source=link) + testScript._lastCommandWasCaretNav = True + testScript.utilities.getDocumentForObject.return_value = document + testScript.utilities.isWebAppDescendant.return_value = False + testScript.utilities.handleEventFromContextReplicant.return_value = False + testScript.utilities.getCaretContext.return_value = (section, 4) + testScript.utilities.isLink.side_effect = lambda obj: obj is link + + with ( + mock.patch.object(web_script.cthulhu_state, "locusOfFocus", section), + mock.patch.object(web_script.AXUtilities, "is_editable", return_value=False), + mock.patch.object(web_script.AXUtilities, "is_dialog_or_alert", return_value=False), + mock.patch.object(web_script.AXUtilities, "is_focusable", return_value=True), + mock.patch.object(web_script.AXUtilities, "is_focused", return_value=True), + mock.patch.object(web_script.AXUtilities, "is_section", side_effect=lambda obj: obj is section), + mock.patch.object(web_script.cthulhu, "setLocusOfFocus") as setLocusOfFocus, + ): + result = web_script.Script.onFocusedChanged(testScript, event) + + self.assertTrue(result) + 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() @@ -635,3 +692,178 @@ class WebSelectionRegressionTests(unittest.TestCase): result = web_script_utilities.Utilities.allSelectedText(utilities, obj) self.assertEqual(result, ["Dark", 0, 4]) + + +class WebContentCacheRegressionTests(unittest.TestCase): + def test_clear_cached_objects_clears_current_line_contents(self): + utilities = web_script_utilities.Utilities.__new__(web_script_utilities.Utilities) + utilities._currentObjectContents = ["object"] + utilities._currentSentenceContents = ["sentence"] + utilities._currentLineContents = ["line"] + utilities._currentWordContents = ["word"] + utilities._currentCharacterContents = ["character"] + utilities._currentTextAttrs = {"attrs": True} + utilities._cleanupContexts = mock.Mock() + + web_script_utilities.Utilities.clearCachedObjects(utilities) + + self.assertIsNone(utilities._currentObjectContents) + self.assertIsNone(utilities._currentSentenceContents) + self.assertIsNone(utilities._currentLineContents) + self.assertIsNone(utilities._currentWordContents) + self.assertIsNone(utilities._currentCharacterContents) + self.assertEqual(utilities._currentTextAttrs, {}) + utilities._cleanupContexts.assert_called_once_with() + + +class WebSplitLinkLineNavigationRegressionTests(unittest.TestCase): + def _make_utilities(self, link): + utilities = web_script_utilities.Utilities.__new__(web_script_utilities.Utilities) + utilities.isZombie = mock.Mock(return_value=False) + utilities.getMathAncestor = mock.Mock(return_value=None) + utilities.isLink = mock.Mock(side_effect=lambda obj: obj is link) + utilities.clearCachedObjects = mock.Mock() + return utilities + + def test_next_line_uses_non_layout_contents_for_split_link(self): + link = object() + utilities = self._make_utilities(link) + currentLine = [[link, 0, 20, "https://stormux.org/"]] + nextLine = [[link, 20, 29, "downloads"]] + utilities.nextContext = mock.Mock(return_value=(link, 20)) + utilities.getLineContentsAtOffset = mock.Mock(side_effect=[currentLine, nextLine]) + + manager = mock.Mock() + manager.get_speak_blank_lines.return_value = False + with mock.patch.object( + web_script_utilities.speech_and_verbosity_manager, + "getManager", + return_value=manager, + ): + result = web_script_utilities.Utilities.getNextLineContents(utilities, link, 0) + + self.assertEqual(result, nextLine) + utilities.nextContext.assert_called_once_with(link, 19, True) + self.assertEqual( + utilities.getLineContentsAtOffset.call_args_list, + [ + mock.call(link, 0, None, True), + mock.call(link, 20, layoutMode=False, useCache=False), + ], + ) + + def test_previous_line_uses_non_layout_contents_for_split_link(self): + link = object() + utilities = self._make_utilities(link) + previousLine = [[link, 0, 20, "https://stormux.org/"]] + currentLine = [[link, 20, 29, "downloads"]] + utilities.previousContext = mock.Mock(return_value=(link, 19)) + utilities.getLineContentsAtOffset = mock.Mock(side_effect=[currentLine, previousLine]) + + manager = mock.Mock() + manager.get_speak_blank_lines.return_value = False + with mock.patch.object( + web_script_utilities.speech_and_verbosity_manager, + "getManager", + return_value=manager, + ): + result = web_script_utilities.Utilities.getPreviousLineContents(utilities, link, 20) + + self.assertEqual(result, previousLine) + utilities.previousContext.assert_called_once_with(link, 20, True) + self.assertEqual( + utilities.getLineContentsAtOffset.call_args_list, + [ + mock.call(link, 20, None, True), + mock.call(link, 19, layoutMode=False, useCache=False), + ], + ) + + def test_current_line_on_wrapped_link_continuation_does_not_expand_to_parent(self): + link = object() + utilities = self._make_utilities(link) + utilities._currentLineContents = None + utilities._debugContentsInfo = mock.Mock() + utilities.findObjectInContents = mock.Mock(return_value=-1) + utilities.treatAsEndOfLine = mock.Mock(return_value=False) + utilities.getExtents = mock.Mock(return_value=[321, 763, 86, 19]) + utilities.isInlineListDescendant = mock.Mock(return_value=False) + utilities._getContentsForObj = mock.Mock( + return_value=[[link, 20, 29, "downloads"]] + ) + utilities.getDocumentForObject = mock.Mock() + + with ( + mock.patch.object(web_script_utilities.AXObject, "is_dead", return_value=False), + mock.patch.object(web_script_utilities.AXObject, "find_ancestor", return_value=None), + mock.patch.object(web_script_utilities.AXUtilities, "is_tool_bar", return_value=False), + mock.patch.object(web_script_utilities.AXUtilities, "is_menu_bar", return_value=False), + ): + result = web_script_utilities.Utilities._getLineContentsAtOffset( + utilities, + link, + 23, + layoutMode=True, + ) + + self.assertEqual(result, [[link, 20, 29, "downloads"]]) + utilities.getDocumentForObject.assert_not_called() + utilities._debugContentsInfo.assert_called_once_with( + link, + 23, + [[link, 20, 29, "downloads"]], + "Line (wrapped link)", + ) + + def test_next_line_after_split_link_uses_parent_line_after_embedded_marker(self): + link = object() + paragraph = object() + utilities = self._make_utilities(link) + currentLine = [[link, 20, 29, "downloads"]] + paragraphLine = [[paragraph, 95, 191, "[OBJ]. On first boot"]] + expectedLine = [[paragraph, 96, 191, ". On first boot"]] + utilities.nextContext = mock.Mock(return_value=(paragraph, 96)) + utilities.getLineContentsAtOffset = mock.Mock( + side_effect=[currentLine, paragraphLine] + ) + utilities.getHyperlinkRange = mock.Mock(return_value=(95, 96)) + + manager = mock.Mock() + manager.get_speak_blank_lines.return_value = False + with ( + mock.patch.object( + web_script_utilities.speech_and_verbosity_manager, + "getManager", + return_value=manager, + ), + mock.patch.object( + web_script_utilities.AXObject, + "get_parent", + return_value=paragraph, + ), + mock.patch.object( + web_script_utilities.AXText, + "get_character_at_offset", + return_value=(utilities.EMBEDDED_OBJECT_CHARACTER, 95, 96), + ), + mock.patch.object( + web_script_utilities.AXText, + "get_substring", + return_value=". On first boot", + ), + ): + result = web_script_utilities.Utilities.getNextLineContents( + utilities, + link, + 23, + ) + + self.assertEqual(result, expectedLine) + utilities.nextContext.assert_called_once_with(link, 28, True) + self.assertEqual( + utilities.getLineContentsAtOffset.call_args_list, + [ + mock.call(link, 23, None, True), + mock.call(paragraph, 96, layoutMode=False, useCache=False), + ], + )