Merge branch 'atspi-document-selection' into testing

This commit is contained in:
2026-04-09 05:24:07 -04:00
15 changed files with 2310 additions and 93 deletions
+20
View File
@@ -32,6 +32,25 @@ if ! python3 -c "import tomlkit" 2>/dev/null; then
exit 1
fi
getPythonSiteDir() {
python3 -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")'
}
removeExistingLocalInstall() {
local pythonSite packageDir binPath desktopFile manPage
pythonSite="$HOME/.local/lib/python$(getPythonSiteDir)/site-packages"
packageDir="$pythonSite/cthulhu"
binPath="$HOME/.local/bin/cthulhu"
desktopFile="$HOME/.local/share/applications/cthulhu.desktop"
manPage="$HOME/.local/share/man/man1/cthulhu.1"
echo "Removing existing local installation targets..."
rm -f "$binPath"
rm -rf "$packageDir"
rm -f "$desktopFile"
rm -f "$manPage"
}
# Check for optional dependencies
missingOptional=()
if ! python3 -c "import gi" 2>/dev/null; then
@@ -63,6 +82,7 @@ meson compile -C _build
# Install
echo "Installing Cthulhu to ~/.local..."
removeExistingLocalInstall
meson install -C _build
# Update desktop database and icon cache
+51
View File
@@ -0,0 +1,51 @@
# AT-SPI Document Selection Branch Handoff
Branch: `atspi-document-selection`
## Implemented
- Added document-wide AT-SPI selection normalization in
`src/cthulhu/ax_document_selection.py`.
- Reworked text-selection handling in `src/cthulhu/script_utilities.py` to use
the shared adapter where possible and fall back cleanly where not.
- Updated primary callers in the default and web scripts so copy/cut,
selection announcements, and selected-text queries use the shared selection
path.
- Added keyboard-driven clipboard announcement fallback in
`src/cthulhu/scripts/default.py`, including Wayland-specific policy changes.
- Fixed `build-local.sh` so rebuilding refreshes modified installed modules
under `~/.local`.
## Verified
- `python3 -m pytest -q tests/test_document_selection_regressions.py`
- `python3 -m pytest -q tests/test_default_script_clipboard_regressions.py`
- `python3 -m pytest -q tests/test_web_input_regressions.py`
- `HOME=/tmp/cthulhu-test-home python3 -m pytest -q tests`
- `./build-local.sh`
Latest full-suite result before handoff: `95 passed, 1 warning`.
## Still Open
- Browser link-text selection is still not reliable in Chromium/Edge and in
web-backed content like Steam.
- The current branch includes extra diagnostics around clipboard and selection
paths because they were useful during manual debugging.
- The newest attempted web fix is in `src/cthulhu/scripts/web/script.py`, but
the manual repro still fails:
- selecting link text with `Shift+Left` / `Shift+Right`
- hearing no selection announcement
- then having copy behavior depend on fallback logic instead of real
selection state
## Useful Files For Follow-Up
- `src/cthulhu/ax_document_selection.py`
- `src/cthulhu/script_utilities.py`
- `src/cthulhu/scripts/default.py`
- `src/cthulhu/scripts/web/script.py`
- `src/cthulhu/scripts/web/script_utilities.py`
- `tests/test_document_selection_regressions.py`
- `tests/test_default_script_clipboard_regressions.py`
- `tests/test_web_input_regressions.py`
+571
View File
@@ -0,0 +1,571 @@
#!/usr/bin/env python3
#
# Copyright (c) 2026 Stormux
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# Forked from Orca screen reader.
# Cthulhu project: https://git.stormux.org/storm/cthulhu
"""Helpers for document-wide AT-SPI text selections."""
from __future__ import annotations
import functools
from dataclasses import dataclass, field
from typing import Any, Optional, TYPE_CHECKING
import gi
gi.require_version("Atspi", "2.0")
from gi.repository import Atspi
from gi.repository import GLib
from . import debug
from .ax_object import AXObject
from .ax_text import AXText
if TYPE_CHECKING:
from .script_utilities import Utilities
@dataclass
class AXSelectionRange:
start_object: Atspi.Accessible
start_offset: int
end_object: Atspi.Accessible
end_offset: int
start_is_active: bool = False
def anchor_endpoint(self) -> tuple[Any, int]:
if self.start_is_active:
return self.end_object, self.end_offset
return self.start_object, self.start_offset
def focus_endpoint(self) -> tuple[Any, int]:
if self.start_is_active:
return self.start_object, self.start_offset
return self.end_object, self.end_offset
def active_endpoint(self) -> tuple[Any, int]:
if self.start_is_active:
return self.start_object, self.start_offset
return self.end_object, self.end_offset
@dataclass
class AXSelectionSnapshot:
document: Optional[Atspi.Accessible]
source_object: Optional[Atspi.Accessible]
ranges: list[AXSelectionRange] = field(default_factory=list)
anchor_focus_pairs: list[tuple[Any, int, Any, int]] = field(default_factory=list)
entire_document_selected: bool = False
is_authoritative: bool = False
@dataclass
class AXProjectedSelection:
obj: Optional[Atspi.Accessible]
local_ranges: list[tuple[int, int]] = field(default_factory=list)
text: str = ""
start_offset: int = 0
end_offset: int = 0
class AXDocumentSelection:
"""Normalizes and applies document-wide text selections."""
def __init__(self, utilities: "Utilities") -> None:
self._utilities = utilities
def empty_snapshot(
self,
source_object: Optional[Atspi.Accessible] = None,
document: Optional[Atspi.Accessible] = None,
) -> AXSelectionSnapshot:
return AXSelectionSnapshot(document=document, source_object=source_object)
def get_cached_snapshot(self, obj: Optional[Atspi.Accessible]) -> AXSelectionSnapshot:
document = self._utilities.getDocumentForObject(obj)
state = self._utilities._script.pointOfReference.get("selectionState", {})
snapshot = state.get("snapshot")
if snapshot is None:
return self.empty_snapshot(obj, document)
if state.get("document") != document:
return self.empty_snapshot(obj, document)
return snapshot
def refresh_snapshot(self, obj: Optional[Atspi.Accessible]) -> AXSelectionSnapshot:
document = self._utilities.getDocumentForObject(obj)
snapshot = self._fetch_snapshot(document, obj)
state = self._utilities._script.pointOfReference.get("selectionState", {})
if state.get("document") != document:
state = {}
state["document"] = document
state["snapshot"] = snapshot
state.setdefault("textSelections", {})
self._utilities._script.pointOfReference["selectionState"] = state
if snapshot.is_authoritative:
self._utilities._script.pointOfReference["entireDocumentSelected"] = snapshot.entire_document_selected
return snapshot
def get_selected_objects(self, snapshot: AXSelectionSnapshot) -> list[Atspi.Accessible]:
seen = set()
objects = []
for selectionRange in snapshot.ranges:
for obj in self._objects_for_range(selectionRange):
if hash(obj) in seen:
continue
seen.add(hash(obj))
objects.append(obj)
return objects
def get_projected_selection(
self,
obj: Optional[Atspi.Accessible],
snapshot: Optional[AXSelectionSnapshot] = None,
) -> AXProjectedSelection:
if not obj:
return AXProjectedSelection(obj)
snapshot = snapshot or self.get_cached_snapshot(obj)
if not snapshot.is_authoritative:
return AXProjectedSelection(obj)
localRanges = []
for selectionRange in snapshot.ranges:
localRanges.extend(self._project_range_onto_object(selectionRange, obj))
localRanges = self._merge_ranges(localRanges)
fragments = []
for startOffset, endOffset in localRanges:
text = self._utilities.expandEOCs(obj, startOffset, endOffset)
if text:
fragments.append(text)
if localRanges:
startOffset = localRanges[0][0]
endOffset = localRanges[-1][1]
else:
startOffset = endOffset = 0
return AXProjectedSelection(
obj=obj,
local_ranges=localRanges,
text=" ".join(fragments),
start_offset=startOffset,
end_offset=endOffset,
)
def get_all_selected_text(
self,
obj: Optional[Atspi.Accessible],
snapshot: Optional[AXSelectionSnapshot] = None,
) -> tuple[str, int, int]:
if not obj:
return "", 0, 0
snapshot = snapshot or self.get_cached_snapshot(obj)
projected = self.get_projected_selection(obj, snapshot)
if not snapshot.is_authoritative or not snapshot.ranges:
return projected.text, projected.start_offset, projected.end_offset
fragments = []
for selectionRange in snapshot.ranges:
text = self._range_text(selectionRange)
if text:
fragments.append(text)
return " ".join(fragments), projected.start_offset, projected.end_offset
def clear_selection(self, obj: Optional[Atspi.Accessible]) -> bool:
if not obj:
return False
snapshot = self.refresh_snapshot(obj)
if not snapshot.is_authoritative:
return False
projected = self.get_projected_selection(obj, snapshot)
if not projected.local_ranges and not snapshot.entire_document_selected:
return False
return self._set_document_ranges(snapshot.document, [])
def adjust_selection(self, obj: Optional[Atspi.Accessible], offset: int) -> bool:
if not obj:
return False
snapshot = self.refresh_snapshot(obj)
if not snapshot.is_authoritative:
return False
ranges = list(snapshot.ranges)
anchorOffset = AXText.get_caret_offset(obj)
rangeIndex = self._resolve_primary_range_index(ranges, obj, anchorOffset)
if rangeIndex is None:
rangeIndex = len(ranges)
ranges.append(
AXSelectionRange(
start_object=obj,
start_offset=anchorOffset,
end_object=obj,
end_offset=anchorOffset,
start_is_active=False,
)
)
selectionRange = ranges[rangeIndex]
anchorObject, anchorOffset = selectionRange.anchor_endpoint()
updatedRange = self._build_range(anchorObject, anchorOffset, obj, offset)
if updatedRange is None:
return False
ranges[rangeIndex] = updatedRange
return self._set_document_ranges(snapshot.document, ranges)
def _fetch_snapshot(
self,
document: Optional[Atspi.Accessible],
source_object: Optional[Atspi.Accessible],
) -> AXSelectionSnapshot:
if not document:
return self.empty_snapshot(source_object, document)
try:
rawSelections = Atspi.Document.get_text_selections(document)
except (AttributeError, GLib.GError, TypeError) as error:
msg = f"AXDocumentSelection: Exception in get_text_selections: {error}"
debug.printMessage(debug.LEVEL_INFO, msg, True)
return self.empty_snapshot(source_object, document)
ranges = []
for rawSelection in rawSelections or []:
selectionRange = self._normalize_range(document, rawSelection)
if selectionRange is None:
return self.empty_snapshot(source_object, document)
ranges.append(selectionRange)
ranges.sort(key=functools.cmp_to_key(self._compare_ranges))
return AXSelectionSnapshot(
document=document,
source_object=source_object,
ranges=ranges,
anchor_focus_pairs=[
(
selectionRange.anchor_endpoint()[0],
selectionRange.anchor_endpoint()[1],
selectionRange.focus_endpoint()[0],
selectionRange.focus_endpoint()[1],
)
for selectionRange in ranges
],
entire_document_selected=self._is_entire_document_selected(ranges),
is_authoritative=bool(ranges),
)
def _normalize_range(
self,
document: Atspi.Accessible,
rawSelection: Any,
) -> Optional[AXSelectionRange]:
startObject = getattr(rawSelection, "start_object", None)
endObject = getattr(rawSelection, "end_object", None)
startOffset = getattr(rawSelection, "start_offset", 0)
endOffset = getattr(rawSelection, "end_offset", 0)
startIsActive = bool(getattr(rawSelection, "start_is_active", False))
if not startObject or not endObject:
return None
if not AXObject.supports_text(startObject) or not AXObject.supports_text(endObject):
return None
if self._utilities.getDocumentForObject(startObject) != document:
return None
if self._utilities.getDocumentForObject(endObject) != document:
return None
comparison = self._compare_positions(startObject, startOffset, endObject, endOffset)
if comparison is None:
return None
if comparison > 0:
startObject, endObject = endObject, startObject
startOffset, endOffset = endOffset, startOffset
startIsActive = not startIsActive
if startObject == endObject and startOffset > endOffset:
startOffset, endOffset = endOffset, startOffset
startIsActive = not startIsActive
if startObject == endObject and startOffset == endOffset:
return None
return AXSelectionRange(
start_object=startObject,
start_offset=startOffset,
end_object=endObject,
end_offset=endOffset,
start_is_active=startIsActive,
)
def _is_entire_document_selected(self, ranges: list[AXSelectionRange]) -> bool:
if len(ranges) != 1:
return False
selectionRange = ranges[0]
if selectionRange.start_offset != 0:
return False
if selectionRange.end_offset != AXText.get_character_count(selectionRange.end_object):
return False
if self._find_previous_text_object(selectionRange.start_object) is not None:
return False
if self._find_next_text_object(selectionRange.end_object) is not None:
return False
return True
def _find_previous_text_object(self, obj: Atspi.Accessible) -> Optional[Atspi.Accessible]:
current = obj
while current:
current = self._utilities.findPreviousObject(current)
if current is None:
return None
text = self._utilities.queryNonEmptyText(current)
if text:
return text
return None
def _find_next_text_object(self, obj: Atspi.Accessible) -> Optional[Atspi.Accessible]:
current = obj
while current:
current = self._utilities.findNextObject(current)
if current is None:
return None
text = self._utilities.queryNonEmptyText(current)
if text:
return text
return None
def _compare_ranges(self, range1: AXSelectionRange, range2: AXSelectionRange) -> int:
comparison = self._compare_positions(
range1.start_object,
range1.start_offset,
range2.start_object,
range2.start_offset,
)
if comparison is None:
return 0
return comparison
def _compare_positions(
self,
obj1: Atspi.Accessible,
offset1: int,
obj2: Atspi.Accessible,
offset2: int,
) -> Optional[int]:
if obj1 == obj2:
if offset1 < offset2:
return -1
if offset1 > offset2:
return 1
return 0
seen = {hash(obj1)}
current = obj1
while current:
current = self._utilities.findNextObject(current)
if current is None:
break
if hash(current) in seen:
break
if current == obj2:
return -1
seen.add(hash(current))
seen = {hash(obj2)}
current = obj2
while current:
current = self._utilities.findNextObject(current)
if current is None:
break
if hash(current) in seen:
break
if current == obj1:
return 1
seen.add(hash(current))
return None
def _objects_for_range(self, selectionRange: AXSelectionRange) -> list[Atspi.Accessible]:
objects = [selectionRange.start_object]
if selectionRange.start_object == selectionRange.end_object:
return objects
seen = {hash(selectionRange.start_object)}
current = selectionRange.start_object
while current and current != selectionRange.end_object:
current = self._utilities.findNextObject(current)
if current is None or hash(current) in seen:
return []
seen.add(hash(current))
if self._utilities.queryNonEmptyText(current):
objects.append(current)
if not objects or objects[-1] != selectionRange.end_object:
return []
return objects
def _project_range_onto_object(
self,
selectionRange: AXSelectionRange,
obj: Atspi.Accessible,
) -> list[tuple[int, int]]:
rangeObjects = self._objects_for_range(selectionRange)
if obj not in rangeObjects:
return []
if selectionRange.start_object == selectionRange.end_object == obj:
startOffset = min(selectionRange.start_offset, selectionRange.end_offset)
endOffset = max(selectionRange.start_offset, selectionRange.end_offset)
if startOffset < endOffset:
return [(startOffset, endOffset)]
return []
if obj == selectionRange.start_object:
startOffset = selectionRange.start_offset
endOffset = AXText.get_character_count(obj)
return [(startOffset, endOffset)] if startOffset < endOffset else []
if obj == selectionRange.end_object:
startOffset = 0
endOffset = selectionRange.end_offset
return [(startOffset, endOffset)] if startOffset < endOffset else []
endOffset = AXText.get_character_count(obj)
return [(0, endOffset)] if endOffset else []
@staticmethod
def _merge_ranges(ranges: list[tuple[int, int]]) -> list[tuple[int, int]]:
if not ranges:
return []
merged = []
for startOffset, endOffset in sorted(ranges):
if not merged or startOffset > merged[-1][1]:
merged.append((startOffset, endOffset))
continue
merged[-1] = (merged[-1][0], max(merged[-1][1], endOffset))
return merged
def _range_text(self, selectionRange: AXSelectionRange) -> str:
fragments = []
for obj in self._objects_for_range(selectionRange):
if obj == selectionRange.start_object == selectionRange.end_object:
startOffset = selectionRange.start_offset
endOffset = selectionRange.end_offset
elif obj == selectionRange.start_object:
startOffset = selectionRange.start_offset
endOffset = AXText.get_character_count(obj)
elif obj == selectionRange.end_object:
startOffset = 0
endOffset = selectionRange.end_offset
else:
startOffset = 0
endOffset = AXText.get_character_count(obj)
if startOffset >= endOffset:
continue
text = self._utilities.expandEOCs(obj, startOffset, endOffset)
if text:
fragments.append(text)
return " ".join(fragments)
def _resolve_primary_range_index(
self,
ranges: list[AXSelectionRange],
obj: Atspi.Accessible,
caretOffset: int,
) -> Optional[int]:
containing = []
projected = []
active = []
for index, selectionRange in enumerate(ranges):
localRanges = self._project_range_onto_object(selectionRange, obj)
if localRanges:
projected.append(index)
for startOffset, endOffset in localRanges:
if startOffset <= caretOffset <= endOffset:
containing.append(index)
break
if selectionRange.active_endpoint()[0] == obj:
active.append(index)
if len(containing) == 1:
return containing[0]
if len(projected) == 1:
return projected[0]
if len(active) == 1:
return active[0]
return None
def _build_range(
self,
anchorObject: Atspi.Accessible,
anchorOffset: int,
focusObject: Atspi.Accessible,
focusOffset: int,
) -> Optional[AXSelectionRange]:
comparison = self._compare_positions(anchorObject, anchorOffset, focusObject, focusOffset)
if comparison is None:
return None
if comparison <= 0:
return AXSelectionRange(
start_object=anchorObject,
start_offset=anchorOffset,
end_object=focusObject,
end_offset=focusOffset,
start_is_active=False,
)
return AXSelectionRange(
start_object=focusObject,
start_offset=focusOffset,
end_object=anchorObject,
end_offset=anchorOffset,
start_is_active=True,
)
def _set_document_ranges(
self,
document: Optional[Atspi.Accessible],
ranges: list[AXSelectionRange],
) -> bool:
if not document:
return False
selections = []
for selectionRange in ranges:
selection = Atspi.TextSelection()
selection.start_object = selectionRange.start_object
selection.start_offset = selectionRange.start_offset
selection.end_object = selectionRange.end_object
selection.end_offset = selectionRange.end_offset
selection.start_is_active = selectionRange.start_is_active
selections.append(selection)
try:
result = Atspi.Document.set_text_selections(document, selections)
except (AttributeError, GLib.GError, TypeError) as error:
msg = f"AXDocumentSelection: Exception in set_text_selections: {error}"
debug.printMessage(debug.LEVEL_INFO, msg, True)
return False
return bool(result)
+1
View File
@@ -6,6 +6,7 @@ cthulhu_python_sources = files([
'ax_collection.py',
'ax_component.py',
'ax_document.py',
'ax_document_selection.py',
'ax_event_synthesizer.py',
'ax_hypertext.py',
'ax_object.py',
+216 -66
View File
@@ -66,6 +66,7 @@ from . import pronunciation_dict
from . import settings
from . import settings_manager
from . import text_attribute_names
from .ax_document_selection import AXDocumentSelection
from .ax_object import AXObject
from .ax_selection import AXSelection
from .ax_table import AXTable
@@ -161,6 +162,7 @@ class Utilities:
self._clipboardHandlerId: Optional[int] = None
self._lastIndentationData: dict[Any, Any] = {}
self._selectedMenuBarMenu: dict[Any, Any] = {}
self._documentSelectionAdapter = AXDocumentSelection(self)
#########################################################################
# #
@@ -2405,6 +2407,137 @@ class Utilities:
# #
#########################################################################
def _getSelectionAdapter(self) -> AXDocumentSelection:
adapter = getattr(self, "_documentSelectionAdapter", None)
if adapter is None:
adapter = AXDocumentSelection(self)
self._documentSelectionAdapter = adapter
return adapter
def _getSelectionState(self) -> dict[Any, Any]:
state = self._script.pointOfReference.get('selectionState')
if state is None:
state = {'textSelections': {}}
self._script.pointOfReference['selectionState'] = state
state.setdefault('textSelections', {})
return state
def _getCachedSelectionData(self, obj: Atspi.Accessible) -> dict[str, Any]:
if not obj:
return {'ranges': [], 'start': 0, 'end': 0, 'string': ''}
cached = self._getSelectionState().get('textSelections', {}).get(hash(obj), {})
return {
'ranges': list(cached.get('ranges', [])),
'start': cached.get('start', 0),
'end': cached.get('end', 0),
'string': cached.get('string', ''),
}
def _cacheSelectionData(self, obj: Atspi.Accessible, data: dict[str, Any]) -> None:
textSelections = self._getSelectionState().setdefault('textSelections', {})
if hash(obj) in textSelections:
value = textSelections.pop(hash(obj))
for x in [k for k in textSelections.keys() if textSelections.get(k) == value]:
textSelections.pop(x)
textSelections[hash(obj)] = {
'ranges': list(data.get('ranges', [])),
'start': data.get('start', 0),
'end': data.get('end', 0),
'string': data.get('string', ''),
}
def _getSelectionTextFragment(self, obj: Atspi.Accessible, start: int, end: int) -> str:
text = self.expandEOCs(obj, start, end)
if text:
return text
fallback = AXText.get_substring(obj, start, end)
if fallback:
tokens = [
"SCRIPT UTILITIES: Falling back to raw substring for selection text in",
obj,
f"({start}, {end})",
]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
return fallback
return ""
def _getLocalSelectionData(self, obj: Atspi.Accessible) -> dict[str, Any]:
if not AXObject.supports_text(obj):
return {'ranges': [], 'start': 0, 'end': 0, 'string': ''}
textContents = ""
startOffset = endOffset = 0
ranges = AXText.get_selected_ranges(obj)
for start, end in ranges:
if start == end:
continue
selectedText = self._getSelectionTextFragment(obj, start, end)
if textContents:
textContents += " "
textContents += selectedText
if startOffset == endOffset == 0:
startOffset = start
endOffset = end
return {
'ranges': ranges,
'start': startOffset,
'end': endOffset,
'string': textContents,
}
def _getSelectionData(
self,
obj: Atspi.Accessible,
snapshot: Any = None,
) -> dict[str, Any]:
if not AXObject.supports_text(obj):
return {'ranges': [], 'start': 0, 'end': 0, 'string': ''}
snapshot = snapshot if snapshot is not None else self._getSelectionAdapter().refresh_snapshot(obj)
if snapshot.is_authoritative:
projected = self._getSelectionAdapter().get_projected_selection(obj, snapshot)
return {
'ranges': list(projected.local_ranges),
'start': projected.start_offset,
'end': projected.end_offset,
'string': projected.text,
}
return self._getLocalSelectionData(obj)
@staticmethod
def _rangesToCharSet(ranges: list[tuple[int, int]]) -> set[int]:
chars = set()
for start, end in ranges:
chars.update(range(start, end))
return chars
@staticmethod
def _charSetToRanges(chars: set[int]) -> list[tuple[int, int]]:
if not chars:
return []
change = sorted(chars)
ranges = []
startOffset = change[0]
endOffset = change[0] + 1
for offset in change[1:]:
if offset == endOffset:
endOffset += 1
continue
ranges.append((startOffset, endOffset))
startOffset = offset
endOffset = offset + 1
ranges.append((startOffset, endOffset))
return ranges
def adjustTextSelection(self, obj: Atspi.Accessible, offset: int) -> None:
"""Adjusts the end point of a text selection
@@ -2417,6 +2550,9 @@ class Utilities:
if not AXObject.supports_text(obj):
return
if self._getSelectionAdapter().adjust_selection(obj, offset):
return
selections = AXText.get_selected_ranges(obj)
if not selections:
caretOffset = AXText.get_caret_offset(obj)
@@ -2467,6 +2603,10 @@ class Utilities:
offsets within the text for the given object.
"""
snapshot = self._getSelectionAdapter().refresh_snapshot(obj)
if snapshot.is_authoritative:
return self._getSelectionAdapter().get_all_selected_text(obj, snapshot)
textContents, startOffset, endOffset = self.selectedText(obj)
if textContents and self._script.pointOfReference.get('entireDocumentSelected'):
return textContents, startOffset, endOffset
@@ -2506,6 +2646,10 @@ class Utilities:
the text interface.
"""
snapshot = self._getSelectionAdapter().refresh_snapshot(obj)
if snapshot.is_authoritative:
return self._getSelectionAdapter().get_projected_selection(obj, snapshot).local_ranges
return AXText.get_selected_ranges(obj)
def getChildAtOffset(self, obj: Atspi.Accessible, offset: Any) -> Optional[Any]:
@@ -2589,6 +2733,11 @@ class Utilities:
if not AXObject.supports_text(obj):
return
if self._getSelectionAdapter().clear_selection(obj):
self._script.pointOfReference['entireDocumentSelected'] = False
self._getSelectionState()['textSelections'] = {}
return
for i in range(AXText._get_n_selections(obj)):
AXText._remove_selection(obj, i)
@@ -2738,20 +2887,8 @@ class Utilities:
offsets within the text.
"""
textContents = ""
startOffset = endOffset = 0
for start, end in AXText.get_selected_ranges(obj):
if start == end:
continue
selectedText = self.expandEOCs(obj, start, end)
if textContents:
textContents += " "
textContents += selectedText
if startOffset == endOffset == 0:
startOffset = start
endOffset = end
return [textContents, startOffset, endOffset]
selectionData = self._getSelectionData(obj)
return [selectionData['string'], selectionData['start'], selectionData['end']]
def getCaretContext(self) -> Any:
obj = cthulhu_state.locusOfFocus
@@ -4841,8 +4978,8 @@ class Utilities:
return ""
def getCachedTextSelection(self, obj: Atspi.Accessible) -> Any:
textSelections = self._script.pointOfReference.get('textSelections', {})
start, end, string = textSelections.get(hash(obj), (0, 0, ''))
cached = self._getCachedSelectionData(obj)
start, end, string = cached['start'], cached['end'], cached['string']
tokens = ["SCRIPT UTILITIES: Cached selection for", obj, f"is '{string}' ({start}, {end})"]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
return start, end, string
@@ -4851,44 +4988,41 @@ class Utilities:
if not AXObject.supports_text(obj):
tokens = ["SCRIPT UTILITIES:", obj, "doesn't implement AtspiText"]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
text = None
else:
text = obj
self._cacheSelectionData(obj, {'ranges': [], 'start': 0, 'end': 0, 'string': ''})
return
if self._script.pointOfReference.get('entireDocumentSelected'):
selectedText, selectedStart, selectedEnd = self.allSelectedText(obj)
if not selectedText:
self._script.pointOfReference['entireDocumentSelected'] = False
self._script.pointOfReference['textSelections'] = {}
self._getSelectionState()['textSelections'] = {}
textSelections = self._script.pointOfReference.get('textSelections', {})
snapshot = self._getSelectionAdapter().refresh_snapshot(obj)
selectionData = self._getSelectionData(obj, snapshot)
if snapshot.is_authoritative and not snapshot.ranges:
self._script.pointOfReference['entireDocumentSelected'] = False
self._getSelectionState()['textSelections'] = {}
# Because some apps and toolkits create, destroy, and duplicate objects
# and events.
if hash(obj) in textSelections:
value = textSelections.pop(hash(obj))
for x in [k for k in textSelections.keys() if textSelections.get(k) == value]:
textSelections.pop(x)
start, end, string = 0, 0, ''
if text:
string, start, end = self._getSingleSelectionText(obj)
tokens = ["SCRIPT UTILITIES: New selection for", obj, f"is '{string}' ({start}, {end})"]
tokens = [
"SCRIPT UTILITIES: New selection for",
obj,
f"is '{selectionData['string']}' ({selectionData['start']}, {selectionData['end']})",
]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
textSelections[hash(obj)] = start, end, string
self._script.pointOfReference['textSelections'] = textSelections
self._cacheSelectionData(obj, selectionData)
def _getSingleSelectionText(self, obj: Atspi.Accessible) -> Any:
# NOTE: Does not handle multiple non-contiguous selections.
string, start, end = AXText.get_selected_text(obj)
if string:
string = self.expandEOCs(obj, start, end)
string = self._getSelectionTextFragment(obj, start, end)
return string, start, end
def onClipboardContentsChanged(self, *args: Any) -> None:
script = cthulhu_state.activeScript
if not script:
msg = "SCRIPT UTILITIES: Clipboard contents changed with no active script"
debug.printMessage(debug.LEVEL_INFO, msg, True)
return
if time.time() - Utilities._last_clipboard_update < 0.05:
@@ -4897,6 +5031,8 @@ class Utilities:
return
Utilities._last_clipboard_update = time.time()
tokens = ["SCRIPT UTILITIES: Clipboard contents changed for", script]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
script.onClipboardContentsChanged(*args)
def connectToClipboard(self) -> None:
@@ -5243,6 +5379,10 @@ class Utilities:
if not contents:
return False
string, _, _ = self.allSelectedText(obj)
if string and string in contents:
return True
string, start, end = self.selectedText(obj)
if string and string in contents:
return True
@@ -5251,7 +5391,11 @@ class Utilities:
if AXObject.is_dead(obj):
return False
return obj and AXObject.get_name(obj) in contents
name = AXObject.get_name(obj)
if obj and name in contents:
return True
return False
def clearCachedCommandState(self) -> bool:
self._script.pointOfReference['undo'] = False
@@ -5404,9 +5548,19 @@ class Utilities:
if not AXObject.supports_text(obj):
return False
oldStart, oldEnd, oldString = self.getCachedTextSelection(obj)
self.updateCachedTextSelection(obj)
newStart, newEnd, newString = self.getCachedTextSelection(obj)
oldSelection = self._getCachedSelectionData(obj)
oldStart = oldSelection['start']
oldEnd = oldSelection['end']
oldString = oldSelection['string']
oldRanges = oldSelection['ranges']
snapshot = self._getSelectionAdapter().refresh_snapshot(obj)
newSelection = self._getSelectionData(obj, snapshot)
self._cacheSelectionData(obj, newSelection)
newStart = newSelection['start']
newEnd = newSelection['end']
newString = newSelection['string']
newRanges = newSelection['ranges']
if self._speakTextSelectionState(len(newString)):
return True
@@ -5418,38 +5572,23 @@ class Utilities:
return False
changes = []
oldChars = set(range(oldStart, oldEnd))
newChars = set(range(newStart, newEnd))
oldChars = self._rangesToCharSet(oldRanges)
newChars = self._rangesToCharSet(newRanges)
if not oldChars.union(newChars):
return False
if oldChars and newChars and not oldChars.intersection(newChars):
# A simultaneous unselection and selection centered at one offset.
changes.append([oldStart, oldEnd, messages.TEXT_UNSELECTED])
changes.append([newStart, newEnd, messages.TEXT_SELECTED])
else:
change = sorted(oldChars.symmetric_difference(newChars))
if not change:
return False
for start, end in self._charSetToRanges(oldChars - newChars):
changes.append([start, end, messages.TEXT_UNSELECTED])
changeStart, changeEnd = change[0], change[-1] + 1
if oldChars < newChars:
changes.append([changeStart, changeEnd, messages.TEXT_SELECTED])
if oldString.endswith(self.EMBEDDED_OBJECT_CHARACTER) and oldEnd == changeStart:
# There's a possibility that we have a link spanning multiple lines. If so,
# we want to present the continuation that just became selected.
child = self.findChildAtOffset(obj, oldEnd - 1)
self.handleTextSelectionChange(child, False)
else:
changes.append([changeStart, changeEnd, messages.TEXT_UNSELECTED])
if newString.endswith(self.EMBEDDED_OBJECT_CHARACTER):
# There's a possibility that we have a link spanning multiple lines. If so,
# we want to present the continuation that just became unselected.
child = self.findChildAtOffset(obj, newEnd - 1)
self.handleTextSelectionChange(child, False)
for start, end in self._charSetToRanges(newChars - oldChars):
changes.append([start, end, messages.TEXT_SELECTED])
if not changes:
return False
speakMessage = speakMessage and not cthulhu.cthulhuApp.settingsManager.getSetting('onlySpeakDisplayedText')
for start, end, message in changes:
maxFragments = 3
for start, end, message in changes[:maxFragments]:
string = AXText.get_substring(obj, start, end)
endsWithChild = string.endswith(self.EMBEDDED_OBJECT_CHARACTER)
if endsWithChild:
@@ -5463,6 +5602,17 @@ class Utilities:
child = self.findChildAtOffset(obj, end)
self.handleTextSelectionChange(child, speakMessage)
remaining = changes[maxFragments:]
if speakMessage and remaining:
selectedCount = sum(1 for x in remaining if x[2] == messages.TEXT_SELECTED)
unselectedCount = sum(1 for x in remaining if x[2] == messages.TEXT_UNSELECTED)
if selectedCount and not unselectedCount:
self._script.speakMessage(f"{selectedCount} more ranges selected", interrupt=False)
elif unselectedCount and not selectedCount:
self._script.speakMessage(f"{unselectedCount} more ranges unselected", interrupt=False)
else:
self._script.speakMessage(f"{len(remaining)} more selection changes", interrupt=False)
return True
def _getCtrlShiftSelectionsStrings(self) -> Any:
@@ -313,6 +313,7 @@ class Script(Gecko.Script):
# Mozilla cannot seem to get their ":system" suffix right
# to save their lives, so we'll add yet another sad hack.
# Intentional object-local check: autocomplete selection lives in this entry.
selections = AXText.get_selected_ranges(event.source)
hasSelection = bool(selections)
if hasSelection or isSystemEvent:
@@ -332,6 +333,7 @@ class Script(Gecko.Script):
return
if self.utilities.isEditableMessage(obj) and self.spellcheck.isActive():
# Intentional object-local check: spellcheck position is tracked within this entry.
selections = AXText.get_selected_ranges(obj)
if selections:
selStart, selEnd = selections[0]
@@ -86,6 +86,7 @@ class Utilities(script_utilities.Utilities):
debug.printTokens(debug.LEVEL_INFO, tokens, True)
text = self.queryNonEmptyText(obj)
# Intentional object-local fallback for GNOME Shell's broken selection ranges.
if AXText.get_selected_ranges(obj):
string = AXText.get_all_text(obj)
start, end = 0, len(string)
@@ -333,6 +333,8 @@ class Utilities(script_utilities.Utilities):
relations = filter(lambda r: r.getRelationType() in flows, relationSet)
targets = [r.getTarget(0) for r in relations]
for target in targets:
# Intentional object-local probe: we only care whether this related object
# itself currently exposes selected text.
if AXText.get_selected_ranges(target):
return True
+428 -9
View File
@@ -40,6 +40,7 @@ gi.require_version('Atspi', '2.0')
gi.require_version('Gdk', '3.0')
from gi.repository import Atspi
from gi.repository import Gdk
from gi.repository import GLib
import re
import time
@@ -66,6 +67,7 @@ import cthulhu.settings_manager as settings_manager
import cthulhu.sound as sound
import cthulhu.speech as speech
import cthulhu.speechserver as speechserver
import cthulhu.wnck_support as wnck_support
from cthulhu.ax_object import AXObject
from cthulhu.ax_value import AXValue
from cthulhu.ax_text import AXText
@@ -85,6 +87,9 @@ class Script(script.Script):
EMBEDDED_OBJECT_CHARACTER = '\ufffc'
NO_BREAK_SPACE_CHARACTER = '\u00a0'
KEYBOARD_CLIPBOARD_CHECK_DELAY_MS = 100
KEYBOARD_CLIPBOARD_MAX_CHECKS = 10
KEYBOARD_CLIPBOARD_RELEASE_MAX_AGE_S = 1.0
# generatorCache
#
@@ -133,6 +138,10 @@ class Script(script.Script):
self._sayAllContexts = []
self.grab_ids = []
self._modifierGrabIds = []
self._pendingKeyboardClipboardCommand = None
self._pendingKeyboardClipboardRelease = None
self._keyboardClipboardCommandGeneration = 0
self._sessionType = None
if app:
Atspi.Accessible.set_cache_mask(
@@ -650,6 +659,394 @@ class Script(script.Script):
super().deregisterEventListeners()
self.utilities.disconnectFromClipboard()
@staticmethod
def _normalizeKeyboardCommandKeyName(keyboardEvent):
for value in [
getattr(keyboardEvent, "keyval_name", ""),
getattr(keyboardEvent, "event_string", ""),
]:
keyName = (value or "").lower()
if not keyName:
continue
if len(keyName) == 1:
codePoint = ord(keyName)
if 1 <= codePoint <= 26:
return chr(ord("a") + codePoint - 1)
return keyName
if keyName.startswith("0x"):
try:
codePoint = int(keyName, 16)
except ValueError:
continue
if 1 <= codePoint <= 26:
return chr(ord("a") + codePoint - 1)
return keyName
return ""
@staticmethod
def _getKeyboardClipboardCommandType(keyboardEvent):
keyName = Script._normalizeKeyboardCommandKeyName(keyboardEvent)
modifiers = getattr(keyboardEvent, "modifiers", 0)
control = bool(modifiers & 1 << Atspi.ModifierType.CONTROL)
shift = bool(modifiers & 1 << Atspi.ModifierType.SHIFT)
obj = keyboardEvent.get_object()
if not control:
return None
if keyName == "c":
if AXUtilities.is_terminal(obj):
return "copy" if shift else None
return None if shift else "copy"
if keyName == "x":
return None if shift else "cut"
return None
def _rememberKeyboardClipboardCommand(self, commandType, sourceObj):
self._keyboardClipboardCommandGeneration += 1
capturedSelectionText = self._getKeyboardClipboardSelectionText(sourceObj)
expectedContents = self._getKeyboardClipboardExpectedContents(sourceObj)
self._pendingKeyboardClipboardCommand = {
"generation": self._keyboardClipboardCommandGeneration,
"command": commandType,
"source": sourceObj,
"clipboard": self.utilities.getClipboardContents(),
"capturedSelectionText": capturedSelectionText,
"expectedContents": expectedContents,
"callbackSeen": False,
"presented": False,
"checksRemaining": self.KEYBOARD_CLIPBOARD_MAX_CHECKS,
"scheduled": False,
}
tokens = ["DEFAULT: Tracking keyboard clipboard command", commandType, "from", sourceObj]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
tokens = [
"DEFAULT: Captured keyboard clipboard expectations",
self._describeClipboardContents(expectedContents),
]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
def _getKeyboardClipboardExpectedContents(self, sourceObj):
expectedContents = []
for getter in [self.utilities.allSelectedText, self.utilities.selectedText]:
try:
result = getter(sourceObj)
except Exception:
continue
if not isinstance(result, tuple) or not result:
continue
string = result[0]
if string and string not in expectedContents:
expectedContents.append(string)
try:
name = AXObject.get_name(sourceObj)
except Exception:
name = ""
if name and name not in expectedContents:
expectedContents.append(name)
return expectedContents
def _getKeyboardClipboardSelectionText(self, sourceObj):
for getter in [self.utilities.allSelectedText, self.utilities.selectedText]:
try:
result = getter(sourceObj)
except Exception:
continue
if not isinstance(result, tuple) or not result:
continue
string = result[0]
if string:
return string
return ""
@staticmethod
def _clipboardContentsMatchExpectedContents(currentContents, expectedContents):
if not currentContents:
return False
for string in expectedContents or []:
if string and string in currentContents:
return True
return False
@staticmethod
def _describeClipboardContents(contents):
if isinstance(contents, (list, tuple)):
return [Script._describeClipboardContents(item) for item in contents]
if contents is None:
return {"len": 0, "text": ""}
string = str(contents)
if len(string) > 80:
string = f"{string[:77]}..."
string = string.replace("\n", "\\n")
return {"len": len(str(contents)), "text": string}
def _clipboardMatchesPendingKeyboardCommand(self, pending, currentContents):
if self._clipboardContentsMatchExpectedContents(
currentContents, pending.get("expectedContents")
):
tokens = ["DEFAULT: Keyboard clipboard fallback matched captured contents for",
pending.get("source")]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
return True
sourceObj = pending.get("source")
if bool(currentContents) and self.utilities.objectContentsAreInClipboard(sourceObj):
return True
if sourceObj and self.utilities.isInActiveApp(sourceObj) \
and bool(currentContents) and self.utilities.objectContentsAreInClipboard():
tokens = ["DEFAULT: Keyboard clipboard fallback matched locusOfFocus after source",
"mismatch", sourceObj, cthulhu_state.locusOfFocus]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
return True
return False
def _schedulePendingKeyboardClipboardCheck(self):
pending = self._pendingKeyboardClipboardCommand
if not pending or pending.get("scheduled"):
return
pending["scheduled"] = True
GLib.timeout_add(
self.KEYBOARD_CLIPBOARD_CHECK_DELAY_MS,
self._checkPendingKeyboardClipboardCommand,
pending["generation"],
)
def _scheduleImmediateKeyboardClipboardPresentation(self):
pending = self._pendingKeyboardClipboardCommand
if not pending or pending.get("immediateScheduled") or pending.get("presented"):
return
pending["immediateScheduled"] = True
GLib.idle_add(
self._presentPendingKeyboardClipboardCommandIdle,
pending["generation"],
)
def _getSessionType(self):
sessionType = getattr(self, "_sessionType", None)
if sessionType is None:
sessionType = wnck_support.get_session_type()
self._sessionType = sessionType
return sessionType
def _shouldUseImmediateKeyboardClipboardPolicy(self, pending):
if not pending or self._getSessionType() != "wayland":
return False
if pending.get("command") not in ["copy", "cut"]:
return False
if not pending.get("capturedSelectionText") and not pending.get("expectedContents"):
return False
if pending.get("command") == "cut":
sourceObj = pending.get("source")
if AXUtilities.is_editable(sourceObj):
return False
return True
def _rememberKeyboardClipboardRelease(self, commandType):
self._pendingKeyboardClipboardRelease = {
"command": commandType,
"time": time.time(),
}
def _consumePendingKeyboardClipboardRelease(self, commandType):
pending = getattr(self, "_pendingKeyboardClipboardRelease", None)
self._pendingKeyboardClipboardRelease = None
if not pending or pending.get("command") != commandType:
return False
return time.time() - pending.get("time", 0.0) <= self.KEYBOARD_CLIPBOARD_RELEASE_MAX_AGE_S
def _presentPendingKeyboardClipboardCommand(self, pending, reason):
sourceObj = pending.get("source")
if pending.get("command") == "cut":
if AXUtilities.is_editable(sourceObj):
pending["scheduled"] = False
pending["immediateScheduled"] = False
self._pendingKeyboardClipboardCommand = None
return
self.presentMessage(messages.CLIPBOARD_CUT_FULL, messages.CLIPBOARD_CUT_BRIEF)
else:
self.presentMessage(messages.CLIPBOARD_COPIED_FULL, messages.CLIPBOARD_COPIED_BRIEF)
tokens = ["DEFAULT: Presented keyboard clipboard", reason, "for", pending.get("command")]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
pending["scheduled"] = False
pending["immediateScheduled"] = False
pending["presented"] = True
def _presentPendingKeyboardClipboardCommandIdle(self, generation):
pending = self._pendingKeyboardClipboardCommand
if not pending or pending.get("generation") != generation:
return False
if pending.get("presented"):
pending["immediateScheduled"] = False
return False
self._presentPendingKeyboardClipboardCommand(pending, "immediately")
return False
def _checkPendingKeyboardClipboardCommand(self, generation):
pending = self._pendingKeyboardClipboardCommand
if not pending or pending.get("generation") != generation:
return False
if pending.get("presented"):
pending["scheduled"] = False
self._pendingKeyboardClipboardCommand = None
return False
pending["checksRemaining"] = max(pending.get("checksRemaining", 1) - 1, 0)
currentContents = self.utilities.getClipboardContents()
sourceObj = pending.get("source")
tokens = [
"DEFAULT: Keyboard clipboard retry snapshot",
{
"command": pending.get("command"),
"source": sourceObj,
"clipboard": self._describeClipboardContents(currentContents),
"expected": self._describeClipboardContents(pending.get("expectedContents")),
"checksRemaining": pending.get("checksRemaining"),
},
]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
clipboardMatchesSource = self._clipboardMatchesPendingKeyboardCommand(
pending, currentContents
)
selectionFallback = False
if not clipboardMatchesSource:
selectionFallback = self._shouldPresentKeyboardClipboardSelectionFallback(
pending, currentContents
)
if selectionFallback:
msg = "DEFAULT: Keyboard clipboard fallback matched captured selection without clipboard access"
debug.printMessage(debug.LEVEL_INFO, msg, True)
clipboardMatchesSource = True
if not clipboardMatchesSource:
if pending.get("checksRemaining", 0):
tokens = ["DEFAULT: Keyboard clipboard fallback retrying", pending.get("command"),
"for", sourceObj, "checks remaining", pending.get("checksRemaining")]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
return True
tokens = ["DEFAULT: Keyboard clipboard fallback did not match", sourceObj]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
pending["scheduled"] = False
self._pendingKeyboardClipboardCommand = None
return False
if not selectionFallback and currentContents == pending.get("clipboard"):
if pending.get("command") != "copy":
msg = "DEFAULT: Keyboard clipboard fallback found no clipboard change"
debug.printMessage(debug.LEVEL_INFO, msg, True)
pending["scheduled"] = False
self._pendingKeyboardClipboardCommand = None
return False
msg = "DEFAULT: Keyboard clipboard fallback matched unchanged clipboard for copy"
debug.printMessage(debug.LEVEL_INFO, msg, True)
if pending.get("command") == "cut":
if AXUtilities.is_editable(sourceObj):
pending["scheduled"] = False
self._pendingKeyboardClipboardCommand = None
return False
self._presentPendingKeyboardClipboardCommand(pending, "fallback")
return False
@staticmethod
def _shouldPresentKeyboardClipboardSelectionFallback(pending, currentContents):
if pending.get("command") != "copy":
return False
if currentContents:
return False
return bool(pending.get("capturedSelectionText"))
def updateKeyboardEventState(self, keyboardEvent, handler):
super().updateKeyboardEventState(keyboardEvent, handler)
commandType = self._getKeyboardClipboardCommandType(keyboardEvent)
if not commandType:
return
if keyboardEvent.is_pressed_key():
self._rememberKeyboardClipboardCommand(commandType, keyboardEvent.get_object())
pending = self._pendingKeyboardClipboardCommand
if self._shouldUseImmediateKeyboardClipboardPolicy(pending):
self._consumePendingKeyboardClipboardRelease(commandType)
msg = "DEFAULT: Scheduling keyboard clipboard immediate presentation on press"
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._scheduleImmediateKeyboardClipboardPresentation()
return
if self._consumePendingKeyboardClipboardRelease(commandType):
if self._shouldUseImmediateKeyboardClipboardPolicy(pending):
msg = "DEFAULT: Scheduling keyboard clipboard immediate presentation after out-of-order release"
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._scheduleImmediateKeyboardClipboardPresentation()
else:
msg = "DEFAULT: Scheduling keyboard clipboard fallback after out-of-order release"
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._schedulePendingKeyboardClipboardCheck()
return
pending = self._pendingKeyboardClipboardCommand
if self._getSessionType() == "wayland":
if pending is None:
return
if pending.get("command") == commandType and pending.get("immediateScheduled"):
msg = "DEFAULT: Ignoring keyboard clipboard release while immediate presentation is pending"
debug.printMessage(debug.LEVEL_INFO, msg, True)
return
if pending.get("command") == commandType and pending.get("presented"):
msg = "DEFAULT: Ignoring keyboard clipboard release after immediate presentation"
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._pendingKeyboardClipboardCommand = None
return
if not pending or pending.get("command") != commandType:
self._rememberKeyboardClipboardRelease(commandType)
return
if self._shouldUseImmediateKeyboardClipboardPolicy(pending):
msg = "DEFAULT: Scheduling keyboard clipboard immediate presentation"
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._scheduleImmediateKeyboardClipboardPresentation()
return
self._schedulePendingKeyboardClipboardCheck()
def _saveFocusedObjectInfo(self, obj):
"""Saves some basic information about obj. Note that this method is
intended to be called primarily (if not only) by locus_of_focus_changed().
@@ -1048,10 +1445,9 @@ class Script(script.Script):
if caretOffset >= 0:
self.utilities.adjustTextSelection(obj, caretOffset)
selections = AXText.get_selected_ranges(obj)
if selections:
startOffset, endOffset = selections[0]
self.utilities.setClipboardText(AXText.get_substring(obj, startOffset, endOffset))
selectedText, startOffset, endOffset = self.utilities.allSelectedText(obj)
if selectedText:
self.utilities.setClipboardText(selectedText)
return True
@@ -1367,7 +1763,7 @@ class Script(script.Script):
return
self._saveLastCursorPosition(event.source, caretOffset)
if AXText.get_selected_ranges(event.source):
if self.utilities.allTextSelections(event.source):
msg = "DEFAULT: Event source has text selections"
debug.printMessage(debug.LEVEL_INFO, msg, True)
self.utilities.handleTextSelectionChange(event.source)
@@ -2013,10 +2409,30 @@ class Script(script.Script):
cthulhu.cthulhuApp.scriptManager.set_active_script(None, "focus: window-deactivated")
def onClipboardContentsChanged(self, *args):
pending = self._pendingKeyboardClipboardCommand
currentContents = self.utilities.getClipboardContents()
tokens = [
"DEFAULT: Clipboard callback snapshot",
self._describeClipboardContents(currentContents),
]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
if pending is not None:
pending["callbackSeen"] = True
if pending.get("presented"):
self._pendingKeyboardClipboardCommand = None
return
if self.flatReviewPresenter.is_active():
return
if not self.utilities.objectContentsAreInClipboard():
if pending is not None:
clipboardMatchesSource = self._clipboardMatchesPendingKeyboardCommand(
pending, currentContents
)
else:
clipboardMatchesSource = False
if not clipboardMatchesSource and not self.utilities.objectContentsAreInClipboard():
return
if not self.utilities.topLevelObjectIsActiveAndCurrent():
@@ -2024,6 +2440,8 @@ class Script(script.Script):
if self.utilities.lastInputEventWasCopy():
self.presentMessage(messages.CLIPBOARD_COPIED_FULL, messages.CLIPBOARD_COPIED_BRIEF)
if self._pendingKeyboardClipboardCommand is pending:
self._pendingKeyboardClipboardCommand = None
return
if not self.utilities.lastInputEventWasCut():
@@ -2033,6 +2451,8 @@ class Script(script.Script):
return
self.presentMessage(messages.CLIPBOARD_CUT_FULL, messages.CLIPBOARD_CUT_BRIEF)
if self._pendingKeyboardClipboardCommand is pending:
self._pendingKeyboardClipboardCommand = None
########################################################################
# #
@@ -2162,9 +2582,8 @@ class Script(script.Script):
# If there is a selection, clear it. See bug #489504 for more details.
#
if AXText.get_selected_ranges(context.obj):
AXText.set_selected_text(
context.obj, context.currentOffset, context.currentOffset)
if self.utilities.allTextSelections(context.obj):
self.utilities.clearTextSelection(context.obj)
def inSayAll(self, treatInterruptedAsIn=True):
if self._inSayAll:
+151 -1
View File
@@ -129,6 +129,128 @@ class Script(default.Script):
self.attributeNamesDict["text-align"] = "justification"
self.attributeNamesDict["text-indent"] = "indent"
@staticmethod
def _selectionContentKey(content):
obj, start, end, string = content
return hash(obj), start, end, string
def _clearSyntheticWebSelection(self):
pointOfReference = getattr(self, "pointOfReference", None)
if isinstance(pointOfReference, dict):
pointOfReference.pop("syntheticWebSelection", None)
def _compareCaretContexts(self, firstObj, firstOffset, secondObj, secondOffset):
if firstObj == secondObj:
if firstOffset < secondOffset:
return -1
if firstOffset > secondOffset:
return 1
return 0
return self.utilities.pathComparison(AXObject.get_path(firstObj), AXObject.get_path(secondObj))
def _getContentsBetweenCaretContexts(self, startObj, startOffset, endObj, endOffset):
contents = []
currentObj, currentOffset = startObj, startOffset
seen = set()
while currentObj and (currentObj, currentOffset) != (endObj, endOffset):
key = (hash(currentObj), currentOffset)
if key in seen:
break
seen.add(key)
contents.extend(self.utilities.getCharacterContentsAtOffset(currentObj, currentOffset))
currentObj, currentOffset = self.utilities.nextContext(currentObj, currentOffset)
return contents
def _presentSyntheticCaretSelection(
self,
event,
document,
obj,
offset,
focusOverride=None,
):
if not self.utilities.lastInputEventWasCaretNavWithSelection():
return False
if focusOverride is not None:
focusObj, focusOffset = focusOverride
else:
manager = input_event_manager.get_manager()
if manager.last_event_was_forward_caret_selection():
focusObj, focusOffset = self.utilities.nextContext(obj, offset)
elif manager.last_event_was_backward_caret_selection():
focusObj, focusOffset = self.utilities.previousContext(obj, offset)
else:
return False
if not focusObj:
return False
selectionState = self.pointOfReference.get("syntheticWebSelection", {})
if selectionState.get("document") == document:
anchorObj = selectionState.get("anchorObj", obj)
anchorOffset = selectionState.get("anchorOffset", offset)
oldContents = selectionState.get("contents", [])
else:
anchorObj, anchorOffset = obj, offset
oldContents = []
if self._compareCaretContexts(anchorObj, anchorOffset, focusObj, focusOffset) <= 0:
startObj, startOffset = anchorObj, anchorOffset
endObj, endOffset = focusObj, focusOffset
else:
startObj, startOffset = focusObj, focusOffset
endObj, endOffset = anchorObj, anchorOffset
newContents = self._getContentsBetweenCaretContexts(startObj, startOffset, endObj, endOffset)
self.utilities.setCaretContext(focusObj, focusOffset, document)
cthulhu.setLocusOfFocus(event, focusObj, False, True)
self.updateBraille(focusObj)
if not newContents:
self._clearSyntheticWebSelection()
if oldContents:
self.speakContents(oldContents)
self.speakMessage(messages.TEXT_UNSELECTED, interrupt=False)
return True
self.pointOfReference["syntheticWebSelection"] = {
"document": document,
"anchorObj": anchorObj,
"anchorOffset": anchorOffset,
"focusObj": focusObj,
"focusOffset": focusOffset,
"contents": newContents,
"string": "".join(item[3] for item in newContents),
}
oldKeys = [self._selectionContentKey(item) for item in oldContents]
newKeys = [self._selectionContentKey(item) for item in newContents]
deltaContents = newContents
message = messages.TEXT_SELECTED
if oldContents:
if len(newKeys) >= len(oldKeys) and newKeys[:len(oldKeys)] == oldKeys:
deltaContents = newContents[len(oldContents):]
elif len(newKeys) >= len(oldKeys) and newKeys[-len(oldKeys):] == oldKeys:
deltaContents = newContents[:-len(oldContents)]
elif len(oldKeys) >= len(newKeys) and oldKeys[:len(newKeys)] == newKeys:
deltaContents = oldContents[len(newContents):]
message = messages.TEXT_UNSELECTED
elif len(oldKeys) >= len(newKeys) and oldKeys[-len(newKeys):] == newKeys:
deltaContents = oldContents[:-len(newContents)]
message = messages.TEXT_UNSELECTED
if deltaContents:
self.speakContents(deltaContents)
self.speakMessage(message, interrupt=False)
return True
def activate(self):
"""Called when this script is activated."""
@@ -600,6 +722,7 @@ class Script(default.Script):
# won't know to dump the generator cache. See bgo#618827.
self.generatorCache = {}
self._lastMouseButtonContext = None, -1
super().updateKeyboardEventState(keyboardEvent, handler)
def shouldConsumeKeyboardEvent(self, keyboardEvent, handler):
"""Returns True if the script will consume this keyboard event."""
@@ -770,7 +893,7 @@ class Script(default.Script):
"""Updates the context and presents the find results if appropriate."""
text = self.utilities.queryNonEmptyText(obj)
selections = AXText.get_selected_ranges(obj) if text else []
selections = self.utilities.allTextSelections(obj) if text else []
if not selections:
return
@@ -1961,6 +2084,14 @@ class Script(default.Script):
tokens = ["WEB: Context: ", obj, ", ", offset, "(focus: ", cthulhu_state.locusOfFocus, ")"]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
if self.utilities.lastInputEventWasCaretNavWithSelection() and event.detail1 < 0:
if self._presentSyntheticCaretSelection(event, document, obj, offset):
msg = "WEB: Event handled: synthetic caret selection"
debug.printMessage(debug.LEVEL_INFO, msg, True)
return True
else:
self._clearSyntheticWebSelection()
if self._lastCommandWasCaretNav:
msg = "WEB: Event ignored: Last command was caret nav"
debug.printMessage(debug.LEVEL_INFO, msg, True)
@@ -2853,6 +2984,8 @@ class Script(default.Script):
def onTextSelectionChanged(self, event):
"""Callback for object:text-selection-changed accessibility events."""
self._clearSyntheticWebSelection()
if self.utilities.isZombie(event.source):
msg = "WEB: Event source is Zombie"
debug.printMessage(debug.LEVEL_INFO, msg, True)
@@ -2895,6 +3028,22 @@ class Script(default.Script):
debug.printMessage(debug.LEVEL_INFO, msg, True)
return True
if self.utilities.lastInputEventWasCaretNavWithSelection() \
and not AXText.get_selected_ranges(event.source):
document = self.utilities.getTopLevelDocumentForObject(event.source)
obj, offset = self.utilities.getCaretContext(document, False, False)
focusOffset = AXText.get_caret_offset(event.source)
if self._presentSyntheticCaretSelection(
event,
document,
obj,
offset,
focusOverride=(event.source, focusOffset),
):
msg = "WEB: Event handled: synthetic selection from event source"
debug.printMessage(debug.LEVEL_INFO, msg, True)
return True
if AXUtilities.is_text_input(event.source) \
and AXUtilities.is_focused(event.source) \
and event.source != cthulhu_state.locusOfFocus:
@@ -2930,6 +3079,7 @@ class Script(default.Script):
msg = "WEB: Clearing command state"
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._clearSyntheticWebSelection()
self._lastCommandWasCaretNav = False
self._lastCommandWasStructNav = False
self._lastCommandWasMouseButton = False
+42 -16
View File
@@ -721,6 +721,34 @@ class Utilities(script_utilities.Utilities):
return prevobj, prevoffset
def getSyntheticSelection(self, obj=None):
obj = obj or cthulhu_state.locusOfFocus
state = self._script.pointOfReference.get("syntheticWebSelection")
if not state or not state.get("string"):
return None
if obj and self.inDocumentContent(obj):
document = self.getTopLevelDocumentForObject(obj)
if document != state.get("document"):
return None
return state
def allSelectedText(self, obj):
result = super().allSelectedText(obj)
if result[0] or not self.inDocumentContent(obj):
return result
state = self.getSyntheticSelection(obj)
if not state:
return result
string = state.get("string", "")
if not string:
return result
return [string, 0, len(string)]
def lastContext(self, root):
offset = 0
text = self.queryNonEmptyText(root)
@@ -2164,30 +2192,28 @@ class Utilities(script_utilities.Utilities):
if not self.inDocumentContent(obj):
return super().handleTextSelectionChange(obj)
oldStart, oldEnd = \
self._script.pointOfReference.get('selectionAnchorAndFocus', (None, None))
start, end = self._getSelectionAnchorAndFocus(obj)
self._script.pointOfReference['selectionAnchorAndFocus'] = (start, end)
adapter = self._getSelectionAdapter()
oldSnapshot = adapter.get_cached_snapshot(obj)
newSnapshot = adapter.refresh_snapshot(obj)
def _cmp(obj1, obj2):
return self.pathComparison(AXObject.get_path(obj1), AXObject.get_path(obj2))
oldSubtree = self._getSubtree(oldStart, oldEnd)
if start == oldStart and end == oldEnd:
descendants = oldSubtree
else:
newSubtree = self._getSubtree(start, end)
descendants = sorted(set(oldSubtree).union(newSubtree), key=functools.cmp_to_key(_cmp))
descendants = []
seen = set()
for descendant in adapter.get_selected_objects(oldSnapshot) + adapter.get_selected_objects(newSnapshot):
if hash(descendant) in seen:
continue
seen.add(hash(descendant))
descendants.append(descendant)
descendants.sort(key=functools.cmp_to_key(_cmp))
if not descendants:
return False
return super().handleTextSelectionChange(obj, speakMessage)
for descendant in descendants:
if descendant not in (oldStart, oldEnd, start, end) \
and AXObject.find_ancestor(descendant, lambda x: x in descendants):
super().updateCachedTextSelection(descendant)
else:
super().handleTextSelectionChange(descendant, speakMessage)
super().handleTextSelectionChange(descendant, speakMessage)
return True
+1 -1
View File
@@ -1673,7 +1673,7 @@ class SpeechGenerator(generator.Generator):
result = []
if AXObject.supports_text(obj):
selections = AXText.get_selected_ranges(obj)
selections = self._script.utilities.allTextSelections(obj)
if len(selections) == 1:
startOffset, endOffset = selections[0]
if startOffset == 0 and endOffset >= AXText.get_character_count(obj):
@@ -0,0 +1,472 @@
import sys
import unittest
from pathlib import Path
from unittest import mock
import gi
gi.require_version("Atspi", "2.0")
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from gi.repository import Atspi
from cthulhu import messages
from cthulhu.scripts import default
CONTROL_MASK = 1 << Atspi.ModifierType.CONTROL
class FakeKeyboardEvent:
def __init__(self, pressed, keyval_name, modifiers, obj, event_string=None):
self._pressed = pressed
self.keyval_name = keyval_name
self.modifiers = modifiers
self._obj = obj
self.event_string = keyval_name if event_string is None else event_string
def get_object(self):
return self._obj
def is_pressed_key(self):
return self._pressed
def run_repeating_timeout(_delay, callback, generation, max_checks=10):
for _ in range(max_checks):
if not callback(generation):
break
class DefaultScriptClipboardRegressionTests(unittest.TestCase):
def _make_script(self):
script = object.__new__(default.Script)
script.utilities = mock.Mock()
script.presentMessage = mock.Mock()
script.flatReviewPresenter = mock.Mock()
script.flatReviewPresenter.is_active.return_value = False
script._pendingKeyboardClipboardCommand = None
script._pendingKeyboardClipboardRelease = None
script._keyboardClipboardCommandGeneration = 0
script._sessionType = "x11"
return script
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch(
"cthulhu.scripts.default.GLib.timeout_add",
side_effect=lambda _delay, callback, generation: callback(generation),
)
def test_keyboard_copy_fallback_presents_message_when_callback_never_arrives(
self,
_timeout_add,
_is_terminal,
):
script = self._make_script()
source = object()
script.utilities.getClipboardContents.side_effect = ["before", "copied text"]
script.utilities.objectContentsAreInClipboard.return_value = True
script.updateKeyboardEventState(FakeKeyboardEvent(True, "c", CONTROL_MASK, source), None)
script.updateKeyboardEventState(FakeKeyboardEvent(False, "c", CONTROL_MASK, source), None)
script.utilities.objectContentsAreInClipboard.assert_called_once_with(source)
script.presentMessage.assert_called_once_with(
messages.CLIPBOARD_COPIED_FULL,
messages.CLIPBOARD_COPIED_BRIEF,
)
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch(
"cthulhu.scripts.default.GLib.timeout_add",
side_effect=run_repeating_timeout,
)
def test_keyboard_copy_fallback_uses_captured_selection_when_clipboard_stays_empty(
self,
_timeout_add,
_is_terminal,
):
script = self._make_script()
source = object()
script.utilities.getClipboardContents.side_effect = ["before"] + [""] * 10
script.utilities.allSelectedText.return_value = ("The quick brown", 0, 15)
script.utilities.selectedText.return_value = ("The quick brown", 0, 15)
script.utilities.objectContentsAreInClipboard.return_value = False
script.utilities.isInActiveApp.return_value = False
script.updateKeyboardEventState(FakeKeyboardEvent(True, "c", CONTROL_MASK, source), None)
script.updateKeyboardEventState(FakeKeyboardEvent(False, "c", CONTROL_MASK, source), None)
script.presentMessage.assert_called_once_with(
messages.CLIPBOARD_COPIED_FULL,
messages.CLIPBOARD_COPIED_BRIEF,
)
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch(
"cthulhu.scripts.default.GLib.timeout_add",
side_effect=lambda _delay, callback, generation: callback(generation),
)
def test_keyboard_copy_fallback_handles_release_before_press_order(
self,
_timeout_add,
_is_terminal,
):
script = self._make_script()
source = object()
script.utilities.getClipboardContents.side_effect = ["before", ""]
script.utilities.allSelectedText.return_value = ("3086180", 0, 7)
script.utilities.selectedText.return_value = ("3086180", 0, 7)
script.utilities.objectContentsAreInClipboard.return_value = False
script.utilities.isInActiveApp.return_value = False
script.updateKeyboardEventState(
FakeKeyboardEvent(False, "0x3", CONTROL_MASK, source, "\x03"),
None,
)
script.updateKeyboardEventState(
FakeKeyboardEvent(True, "0x3", CONTROL_MASK, source, "\x03"),
None,
)
script.presentMessage.assert_called_once_with(
messages.CLIPBOARD_COPIED_FULL,
messages.CLIPBOARD_COPIED_BRIEF,
)
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch("cthulhu.scripts.default.GLib.idle_add")
@mock.patch("cthulhu.scripts.default.GLib.timeout_add")
def test_wayland_keyboard_copy_is_presented_immediately(
self,
timeout_add,
idle_add,
_is_terminal,
):
script = self._make_script()
script._sessionType = "wayland"
source = object()
script.utilities.getClipboardContents.return_value = ""
script.utilities.allSelectedText.return_value = ("The quick brown", 0, 15)
script.utilities.selectedText.return_value = ("The quick brown", 0, 15)
script.utilities.objectContentsAreInClipboard.return_value = False
script.utilities.isInActiveApp.return_value = False
script.updateKeyboardEventState(FakeKeyboardEvent(True, "c", CONTROL_MASK, source), None)
timeout_add.assert_not_called()
idle_add.assert_called_once()
script.presentMessage.assert_not_called()
callback, generation = idle_add.call_args.args
callback(generation)
script.presentMessage.assert_called_once_with(
messages.CLIPBOARD_COPIED_FULL,
messages.CLIPBOARD_COPIED_BRIEF,
)
script.updateKeyboardEventState(FakeKeyboardEvent(False, "c", CONTROL_MASK, source), None)
script.presentMessage.assert_called_once_with(
messages.CLIPBOARD_COPIED_FULL,
messages.CLIPBOARD_COPIED_BRIEF,
)
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch("cthulhu.scripts.default.GLib.idle_add")
@mock.patch("cthulhu.scripts.default.GLib.timeout_add")
def test_wayland_keyboard_copy_handles_release_before_press_order_immediately(
self,
timeout_add,
idle_add,
_is_terminal,
):
script = self._make_script()
script._sessionType = "wayland"
source = object()
script.utilities.getClipboardContents.return_value = ""
script.utilities.allSelectedText.return_value = ("3086180", 0, 7)
script.utilities.selectedText.return_value = ("3086180", 0, 7)
script.utilities.objectContentsAreInClipboard.return_value = False
script.utilities.isInActiveApp.return_value = False
script.updateKeyboardEventState(
FakeKeyboardEvent(False, "0x3", CONTROL_MASK, source, "\x03"),
None,
)
script.updateKeyboardEventState(
FakeKeyboardEvent(True, "0x3", CONTROL_MASK, source, "\x03"),
None,
)
timeout_add.assert_not_called()
idle_add.assert_called_once()
script.presentMessage.assert_not_called()
callback, generation = idle_add.call_args.args
callback(generation)
script.presentMessage.assert_called_once_with(
messages.CLIPBOARD_COPIED_FULL,
messages.CLIPBOARD_COPIED_BRIEF,
)
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch(
"cthulhu.scripts.default.GLib.timeout_add",
side_effect=lambda _delay, callback, generation: callback(generation),
)
def test_keyboard_copy_fallback_recognizes_control_character_key_name(
self,
_timeout_add,
_is_terminal,
):
script = self._make_script()
source = object()
script.utilities.getClipboardContents.side_effect = ["before", "copied text"]
script.utilities.objectContentsAreInClipboard.return_value = True
script.updateKeyboardEventState(
FakeKeyboardEvent(True, "0x3", CONTROL_MASK, source, "\x03"),
None,
)
script.updateKeyboardEventState(
FakeKeyboardEvent(False, "0x3", CONTROL_MASK, source, "\x03"),
None,
)
script.utilities.objectContentsAreInClipboard.assert_called_once_with(source)
script.presentMessage.assert_called_once_with(
messages.CLIPBOARD_COPIED_FULL,
messages.CLIPBOARD_COPIED_BRIEF,
)
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch(
"cthulhu.scripts.default.GLib.timeout_add",
side_effect=lambda _delay, callback, generation: callback(generation),
)
def test_keyboard_copy_fallback_uses_captured_source_not_current_focus(
self,
_timeout_add,
_is_terminal,
):
script = self._make_script()
source = object()
script.utilities.getClipboardContents.side_effect = ["before", "copied text"]
script.utilities.objectContentsAreInClipboard.return_value = True
script.updateKeyboardEventState(FakeKeyboardEvent(True, "c", CONTROL_MASK, source), None)
script.updateKeyboardEventState(FakeKeyboardEvent(False, "c", CONTROL_MASK, object()), None)
script.utilities.objectContentsAreInClipboard.assert_called_once_with(source)
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch(
"cthulhu.scripts.default.GLib.timeout_add",
side_effect=lambda _delay, callback, generation: callback(generation),
)
def test_keyboard_copy_fallback_stays_silent_if_clipboard_is_unchanged(
self,
_timeout_add,
_is_terminal,
):
script = self._make_script()
source = object()
script.utilities.getClipboardContents.side_effect = ["same text", "same text"]
script.utilities.objectContentsAreInClipboard.return_value = False
script.utilities.isInActiveApp.return_value = True
script.updateKeyboardEventState(FakeKeyboardEvent(True, "c", CONTROL_MASK, source), None)
script.updateKeyboardEventState(FakeKeyboardEvent(False, "c", CONTROL_MASK, source), None)
self.assertEqual(
script.utilities.objectContentsAreInClipboard.call_args_list,
[mock.call(source), mock.call()],
)
script.presentMessage.assert_not_called()
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch(
"cthulhu.scripts.default.GLib.timeout_add",
side_effect=lambda _delay, callback, generation: callback(generation),
)
def test_keyboard_copy_fallback_presents_when_clipboard_unchanged_but_matches_source(
self,
_timeout_add,
_is_terminal,
):
script = self._make_script()
source = object()
script.utilities.getClipboardContents.side_effect = ["same text", "same text"]
script.utilities.objectContentsAreInClipboard.return_value = True
script.updateKeyboardEventState(FakeKeyboardEvent(True, "c", CONTROL_MASK, source), None)
script.updateKeyboardEventState(FakeKeyboardEvent(False, "c", CONTROL_MASK, source), None)
script.utilities.objectContentsAreInClipboard.assert_called_once_with(source)
script.presentMessage.assert_called_once_with(
messages.CLIPBOARD_COPIED_FULL,
messages.CLIPBOARD_COPIED_BRIEF,
)
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch(
"cthulhu.scripts.default.GLib.timeout_add",
side_effect=lambda _delay, callback, generation: callback(generation),
)
def test_keyboard_copy_fallback_uses_current_focus_when_captured_source_no_longer_matches(
self,
_timeout_add,
_is_terminal,
):
script = self._make_script()
source = object()
script.utilities.getClipboardContents.side_effect = ["before", "copied text"]
script.utilities.objectContentsAreInClipboard.side_effect = [False, True]
script.utilities.isInActiveApp.return_value = True
script.updateKeyboardEventState(FakeKeyboardEvent(True, "c", CONTROL_MASK, source), None)
script.updateKeyboardEventState(FakeKeyboardEvent(False, "c", CONTROL_MASK, source), None)
self.assertEqual(
script.utilities.objectContentsAreInClipboard.call_args_list,
[mock.call(source), mock.call()],
)
script.presentMessage.assert_called_once_with(
messages.CLIPBOARD_COPIED_FULL,
messages.CLIPBOARD_COPIED_BRIEF,
)
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch(
"cthulhu.scripts.default.GLib.timeout_add",
side_effect=lambda _delay, callback, generation: callback(generation),
)
def test_keyboard_copy_fallback_does_not_probe_current_focus_when_source_left_active_app(
self,
_timeout_add,
_is_terminal,
):
script = self._make_script()
source = object()
script.utilities.getClipboardContents.side_effect = ["before", "copied text"]
script.utilities.objectContentsAreInClipboard.return_value = False
script.utilities.isInActiveApp.return_value = False
script.updateKeyboardEventState(FakeKeyboardEvent(True, "c", CONTROL_MASK, source), None)
script.updateKeyboardEventState(FakeKeyboardEvent(False, "c", CONTROL_MASK, source), None)
script.utilities.objectContentsAreInClipboard.assert_called_once_with(source)
script.presentMessage.assert_not_called()
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch(
"cthulhu.scripts.default.GLib.timeout_add",
side_effect=run_repeating_timeout,
)
def test_keyboard_copy_fallback_retries_until_clipboard_match_appears(
self,
_timeout_add,
_is_terminal,
):
script = self._make_script()
source = object()
script.utilities.getClipboardContents.side_effect = ["before", "before", "copied text"]
script.utilities.objectContentsAreInClipboard.side_effect = [False, True]
script.utilities.isInActiveApp.return_value = False
script.updateKeyboardEventState(FakeKeyboardEvent(True, "c", CONTROL_MASK, source), None)
script.updateKeyboardEventState(FakeKeyboardEvent(False, "c", CONTROL_MASK, source), None)
self.assertEqual(
script.utilities.objectContentsAreInClipboard.call_args_list,
[mock.call(source), mock.call(source)],
)
script.presentMessage.assert_called_once_with(
messages.CLIPBOARD_COPIED_FULL,
messages.CLIPBOARD_COPIED_BRIEF,
)
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch(
"cthulhu.scripts.default.GLib.timeout_add",
side_effect=lambda _delay, callback, generation: callback(generation),
)
def test_keyboard_copy_fallback_uses_captured_selected_text_when_live_selection_clears(
self,
_timeout_add,
_is_terminal,
):
script = self._make_script()
source = object()
script.utilities.getClipboardContents.side_effect = ["before", "copied text"]
script.utilities.allSelectedText.return_value = ("copied text", 0, 11)
script.utilities.selectedText.return_value = ("", 0, 0)
script.utilities.objectContentsAreInClipboard.return_value = False
script.utilities.isInActiveApp.return_value = False
script.updateKeyboardEventState(FakeKeyboardEvent(True, "c", CONTROL_MASK, source), None)
script.updateKeyboardEventState(FakeKeyboardEvent(False, "c", CONTROL_MASK, source), None)
script.presentMessage.assert_called_once_with(
messages.CLIPBOARD_COPIED_FULL,
messages.CLIPBOARD_COPIED_BRIEF,
)
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch(
"cthulhu.scripts.default.GLib.timeout_add",
side_effect=lambda _delay, callback, generation: callback(generation),
)
def test_clipboard_callback_after_fallback_does_not_duplicate_message(
self,
_timeout_add,
_is_terminal,
):
script = self._make_script()
source = object()
script.utilities.getClipboardContents.side_effect = ["before", "copied text", "copied text"]
script.utilities.objectContentsAreInClipboard.return_value = True
script.utilities.topLevelObjectIsActiveAndCurrent.return_value = True
script.utilities.lastInputEventWasCopy.return_value = True
script.updateKeyboardEventState(FakeKeyboardEvent(True, "c", CONTROL_MASK, source), None)
script.updateKeyboardEventState(FakeKeyboardEvent(False, "c", CONTROL_MASK, source), None)
script.onClipboardContentsChanged()
script.presentMessage.assert_called_once_with(
messages.CLIPBOARD_COPIED_FULL,
messages.CLIPBOARD_COPIED_BRIEF,
)
@mock.patch("cthulhu.scripts.default.AXUtilities.is_terminal", return_value=False)
@mock.patch(
"cthulhu.scripts.default.GLib.timeout_add",
side_effect=lambda _delay, callback, generation: callback(generation),
)
def test_clipboard_callback_without_match_does_not_cancel_keyboard_fallback(
self,
_timeout_add,
_is_terminal,
):
script = self._make_script()
source = object()
script.utilities.getClipboardContents.side_effect = ["before", "copied text", "copied text"]
script.utilities.objectContentsAreInClipboard.side_effect = [False, False, True]
script.utilities.isInActiveApp.return_value = False
script.updateKeyboardEventState(FakeKeyboardEvent(True, "c", CONTROL_MASK, source), None)
script.onClipboardContentsChanged()
script.updateKeyboardEventState(FakeKeyboardEvent(False, "c", CONTROL_MASK, source), None)
self.assertEqual(
script.utilities.objectContentsAreInClipboard.call_args_list,
[mock.call(source), mock.call(), mock.call(source)],
)
script.presentMessage.assert_called_once_with(
messages.CLIPBOARD_COPIED_FULL,
messages.CLIPBOARD_COPIED_BRIEF,
)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,223 @@
import sys
import types
import unittest
from contextlib import ExitStack
from pathlib import Path
from unittest import mock
import gi
gi.require_version("Gdk", "3.0")
gi.require_version("Gtk", "3.0")
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
sys.modules.setdefault(
"cthulhu.cthulhu_i18n",
types.SimpleNamespace(
C_=lambda _context, message: message,
_=lambda message: message,
ngettext=lambda singular, plural, count: singular if count == 1 else plural,
),
)
sys.modules.setdefault(
"cthulhu.cthulhu_platform",
types.SimpleNamespace(
version="test",
revision="",
tablesdir="",
datadir="",
),
)
from cthulhu.script_utilities import Utilities
class FakeDocument:
def __init__(self):
self.currentSelections = []
self.setCalls = []
class FakeTextObject:
def __init__(self, name, text, document):
self.name = name
self.text = text
self.document = document
self.next_obj = None
self.prev_obj = None
self.caret_offset = 0
def __repr__(self):
return f"<FakeTextObject {self.name}>"
class DocumentSelectionRegressionTests(unittest.TestCase):
def setUp(self):
self.document = FakeDocument()
self.first = FakeTextObject("first", "alpha", self.document)
self.second = FakeTextObject("second", "beta", self.document)
self.first.next_obj = self.second
self.second.prev_obj = self.first
self.script = types.SimpleNamespace(
pointOfReference={},
sayPhrase=mock.Mock(),
speakMessage=mock.Mock(),
)
self.utility = object.__new__(Utilities)
self.utility._script = self.script
self.utility._clipboardHandlerId = None
self.utility.expandEOCs = lambda obj, start=0, end=-1: obj.text[start:len(obj.text) if end == -1 else end]
self.utility.queryNonEmptyText = lambda obj: obj if getattr(obj, "text", "") else None
self.utility.findNextObject = lambda obj: getattr(obj, "next_obj", None)
self.utility.findPreviousObject = lambda obj: getattr(obj, "prev_obj", None)
self.utility.getDocumentForObject = lambda obj: self.document if isinstance(obj, FakeTextObject) else obj if obj is self.document else None
self.utility.isSpreadSheetCell = lambda obj: False
@staticmethod
def _make_selection(start_object, start_offset, end_object, end_offset, start_is_active=False):
return types.SimpleNamespace(
start_object=start_object,
start_offset=start_offset,
end_object=end_object,
end_offset=end_offset,
start_is_active=start_is_active,
)
@staticmethod
def _new_text_selection():
return types.SimpleNamespace(
start_object=None,
start_offset=0,
end_object=None,
end_offset=0,
start_is_active=False,
)
def _patch_environment(self, local_ranges=None):
local_ranges = local_ranges or {}
remove_selection = mock.Mock()
set_selected_text = mock.Mock()
def get_selected_ranges(obj):
return list(local_ranges.get(obj, []))
def get_character_count(obj):
return len(obj.text)
def get_caret_offset(obj):
return obj.caret_offset
def get_substring(obj, start, end):
if end == -1:
end = len(obj.text)
return obj.text[start:end]
def get_text_selections(document):
return list(document.currentSelections)
def set_text_selections(document, selections):
document.setCalls.append(list(selections))
document.currentSelections = list(selections)
return True
stack = ExitStack()
stack.enter_context(mock.patch("cthulhu.script_utilities.AXObject.supports_text", side_effect=lambda obj: isinstance(obj, FakeTextObject)))
stack.enter_context(mock.patch("cthulhu.script_utilities.AXText.get_selected_ranges", side_effect=get_selected_ranges))
stack.enter_context(mock.patch("cthulhu.script_utilities.AXText.get_character_count", side_effect=get_character_count))
stack.enter_context(mock.patch("cthulhu.script_utilities.AXText.get_caret_offset", side_effect=get_caret_offset))
stack.enter_context(mock.patch("cthulhu.script_utilities.AXText.get_substring", side_effect=get_substring))
stack.enter_context(mock.patch("cthulhu.script_utilities.AXText.set_selected_text", set_selected_text))
stack.enter_context(mock.patch("cthulhu.script_utilities.AXText._get_n_selections", return_value=1))
stack.enter_context(mock.patch("cthulhu.script_utilities.AXText._remove_selection", remove_selection))
stack.enter_context(mock.patch("cthulhu.script_utilities.Atspi.Document.get_text_selections", side_effect=get_text_selections))
stack.enter_context(mock.patch("cthulhu.script_utilities.Atspi.Document.set_text_selections", side_effect=set_text_selections))
stack.enter_context(mock.patch("cthulhu.script_utilities.Atspi.TextSelection", side_effect=self._new_text_selection))
stack.enter_context(
mock.patch(
"cthulhu.script_utilities.cthulhu.cthulhuApp",
new=types.SimpleNamespace(
settingsManager=types.SimpleNamespace(
getSetting=lambda *_args, **_kwargs: False,
)
),
create=True,
)
)
self.addCleanup(stack.close)
return remove_selection, set_selected_text
def test_selected_text_uses_document_selection_when_local_ranges_are_empty(self):
self.document.currentSelections = [
self._make_selection(self.first, 2, self.second, 2),
]
self._patch_environment()
self.assertEqual(self.utility.selectedText(self.first), ["pha", 2, 5])
self.assertEqual(self.utility.selectedText(self.second), ["be", 0, 2])
def test_all_text_selections_project_document_ranges_onto_each_object(self):
self.document.currentSelections = [
self._make_selection(self.first, 2, self.second, 2),
]
self._patch_environment()
self.assertEqual(self.utility.allTextSelections(self.first), [(2, 5)])
self.assertEqual(self.utility.allTextSelections(self.second), [(0, 2)])
def test_all_selected_text_aggregates_multi_object_document_selection(self):
self.document.currentSelections = [
self._make_selection(self.first, 2, self.second, 2),
]
self._patch_environment()
self.assertEqual(self.utility.allSelectedText(self.first), ("pha be", 2, 5))
def test_clear_text_selection_uses_document_api_when_selection_is_authoritative(self):
self.document.currentSelections = [
self._make_selection(self.first, 1, self.second, 3),
]
remove_selection, _ = self._patch_environment()
self.utility.clearTextSelection(self.first)
self.assertEqual(self.document.setCalls, [[]])
remove_selection.assert_not_called()
def test_adjust_text_selection_preserves_other_document_ranges(self):
self.document.currentSelections = [
self._make_selection(self.first, 2, self.first, 4),
self._make_selection(self.second, 1, self.second, 3),
]
self.first.caret_offset = 4
_, set_selected_text = self._patch_environment()
self.utility.adjustTextSelection(self.first, 5)
self.assertEqual(len(self.document.setCalls), 1)
stored = [
(selection.start_object, selection.start_offset,
selection.end_object, selection.end_offset,
selection.start_is_active)
for selection in self.document.setCalls[0]
]
self.assertEqual(
stored,
[
(self.first, 2, self.first, 5, False),
(self.second, 1, self.second, 3, False),
],
)
set_selected_text.assert_not_called()
def test_selected_text_falls_back_to_raw_substring_when_expand_eocs_is_empty(self):
self.document.currentSelections = []
self.utility.expandEOCs = lambda _obj, _start=0, _end=-1: ""
self._patch_environment(local_ranges={self.first: [(2, 5)]})
self.assertEqual(self.utility.selectedText(self.first), ["pha", 2, 5])
if __name__ == "__main__":
unittest.main()
+129
View File
@@ -20,7 +20,9 @@ if soundGeneratorModule is not None and not hasattr(soundGeneratorModule, "Sound
soundGeneratorModule.SoundGenerator = _StubSoundGenerator
from cthulhu import messages
from cthulhu.scripts.web import script as web_script
from cthulhu.scripts.web import script_utilities as web_script_utilities
class WebKeyGrabRegressionTests(unittest.TestCase):
@@ -100,3 +102,130 @@ class WebKeyGrabRegressionTests(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class WebSelectionRegressionTests(unittest.TestCase):
def _make_script(self):
testScript = web_script.Script.__new__(web_script.Script)
testScript.pointOfReference = {}
testScript.utilities = mock.Mock()
testScript.utilities.sanityCheckActiveWindow = mock.Mock()
testScript.utilities.isZombie.return_value = False
testScript.utilities.eventIsBrowserUINoise.return_value = False
testScript.utilities.eventIsBrowserUIAutocompleteNoise.return_value = False
testScript.utilities.lastInputEventWasTab.return_value = False
testScript.utilities.inFindContainer.return_value = False
testScript.utilities.eventIsFromLocusOfFocusDocument.return_value = True
testScript.utilities.textEventIsDueToInsertion.return_value = False
testScript.utilities.textEventIsDueToDeletion.return_value = False
testScript.utilities.isItemForEditableComboBox.return_value = False
testScript.utilities.eventIsAutocompleteNoise.return_value = False
testScript.utilities.eventIsSpinnerNoise.return_value = False
testScript.utilities.caretMovedOutsideActiveGrid.return_value = False
testScript.utilities.treatEventAsSpinnerValueChange.return_value = False
testScript.utilities.queryNonEmptyText.return_value = "Darkest Dungeon"
testScript.utilities.lastInputEventWasCaretNavWithSelection.return_value = True
testScript.utilities.getCharacterContentsAtOffset.return_value = [[None, 0, 1, "D"]]
testScript.utilities.pathComparison.return_value = -1
testScript.updateBraille = mock.Mock()
testScript.speakContents = mock.Mock()
testScript.speakMessage = mock.Mock()
testScript._lastCommandWasCaretNav = True
testScript._lastCommandWasStructNav = False
testScript._lastCommandWasMouseButton = False
testScript._lastMouseButtonContext = (None, -1)
testScript._inFocusMode = False
testScript._focusModeIsSticky = False
testScript._browseModeIsSticky = False
return testScript
def test_broken_web_caret_selection_synthesizes_selection_from_caret_context(self):
testScript = self._make_script()
document = object()
section = object()
event = mock.Mock(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
testScript.utilities.getTopLevelDocumentForObject.return_value = document
testScript.utilities.getCaretContext.return_value = (section, 0)
testScript.utilities.nextContext.return_value = (section, 1)
testScript.utilities.getCharacterContentsAtOffset.return_value = [[section, 0, 1, "D"]]
with (
mock.patch.object(web_script.cthulhu_state, "locusOfFocus", section),
mock.patch.object(web_script.input_event_manager, "get_manager", return_value=manager),
mock.patch.object(web_script.cthulhu, "setLocusOfFocus") as setLocusOfFocus,
):
result = web_script.Script.onCaretMoved(testScript, event)
self.assertTrue(result)
testScript.utilities.setCaretContext.assert_called_once_with(section, 1, document)
setLocusOfFocus.assert_called_once_with(event, section, False, True)
testScript.speakContents.assert_called_once_with([[section, 0, 1, "D"]])
testScript.speakMessage.assert_called_once_with(messages.TEXT_SELECTED, interrupt=False)
self.assertEqual(
testScript.pointOfReference["syntheticWebSelection"]["string"],
"D",
)
def test_broken_web_text_selection_event_synthesizes_selection_from_event_source(self):
testScript = self._make_script()
document = object()
anchor = object()
link = object()
event = mock.Mock(source=link)
testScript.utilities.getTopLevelDocumentForObject.return_value = document
testScript.utilities.getCaretContext.return_value = (anchor, 0)
testScript.utilities.nextContext.side_effect = [(anchor, 0)]
testScript.utilities.getCharacterContentsAtOffset.side_effect = (
lambda obj, _offset: [[link, 0, 1, "S"]] if obj is link else []
)
testScript.utilities.textEventIsForNonNavigableTextObject.return_value = False
testScript.utilities.isContentEditableWithEmbeddedObjects.return_value = False
testScript._compareCaretContexts = mock.Mock(return_value=1)
with (
mock.patch.object(web_script.cthulhu_state, "locusOfFocus", anchor),
mock.patch.object(web_script.AXText, "get_caret_offset", return_value=0),
mock.patch.object(web_script.AXText, "get_selected_ranges", return_value=[]),
mock.patch.object(web_script.AXText, "get_substring", return_value="S"),
mock.patch.object(web_script.cthulhu, "setLocusOfFocus") as setLocusOfFocus,
):
result = web_script.Script.onTextSelectionChanged(testScript, event)
self.assertTrue(result)
testScript.utilities.setCaretContext.assert_called_once_with(link, 0, document)
setLocusOfFocus.assert_called_once_with(event, link, False, True)
testScript.speakContents.assert_called_once_with([[link, 0, 1, "S"]])
testScript.speakMessage.assert_called_once_with(messages.TEXT_SELECTED, interrupt=False)
self.assertEqual(
testScript.pointOfReference["syntheticWebSelection"]["string"],
"S",
)
def test_web_utilities_all_selected_text_falls_back_to_synthetic_selection(self):
testScript = mock.Mock(pointOfReference={})
utilities = web_script_utilities.Utilities.__new__(web_script_utilities.Utilities)
utilities._script = testScript
document = object()
obj = object()
testScript.pointOfReference["syntheticWebSelection"] = {
"document": document,
"string": "Dark",
}
with (
mock.patch.object(
web_script_utilities.script_utilities.Utilities,
"allSelectedText",
return_value=["", 0, 0],
),
mock.patch.object(utilities, "inDocumentContent", return_value=True),
mock.patch.object(utilities, "getTopLevelDocumentForObject", return_value=document),
):
result = web_script_utilities.Utilities.allSelectedText(utilities, obj)
self.assertEqual(result, ["Dark", 0, 4])