7 Commits

32 changed files with 1656 additions and 730 deletions
+27 -5
View File
@@ -196,6 +196,8 @@ class Parser(argparse.ArgumentParser):
help=messages.CLI_DEBUG_FILE, metavar=messages.CLI_DEBUG_FILE_NAME) help=messages.CLI_DEBUG_FILE, metavar=messages.CLI_DEBUG_FILE_NAME)
self.add_argument( self.add_argument(
"--debug", action="store_true", help=messages.CLI_ENABLE_DEBUG) "--debug", action="store_true", help=messages.CLI_ENABLE_DEBUG)
self.add_argument(
"--debug-startup", action="store_true", help=messages.CLI_ENABLE_STARTUP_DEBUG)
self._optionals.title = messages.CLI_OPTIONAL_ARGUMENTS self._optionals.title = messages.CLI_OPTIONAL_ARGUMENTS
@@ -323,6 +325,8 @@ def cleanup(sigval):
time.sleep(0.5) time.sleep(0.5)
def main(): def main():
debug.reset_startup_timing()
setProcessName('cthulhu') setProcessName('cthulhu')
parser = Parser() parser = Parser()
@@ -333,16 +337,26 @@ def main():
debug.eventDebugLevel = debug.LEVEL_OFF debug.eventDebugLevel = debug.LEVEL_OFF
debug.debugFile = open(args.debug_file, 'w') debug.debugFile = open(args.debug_file, 'w')
debug.set_startup_timing_enabled(args.debug or args.debug_startup)
debug.print_startup_timing("launcher main start")
if args.replace: if args.replace:
with debug.startup_timing("replace cleanup"):
cleanup(signal.SIGKILL) cleanup(signal.SIGKILL)
settingsDict = getattr(args, 'settings', {}) settingsDict = getattr(args, 'settings', {})
if not inGraphicalDesktop(): with debug.startup_timing("graphical desktop check"):
graphicalDesktop = inGraphicalDesktop()
if not graphicalDesktop:
print(messages.CLI_NO_DESKTOP_ERROR) print(messages.CLI_NO_DESKTOP_ERROR)
return 1 return 1
with debug.startup_timing("session check"):
sessionType = getSessionType() sessionType = getSessionType()
xServerVendor = None
if sessionType == "x11":
xServerVendor = getXServerVendor()
sessionDetails = [] sessionDetails = []
xdgSessionType = os.environ.get("XDG_SESSION_TYPE") xdgSessionType = os.environ.get("XDG_SESSION_TYPE")
if xdgSessionType: if xdgSessionType:
@@ -355,9 +369,8 @@ def main():
display = os.environ.get("DISPLAY") display = os.environ.get("DISPLAY")
if display: if display:
sessionDetails.append(f"DISPLAY={display}") sessionDetails.append(f"DISPLAY={display}")
vendor = getXServerVendor() if xServerVendor:
if vendor: sessionDetails.append(f"X server vendor={xServerVendor}")
sessionDetails.append(f"X server vendor={vendor}")
if sessionDetails: if sessionDetails:
msg = f"INFO: Session: {sessionType} ({', '.join(sessionDetails)})" msg = f"INFO: Session: {sessionType} ({', '.join(sessionDetails)})"
@@ -367,7 +380,9 @@ def main():
debug.printMessage(debug.LEVEL_INFO, "INFO: Preparing to launch.", True) debug.printMessage(debug.LEVEL_INFO, "INFO: Preparing to launch.", True)
debug.print_startup_timing("cthulhu import start")
from cthulhu import cthulhu from cthulhu import cthulhu
debug.print_startup_timing("cthulhu import complete")
manager = cthulhu.cthulhuApp.settingsManager manager = cthulhu.cthulhuApp.settingsManager
if not manager: if not manager:
@@ -375,10 +390,12 @@ def main():
return 1 return 1
debug.printMessage(debug.LEVEL_INFO, "INFO: About to activate settings manager.", True) debug.printMessage(debug.LEVEL_INFO, "INFO: About to activate settings manager.", True)
with debug.startup_timing("settings manager launcher activation"):
manager.activate(args.user_prefs, settingsDict) manager.activate(args.user_prefs, settingsDict)
sys.path.insert(0, manager.getPrefsDir()) sys.path.insert(0, manager.getPrefsDir())
if args.profile: if args.profile:
with debug.startup_timing("profile selection"):
try: try:
manager.setProfile(args.profile) manager.setProfile(args.profile)
except Exception: except Exception:
@@ -386,14 +403,19 @@ def main():
manager.setProfile() manager.setProfile()
if args.setup: if args.setup:
with debug.startup_timing("setup cleanup"):
cleanup(signal.SIGKILL) cleanup(signal.SIGKILL)
with debug.startup_timing("preferences GUI startup"):
cthulhu.showPreferencesGUI() cthulhu.showPreferencesGUI()
if otherCthulhus(): with debug.startup_timing("other instance check"):
otherInstances = otherCthulhus()
if otherInstances:
print(messages.CLI_OTHER_CTHULHUS_ERROR) print(messages.CLI_OTHER_CTHULHUS_ERROR)
return 1 return 1
debug.printMessage(debug.LEVEL_INFO, "INFO: About to launch Cthulhu.", True) debug.printMessage(debug.LEVEL_INFO, "INFO: About to launch Cthulhu.", True)
debug.print_startup_timing("cthulhu main handoff")
return cthulhu.main() return cthulhu.main()
if __name__ == "__main__": if __name__ == "__main__":
+79 -5
View File
@@ -402,20 +402,24 @@ def loadUserSettings(script: Optional[Any] = None, inputEvent: Optional[Any] = N
""" """
debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: Loading User Settings', True) debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: Loading User Settings', True)
debug.print_startup_timing("load user settings start")
global _userSettings global _userSettings
# Shutdown the output drivers and give them a chance to die. # Shutdown the output drivers and give them a chance to die.
with debug.startup_timing("output driver shutdown"):
player = sound.getPlayer() player = sound.getPlayer()
player.shutdown() player.shutdown()
speech.shutdown() speech.shutdown()
braille.shutdown() braille.shutdown()
with debug.startup_timing("script manager deactivate for settings load"):
cthulhuApp.scriptManager.deactivate() cthulhuApp.scriptManager.deactivate()
cthulhuApp.getSignalManager().emitSignal('load-setting-begin') cthulhuApp.getSignalManager().emitSignal('load-setting-begin')
reloaded: bool = False reloaded: bool = False
with debug.startup_timing("settings profile load"):
if _userSettings: if _userSettings:
_profile = cthulhuApp.settingsManager.getSetting('activeProfile')[1] _profile = cthulhuApp.settingsManager.getSetting('activeProfile')[1]
try: try:
@@ -438,11 +442,13 @@ def loadUserSettings(script: Optional[Any] = None, inputEvent: Optional[Any] = N
if not script: if not script:
script = cthulhuApp.scriptManager.get_default_script() script = cthulhuApp.scriptManager.get_default_script()
with debug.startup_timing("app settings load"):
cthulhuApp.settingsManager.loadAppSettings(script) cthulhuApp.settingsManager.loadAppSettings(script)
if cthulhuApp.settingsManager.getSetting('enableSpeech'): if cthulhuApp.settingsManager.getSetting('enableSpeech'):
msg = 'CTHULHU: About to enable speech' msg = 'CTHULHU: About to enable speech'
debug.printMessage(debug.LEVEL_INFO, msg, True) debug.printMessage(debug.LEVEL_INFO, msg, True)
with debug.startup_timing("speech initialization"):
try: try:
speech.init() speech.init()
if reloaded and not skipReloadMessage: if reloaded and not skipReloadMessage:
@@ -456,6 +462,7 @@ def loadUserSettings(script: Optional[Any] = None, inputEvent: Optional[Any] = N
if cthulhuApp.settingsManager.getSetting('enableBraille'): if cthulhuApp.settingsManager.getSetting('enableBraille'):
msg = 'CTHULHU: About to enable braille' msg = 'CTHULHU: About to enable braille'
debug.printMessage(debug.LEVEL_INFO, msg, True) debug.printMessage(debug.LEVEL_INFO, msg, True)
with debug.startup_timing("braille initialization"):
try: try:
braille.init(_processBrailleEvent) braille.init(_processBrailleEvent)
except Exception: except Exception:
@@ -467,17 +474,21 @@ def loadUserSettings(script: Optional[Any] = None, inputEvent: Optional[Any] = N
debug.printMessage(debug.LEVEL_INFO, msg, True) debug.printMessage(debug.LEVEL_INFO, msg, True)
with debug.startup_timing("mouse review update"):
if cthulhuApp.settingsManager.getSetting('enableMouseReview'): if cthulhuApp.settingsManager.getSetting('enableMouseReview'):
mouse_review.getReviewer().activate() mouse_review.getReviewer().activate()
else: else:
mouse_review.getReviewer().deactivate() mouse_review.getReviewer().deactivate()
if cthulhuApp.settingsManager.getSetting('enableSound'): if cthulhuApp.settingsManager.getSetting('enableSound'):
with debug.startup_timing("sound initialization"):
player.init() player.init()
with debug.startup_timing("cthulhu modifier refresh"):
cthulhu_modifier_manager.getManager().refreshCthulhuModifiers("Loading user settings.") cthulhu_modifier_manager.getManager().refreshCthulhuModifiers("Loading user settings.")
# Activate core systems FIRST before loading plugins # Activate core systems FIRST before loading plugins
with debug.startup_timing("core manager activation"):
cthulhuApp.compositorStateAdapter.activate() cthulhuApp.compositorStateAdapter.activate()
cthulhuApp.scriptManager.activate() cthulhuApp.scriptManager.activate()
cthulhuApp.eventManager.activate() cthulhuApp.eventManager.activate()
@@ -498,14 +509,17 @@ def loadUserSettings(script: Optional[Any] = None, inputEvent: Optional[Any] = N
) )
activePlugins = list(activePluginsSetting) activePlugins = list(activePluginsSetting)
debug.printMessage(debug.LEVEL_INFO, f'CTHULHU: Loading active plugins: {activePlugins}', True) debug.printMessage(debug.LEVEL_INFO, f'CTHULHU: Loading active plugins: {activePlugins}', True)
with debug.startup_timing("active plugin load"):
cthulhuApp.getPluginSystemManager().setActivePlugins(activePlugins) cthulhuApp.getPluginSystemManager().setActivePlugins(activePlugins)
# Refresh keybindings to include plugin bindings (after script manager is active) # Refresh keybindings to include plugin bindings (after script manager is active)
with debug.startup_timing("plugin keybinding refresh"):
cthulhuApp.getPluginSystemManager().refresh_active_script_keybindings() cthulhuApp.getPluginSystemManager().refresh_active_script_keybindings()
cthulhuApp.getSignalManager().emitSignal('load-setting-completed') cthulhuApp.getSignalManager().emitSignal('load-setting-completed')
debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: User Settings Loaded', True) debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: User Settings Loaded', True)
debug.print_startup_timing("load user settings complete")
return True return True
@@ -612,6 +626,7 @@ def init() -> bool:
""" """
debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: Initializing', True) debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: Initializing', True)
debug.print_startup_timing("cthulhu init start")
global _initialized global _initialized
@@ -627,8 +642,10 @@ def init() -> bool:
# Activate settings manager before loading user settings # Activate settings manager before loading user settings
debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: Activating settings manager', True) debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: Activating settings manager', True)
with debug.startup_timing("settings manager init activation"):
cthulhuApp.settingsManager.activate() cthulhuApp.settingsManager.activate()
with debug.startup_timing("load user settings"):
loadUserSettings() loadUserSettings()
if settings.timeoutCallback and (settings.timeoutTime > 0): if settings.timeoutCallback and (settings.timeoutTime > 0):
@@ -636,6 +653,7 @@ def init() -> bool:
_initialized = True _initialized = True
debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: Initialized', True) debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: Initialized', True)
debug.print_startup_timing("cthulhu init complete")
return True return True
@@ -672,6 +690,7 @@ def start() -> None:
"""Starts Cthulhu.""" """Starts Cthulhu."""
debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: Starting', True) debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: Starting', True)
debug.print_startup_timing("cthulhu start begin")
if not _initialized: if not _initialized:
init() init()
@@ -689,11 +708,13 @@ def start() -> None:
# Used to re-create the Xmodmap when a new keyboard is plugged in. # Used to re-create the Xmodmap when a new keyboard is plugged in.
# Necessary, because plugging in a new keyboard resets the Xmodmap # Necessary, because plugging in a new keyboard resets the Xmodmap
# and stomps our changes # and stomps our changes
with debug.startup_timing("input device monitor setup"):
display = Gdk.Display.get_default() display = Gdk.Display.get_default()
devmanager = display.get_device_manager() devmanager = display.get_device_manager()
devmanager.connect("device-added", deviceChangeHandler) devmanager.connect("device-added", deviceChangeHandler)
devmanager.connect("device-removed", deviceChangeHandler) devmanager.connect("device-removed", deviceChangeHandler)
with debug.startup_timing("desktop startup notification"):
Gdk.notify_startup_complete() Gdk.notify_startup_complete()
msg = 'CTHULHU: Startup complete notification made' msg = 'CTHULHU: Startup complete notification made'
debug.printMessage(debug.LEVEL_INFO, msg, True) debug.printMessage(debug.LEVEL_INFO, msg, True)
@@ -701,9 +722,11 @@ def start() -> None:
debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: Starting Atspi main event loop', True) debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: Starting Atspi main event loop', True)
# Start D-Bus remote controller service after ATSPI is ready # Start D-Bus remote controller service after ATSPI is ready
with debug.startup_timing("startup idle services"):
cthulhuApp.getMakoNotificationMonitor().start() cthulhuApp.getMakoNotificationMonitor().start()
GObject.idle_add(_start_dbus_service) GObject.idle_add(_start_dbus_service)
debug.print_startup_timing("AT-SPI main loop start")
Atspi.event_main() Atspi.event_main()
def die(exitCode: int = 1) -> None: def die(exitCode: int = 1) -> None:
@@ -851,7 +874,9 @@ def main() -> int:
if session: if session:
msg += f" session: {session}" msg += f" session: {session}"
debug.printMessage(debug.LEVEL_INFO, msg, True) debug.printMessage(debug.LEVEL_INFO, msg, True)
debug.print_startup_timing("cthulhu main start")
with debug.startup_timing("faulthandler setup"):
if debug.debugFile and os.path.exists(debug.debugFile.name): if debug.debugFile and os.path.exists(debug.debugFile.name):
faulthandler.enable(file=debug.debugFile, all_threads=True) faulthandler.enable(file=debug.debugFile, all_threads=True)
else: else:
@@ -863,6 +888,7 @@ def main() -> int:
# Various signal handlers we want to listen for. # Various signal handlers we want to listen for.
# #
with debug.startup_timing("signal handler setup"):
signal.signal(signal.SIGHUP, shutdownOnSignal) signal.signal(signal.SIGHUP, shutdownOnSignal)
signal.signal(signal.SIGINT, shutdownOnSignal) signal.signal(signal.SIGINT, shutdownOnSignal)
signal.signal(signal.SIGTERM, shutdownOnSignal) signal.signal(signal.SIGTERM, shutdownOnSignal)
@@ -870,14 +896,17 @@ def main() -> int:
signal.signal(signal.SIGSEGV, crashOnSignal) signal.signal(signal.SIGSEGV, crashOnSignal)
debug.printMessage(debug.LEVEL_INFO, "CTHULHU: Enabling accessibility (if needed).", True) debug.printMessage(debug.LEVEL_INFO, "CTHULHU: Enabling accessibility (if needed).", True)
with debug.startup_timing("accessibility enable check"):
if not cthulhuApp.settingsManager.isAccessibilityEnabled(): if not cthulhuApp.settingsManager.isAccessibilityEnabled():
cthulhuApp.settingsManager.setAccessibility(True) cthulhuApp.settingsManager.setAccessibility(True)
debug.printMessage(debug.LEVEL_INFO, "CTHULHU: Initializing.", True) debug.printMessage(debug.LEVEL_INFO, "CTHULHU: Initializing.", True)
with debug.startup_timing("cthulhu init"):
init() init()
debug.printMessage(debug.LEVEL_INFO, "CTHULHU: Initialized.", True) debug.printMessage(debug.LEVEL_INFO, "CTHULHU: Initialized.", True)
script = cthulhu_state.activeScript script = cthulhu_state.activeScript
with debug.startup_timing("startup sound"):
sound_theme_manager.getManager().playStartSound(wait=False) sound_theme_manager.getManager().playStartSound(wait=False)
soundFailureReason = sound.getSoundSystemFailureReason() soundFailureReason = sound.getSoundSystemFailureReason()
if soundFailureReason: if soundFailureReason:
@@ -886,6 +915,7 @@ def main() -> int:
f"CTHULHU: Startup sound failed. Continuing without sound. Reason: {soundFailureReason}", f"CTHULHU: Startup sound failed. Continuing without sound. Reason: {soundFailureReason}",
True True
) )
with debug.startup_timing("startup focus recovery"):
cthulhuApp.getSignalManager().emitSignal('start-application-completed') cthulhuApp.getSignalManager().emitSignal('start-application-completed')
if script: if script:
window = script.utilities.activeWindow() window = script.utilities.activeWindow()
@@ -910,6 +940,7 @@ def main() -> int:
try: try:
msg = "CTHULHU: Starting ATSPI registry." msg = "CTHULHU: Starting ATSPI registry."
debug.printMessage(debug.LEVEL_INFO, msg, True) debug.printMessage(debug.LEVEL_INFO, msg, True)
debug.print_startup_timing("AT-SPI registry start")
start() # waits until we stop the registry start() # waits until we stop the registry
except Exception: except Exception:
msg = "CTHULHU: Exception starting ATSPI registry." msg = "CTHULHU: Exception starting ATSPI registry."
@@ -930,27 +961,44 @@ class Cthulhu(GObject.Object):
"active-script-changed": (GObject.SignalFlags.RUN_LAST, None, (GObject.TYPE_PYOBJECT,)), # New signal to indicate active script change "active-script-changed": (GObject.SignalFlags.RUN_LAST, None, (GObject.TYPE_PYOBJECT,)), # New signal to indicate active script change
} }
def __init__(self) -> None: def __init__(self) -> None:
debug.print_startup_timing("cthulhuApp construction start")
GObject.Object.__init__(self) GObject.Object.__init__(self)
# add members # add members
with debug.startup_timing("resource manager construction"):
self.resourceManager: resource_manager.ResourceManager = resource_manager.ResourceManager(self) self.resourceManager: resource_manager.ResourceManager = resource_manager.ResourceManager(self)
self.settingsManager: SettingsManager = settings_manager.SettingsManager(self) # Directly instantiate with debug.startup_timing("settings manager construction"):
self.eventManager: EventManager = event_manager.EventManager(self) # Directly instantiate self.settingsManager: SettingsManager = settings_manager.SettingsManager(self)
self.compositorStateAdapter: CompositorStateAdapter = compositor_state_adapter.CompositorStateAdapter() with debug.startup_timing("event manager construction"):
self.eventManager: EventManager = event_manager.EventManager(self)
with debug.startup_timing("compositor adapter construction"):
self.compositorStateAdapter: CompositorStateAdapter = \
compositor_state_adapter.CompositorStateAdapter()
self.eventManager.set_compositor_state_adapter(self.compositorStateAdapter) self.eventManager.set_compositor_state_adapter(self.compositorStateAdapter)
self.scriptManager: ScriptManager = script_manager.ScriptManager(self) # Directly instantiate with debug.startup_timing("script manager construction"):
self.scriptManager: ScriptManager = script_manager.ScriptManager(self)
script_manager._manager = self.scriptManager script_manager._manager = self.scriptManager
self.logger: logger.Logger = logger.Logger() # Directly instantiate with debug.startup_timing("logger construction"):
self.logger: logger.Logger = logger.Logger()
with debug.startup_timing("signal manager construction"):
self.signalManager: SignalManager = signal_manager.SignalManager(self) self.signalManager: SignalManager = signal_manager.SignalManager(self)
with debug.startup_timing("dynamic API manager construction"):
self.dynamicApiManager: DynamicApiManager = dynamic_api_manager.DynamicApiManager(self) self.dynamicApiManager: DynamicApiManager = dynamic_api_manager.DynamicApiManager(self)
with debug.startup_timing("translation manager construction"):
self.translationManager: TranslationManager = translation_manager.TranslationManager(self) self.translationManager: TranslationManager = translation_manager.TranslationManager(self)
self.debugManager: Any = debug self.debugManager: Any = debug
with debug.startup_timing("API helper construction"):
self.APIHelper: APIHelper = APIHelper(self) self.APIHelper: APIHelper = APIHelper(self)
with debug.startup_timing("notification monitor construction"):
self.makoNotificationMonitor: mako_notification_monitor.MakoNotificationMonitor = \ self.makoNotificationMonitor: mako_notification_monitor.MakoNotificationMonitor = \
mako_notification_monitor.MakoNotificationMonitor(self) mako_notification_monitor.MakoNotificationMonitor(self)
with debug.startup_timing("compat API registration"):
self.createCompatAPI() self.createCompatAPI()
with debug.startup_timing("plugin manager construction"):
self.pluginSystemManager: PluginSystemManager = plugin_system_manager.PluginSystemManager(self) self.pluginSystemManager: PluginSystemManager = plugin_system_manager.PluginSystemManager(self)
# Scan for available plugins at startup # Scan for available plugins at startup
with debug.startup_timing("plugin scan"):
self.pluginSystemManager.rescanPlugins() self.pluginSystemManager.rescanPlugins()
debug.print_startup_timing("cthulhuApp construction complete")
def getAPIHelper(self) -> APIHelper: def getAPIHelper(self) -> APIHelper:
return self.APIHelper return self.APIHelper
@@ -994,6 +1042,32 @@ class Cthulhu(GObject.Object):
def getLogger(self) -> logger.Logger: # New getter for the logger def getLogger(self) -> logger.Logger: # New getter for the logger
return self.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]: def addKeyGrab(self, binding: Any) -> List[int]:
return addKeyGrab(binding) return addKeyGrab(binding)
+27 -5
View File
@@ -154,6 +154,8 @@ class Parser(argparse.ArgumentParser):
help=messages.CLI_DEBUG_FILE, metavar=messages.CLI_DEBUG_FILE_NAME) help=messages.CLI_DEBUG_FILE, metavar=messages.CLI_DEBUG_FILE_NAME)
self.add_argument( self.add_argument(
"--debug", action="store_true", help=messages.CLI_ENABLE_DEBUG) "--debug", action="store_true", help=messages.CLI_ENABLE_DEBUG)
self.add_argument(
"--debug-startup", action="store_true", help=messages.CLI_ENABLE_STARTUP_DEBUG)
self._optionals.title = messages.CLI_OPTIONAL_ARGUMENTS self._optionals.title = messages.CLI_OPTIONAL_ARGUMENTS
@@ -281,6 +283,8 @@ def cleanup(sigval):
time.sleep(0.5) time.sleep(0.5)
def main(): def main():
debug.reset_startup_timing()
setProcessName('cthulhu') setProcessName('cthulhu')
parser = Parser() parser = Parser()
@@ -291,16 +295,26 @@ def main():
debug.eventDebugLevel = debug.LEVEL_OFF debug.eventDebugLevel = debug.LEVEL_OFF
debug.debugFile = open(args.debug_file, 'w') debug.debugFile = open(args.debug_file, 'w')
debug.set_startup_timing_enabled(args.debug or args.debug_startup)
debug.print_startup_timing("launcher main start")
if args.replace: if args.replace:
with debug.startup_timing("replace cleanup"):
cleanup(signal.SIGKILL) cleanup(signal.SIGKILL)
settingsDict = getattr(args, 'settings', {}) settingsDict = getattr(args, 'settings', {})
if not inGraphicalDesktop(): with debug.startup_timing("graphical desktop check"):
graphicalDesktop = inGraphicalDesktop()
if not graphicalDesktop:
print(messages.CLI_NO_DESKTOP_ERROR) print(messages.CLI_NO_DESKTOP_ERROR)
return 1 return 1
with debug.startup_timing("session check"):
sessionType = getSessionType() sessionType = getSessionType()
xServerVendor = None
if sessionType == "x11":
xServerVendor = getXServerVendor()
sessionDetails = [] sessionDetails = []
xdgSessionType = os.environ.get("XDG_SESSION_TYPE") xdgSessionType = os.environ.get("XDG_SESSION_TYPE")
if xdgSessionType: if xdgSessionType:
@@ -313,9 +327,8 @@ def main():
display = os.environ.get("DISPLAY") display = os.environ.get("DISPLAY")
if display: if display:
sessionDetails.append(f"DISPLAY={display}") sessionDetails.append(f"DISPLAY={display}")
vendor = getXServerVendor() if xServerVendor:
if vendor: sessionDetails.append(f"X server vendor={xServerVendor}")
sessionDetails.append(f"X server vendor={vendor}")
if sessionDetails: if sessionDetails:
msg = f"INFO: Session: {sessionType} ({', '.join(sessionDetails)})" msg = f"INFO: Session: {sessionType} ({', '.join(sessionDetails)})"
@@ -325,7 +338,9 @@ def main():
debug.printMessage(debug.LEVEL_INFO, "INFO: Preparing to launch.", True) debug.printMessage(debug.LEVEL_INFO, "INFO: Preparing to launch.", True)
debug.print_startup_timing("cthulhu import start")
from cthulhu import cthulhu from cthulhu import cthulhu
debug.print_startup_timing("cthulhu import complete")
manager = cthulhu.cthulhuApp.settingsManager manager = cthulhu.cthulhuApp.settingsManager
if not manager: if not manager:
@@ -333,10 +348,12 @@ def main():
return 1 return 1
debug.printMessage(debug.LEVEL_INFO, "INFO: About to activate settings manager.", True) debug.printMessage(debug.LEVEL_INFO, "INFO: About to activate settings manager.", True)
with debug.startup_timing("settings manager launcher activation"):
manager.activate(args.user_prefs, settingsDict) manager.activate(args.user_prefs, settingsDict)
sys.path.insert(0, manager.getPrefsDir()) sys.path.insert(0, manager.getPrefsDir())
if args.profile: if args.profile:
with debug.startup_timing("profile selection"):
try: try:
manager.setProfile(args.profile) manager.setProfile(args.profile)
except Exception: except Exception:
@@ -344,14 +361,19 @@ def main():
manager.setProfile() manager.setProfile()
if args.setup: if args.setup:
with debug.startup_timing("setup cleanup"):
cleanup(signal.SIGKILL) cleanup(signal.SIGKILL)
with debug.startup_timing("preferences GUI startup"):
cthulhu.showPreferencesGUI() cthulhu.showPreferencesGUI()
if otherCthulhus(): with debug.startup_timing("other instance check"):
otherInstances = otherCthulhus()
if otherInstances:
print(messages.CLI_OTHER_CTHULHUS_ERROR) print(messages.CLI_OTHER_CTHULHUS_ERROR)
return 1 return 1
debug.printMessage(debug.LEVEL_INFO, "INFO: About to launch Cthulhu.", True) debug.printMessage(debug.LEVEL_INFO, "INFO: About to launch Cthulhu.", True)
debug.print_startup_timing("cthulhu main handoff")
return cthulhu.main() return cthulhu.main()
if __name__ == "__main__": if __name__ == "__main__":
+8 -5
View File
@@ -619,7 +619,7 @@ class CthulhuSetupGUI(cthulhu_gtkbuilder.GtkBuilderWrapper):
for plugin_info in sorted(plugin_infos, key=lambda item: (item.get_name() or item.get_module_name()).lower()): for plugin_info in sorted(plugin_infos, key=lambda item: (item.get_name() or item.get_module_name()).lower()):
plugin_name = plugin_info.get_module_name() plugin_name = plugin_info.get_module_name()
canonical_name = plugin_info.get_canonical_name() canonical_name = plugin_info.get_canonical_name()
if plugin_info.hidden or canonical_name == "PluginManager": if plugin_info.hidden:
continue continue
self._plugin_canonical_map[plugin_name] = canonical_name self._plugin_canonical_map[plugin_name] = canonical_name
@@ -3911,6 +3911,11 @@ print(json.dumps(result))
combobox.set_model(self.hardwareDeviceModel) combobox.set_model(self.hardwareDeviceModel)
self._setHardwareDeviceChoice(saved_device) self._setHardwareDeviceChoice(saved_device)
@staticmethod
def _isHardwareSpeechFactory(factory):
moduleName = getattr(factory, "__name__", "")
return moduleName.split(".")[-1] == "hardwarefactory"
def _setHardwareDeviceChoice(self, device_name): def _setHardwareDeviceChoice(self, device_name):
"""Set the active item in the hardware device combo box. """Set the active item in the hardware device combo box.
@@ -3941,9 +3946,7 @@ print(json.dumps(result))
is_hardware = False is_hardware = False
if self.speechSystemsChoice: if self.speechSystemsChoice:
try: try:
is_hardware = ( is_hardware = self._isHardwareSpeechFactory(self.speechSystemsChoice)
self.speechSystemsChoice.__name__ == "hardwarefactory"
)
except Exception: except Exception:
pass pass
@@ -5052,7 +5055,7 @@ print(json.dumps(result))
# Save hardware speech device setting when hardware factory is active # Save hardware speech device setting when hardware factory is active
if self.speechSystemsChoice and \ if self.speechSystemsChoice and \
self.speechSystemsChoice.__name__ == "hardwarefactory": self._isHardwareSpeechFactory(self.speechSystemsChoice):
if self.hardwareDeviceChoice is not None: if self.hardwareDeviceChoice is not None:
self.prefsDict["hardwareSpeechDevice"] = self.hardwareDeviceChoice self.prefsDict["hardwareSpeechDevice"] = self.hardwareDeviceChoice
else: else:
+11
View File
@@ -698,6 +698,17 @@ class CthulhuDBusServiceInterface(Publishable):
debug.print_message(debug.LEVEL_INFO, msg, True) debug.print_message(debug.LEVEL_INFO, msg, True)
return result 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 def Quit(self) -> bool: # pylint: disable=invalid-name
"""Quits Cthulhu.""" """Quits Cthulhu."""
+88 -1
View File
@@ -37,13 +37,15 @@ __copyright__ = "Copyright (c) 2005-2008 Sun Microsystems Inc."
__license__ = "LGPL" __license__ = "LGPL"
import inspect import inspect
from contextlib import contextmanager
import traceback import traceback
import os import os
import re import re
import subprocess import subprocess
import sys import sys
import time
import types import types
from typing import Any, Optional, TextIO, Pattern, TYPE_CHECKING from typing import Any, Generator, Mapping, Optional, TextIO, Pattern, TYPE_CHECKING
from datetime import datetime from datetime import datetime
import gi import gi
@@ -149,6 +151,13 @@ eventDebugFilter: Optional[Pattern[str]] = None
# #
debugEventQueue: bool = False debugEventQueue: bool = False
# Startup timing is intentionally separate from normal event tracing. When
# explicitly enabled, it can print concise phase timings without turning on the
# full debug stream.
startupTimingEnabled: bool = False
_startupTimingStartTime: Optional[float] = None
_startupTimingLastTime: Optional[float] = None
# What module(s) should be traced if traceit is being used. By default # What module(s) should be traced if traceit is being used. By default
# we'll just attend to ourself. (And by default, we will not enable # we'll just attend to ourself. (And by default, we will not enable
# traceit.) Note that enabling this functionality will drag your system # traceit.) Note that enabling this functionality will drag your system
@@ -282,6 +291,84 @@ def print_log_tokens(level: int, prefix: str, tokens: list[Any], reason: Optiona
text = format_log_message(prefix, _format_tokens(tokens), reason) text = format_log_message(prefix, _format_tokens(tokens), reason)
printMessage(level, text, timestamp, stack) 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."""
global _startupTimingStartTime
global _startupTimingLastTime
now = time.monotonic()
_startupTimingStartTime = now
_startupTimingLastTime = now
def set_startup_timing_enabled(enabled: bool = True) -> None:
"""Enable or disable concise startup phase timing output."""
global startupTimingEnabled
startupTimingEnabled = enabled
if enabled and _startupTimingStartTime is None:
reset_startup_timing()
def is_startup_timing_enabled() -> bool:
"""Returns True when startup timing output should be emitted."""
return startupTimingEnabled or debugLevel <= LEVEL_INFO
def _startup_timing_level() -> int:
if debugLevel <= LEVEL_INFO:
return LEVEL_INFO
return LEVEL_SEVERE
def print_startup_timing(phase: str) -> None:
"""Print elapsed startup timing for a named phase."""
global _startupTimingStartTime
global _startupTimingLastTime
if not is_startup_timing_enabled():
return
now = time.monotonic()
if _startupTimingStartTime is None or _startupTimingLastTime is None:
_startupTimingStartTime = now
_startupTimingLastTime = now
delta = now - _startupTimingLastTime
total = now - _startupTimingStartTime
_startupTimingLastTime = now
text = f"STARTUP TIMING: {phase}: +{delta:.3f}s, total {total:.3f}s"
printMessage(_startup_timing_level(), text, True)
@contextmanager
def startup_timing(phase: str) -> Generator[None, None, None]:
"""Context manager for timing a startup phase."""
if not is_startup_timing_enabled():
yield
return
print_startup_timing(f"{phase} start")
try:
yield
finally:
print_startup_timing(f"{phase} end")
def printTokens(level: int, tokens: list[Any], timestamp: bool = False, stack: bool = False) -> None: def printTokens(level: int, tokens: list[Any], timestamp: bool = False, stack: bool = False) -> None:
if level < debugLevel: if level < debugLevel:
return return
+24
View File
@@ -194,6 +194,30 @@ class EventManager:
self._deactivateKeyHandling() self._deactivateKeyHandling()
debug.printMessage(debug.LEVEL_INFO, 'EVENT MANAGER: Deactivated', True) 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: def set_compositor_state_adapter(self, adapter: Any) -> None:
"""Stores the compositor state adapter used for startup wiring.""" """Stores the compositor state adapter used for startup wiring."""
+61 -15
View File
@@ -109,6 +109,21 @@ class InputEventManager:
debug.print_message(debug.LEVEL_INFO, msg, True) debug.print_message(debug.LEVEL_INFO, msg, True)
self._device = None 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: def enable_pointer_monitoring(self) -> bool:
"""Enables pointer monitoring on the current device, if possible.""" """Enables pointer monitoring on the current device, if possible."""
@@ -477,28 +492,49 @@ class InputEventManager:
return None return None
self._log_active_x11_window_for_xterm_check(window) self._log_active_x11_window_for_xterm_check(window)
sawIdentifier = False
for attrName in ("get_class_group_name", "get_class_instance_name", "get_name"): for attrName in ("get_class_group_name", "get_class_instance_name", "get_name"):
if self._identifier_is_xterm(self._safe_call(window, attrName)): value = self._safe_call(window, attrName)
if self._identifier_is_xterm(value):
return True return True
if isinstance(value, str) and value.strip():
sawIdentifier = True
classGroup = self._safe_call(window, "get_class_group") classGroup = self._safe_call(window, "get_class_group")
if classGroup is not None: if classGroup is not None:
for attrName in ("get_name", "get_res_class"): for attrName in ("get_name", "get_res_class"):
if self._identifier_is_xterm(self._safe_call(classGroup, attrName)): value = self._safe_call(classGroup, attrName)
if self._identifier_is_xterm(value):
return True return True
if isinstance(value, str) and value.strip():
sawIdentifier = True
try: try:
pid = int(window.get_pid()) pid = int(window.get_pid())
except Exception: except Exception:
pid = -1 pid = -1
if pid < 1: if pid < 1:
if not sawIdentifier:
msg = (
"INPUT EVENT MANAGER: XTerm matcher cannot identify active X11 window; "
"no usable identifiers or pid."
)
debug.print_message(debug.LEVEL_INFO, msg, True)
return None
return False return False
try: try:
with open(f"/proc/{pid}/cmdline", "rb") as cmdlineFile: with open(f"/proc/{pid}/cmdline", "rb") as cmdlineFile:
executable = cmdlineFile.read().split(b"\0", 1)[0].decode(errors="ignore") executable = cmdlineFile.read().split(b"\0", 1)[0].decode(errors="ignore")
except OSError: except OSError:
if not sawIdentifier:
msg = (
"INPUT EVENT MANAGER: XTerm matcher cannot identify active X11 window; "
"pid disappeared before cmdline lookup."
)
debug.print_message(debug.LEVEL_INFO, msg, True)
return None
return False return False
return self._identifier_is_xterm(executable) return self._identifier_is_xterm(executable)
@@ -590,6 +626,20 @@ class InputEventManager:
return result return result
def _clear_stale_atspi_context(self, manager: Any) -> None:
"""Clears cached input context after X11 positively contradicts AT-SPI."""
reason = "active X11 window not found in AT-SPI"
msg = (
"INPUT EVENT MANAGER: Clearing stale AT-SPI focus context; "
"X11 focus moved to an untracked window."
)
debug.print_message(debug.LEVEL_INFO, msg, True)
script_manager.get_manager().set_active_script(None, reason)
manager.clear_state(reason)
self._last_input_event = None
self._last_non_modifier_key_event = None
@staticmethod @staticmethod
def _get_active_script_app() -> Optional[Atspi.Accessible]: def _get_active_script_app() -> Optional[Atspi.Accessible]:
"""Returns the app for the active script, if one is available.""" """Returns the app for the active script, if one is available."""
@@ -621,15 +671,15 @@ class InputEventManager:
) -> bool: ) -> bool:
"""Returns True when XTerm is active and Cthulhu lacks matching AT-SPI context.""" """Returns True when XTerm is active and Cthulhu lacks matching AT-SPI context."""
if pendingFocus is not None:
msg = "INPUT EVENT MANAGER: XTerm matcher false; pending focus exists."
debug.print_message(debug.LEVEL_INFO, msg, True)
return False
match = self._active_x11_window_xterm_match() match = self._active_x11_window_xterm_match()
tokens = ["INPUT EVENT MANAGER: XTerm matcher returned", match] tokens = ["INPUT EVENT MANAGER: XTerm matcher returned", match]
debug.print_tokens(debug.LEVEL_INFO, tokens, True) debug.print_tokens(debug.LEVEL_INFO, tokens, True)
if match is True: if match is True:
return True return True
if pendingFocus is not None:
msg = "INPUT EVENT MANAGER: XTerm matcher false; pending focus exists."
debug.print_message(debug.LEVEL_INFO, msg, True)
return False
if match is None and self._scriptWithSuspendedGrabsForXterm is not None: if match is None and self._scriptWithSuspendedGrabsForXterm is not None:
msg = ( msg = (
"INPUT EVENT MANAGER: Keeping XTerm key-grab suspension; " "INPUT EVENT MANAGER: Keeping XTerm key-grab suspension; "
@@ -743,6 +793,11 @@ class InputEventManager:
manager.set_active_window(window) manager.set_active_window(window)
else: else:
focus_window = self._get_top_level_window(pendingFocus or manager.get_locus_of_focus()) focus_window = self._get_top_level_window(pendingFocus or manager.get_locus_of_focus())
staleContext = focus_window or window or self._get_active_script_app()
if self._active_x11_window_differs_from(staleContext):
self._clear_stale_atspi_context(manager)
return False
if focus_window is not None: if focus_window is not None:
window = focus_window window = focus_window
tokens = [ tokens = [
@@ -755,15 +810,6 @@ class InputEventManager:
# One example: Brave's popup menus live in frames which lack the active # One example: Brave's popup menus live in frames which lack the active
# state. Failing to revalidate the window on a key press is inconclusive; # state. Failing to revalidate the window on a key press is inconclusive;
# do not wipe out the last known window and focus state. # do not wipe out the last known window and focus state.
focus = pendingFocus or manager.get_locus_of_focus()
staleContext = window or self._get_active_script_app()
if self._active_x11_window_differs_from(staleContext):
msg = (
"INPUT EVENT MANAGER: X11 focus moved to an untracked window; "
"preserving Cthulhu key handling."
)
debug.print_message(debug.LEVEL_INFO, msg, True)
tokens = [ tokens = [
"WARNING:", "WARNING:",
window, window,
+5
View File
@@ -293,6 +293,11 @@ CLI_HELP = _("Show this help message and exit")
# start with 'debug' and end with '.out', regardless of the locale.). # start with 'debug' and end with '.out', regardless of the locale.).
CLI_ENABLE_DEBUG = _("Send debug output to debug-YYYY-MM-DD-HH:MM:SS.out") CLI_ENABLE_DEBUG = _("Send debug output to debug-YYYY-MM-DD-HH:MM:SS.out")
# Translators: This is the description of command line option '--debug-startup'
# which causes Cthulhu to print concise startup timing information without
# enabling the full debug output stream.
CLI_ENABLE_STARTUP_DEBUG = _("Send concise startup timing output")
# Translators: This is the description of command line option '--debug-file' # Translators: This is the description of command line option '--debug-file'
# which allows the user to override the default date-based name of the debugging # which allows the user to override the default date-based name of the debugging
# output file. # output file.
+15 -6
View File
@@ -545,31 +545,40 @@ class SpeechServer(speechserver.SpeechServer):
- utteranceIterator: Iterator yielding [SayAllContext, acss] tuples - utteranceIterator: Iterator yielding [SayAllContext, acss] tuples
- progressCallback: Called with progress updates - progressCallback: Called with progress updates
""" """
GLib.idle_add(self._sayAllWorker, utteranceIterator, progressCallback) with self._lock:
self._stopEvent.clear()
self._speakGeneration += 1
generation = self._speakGeneration
GLib.idle_add(self._sayAllWorker, utteranceIterator, progressCallback, generation)
def _sayAllWorker(self, iterator, callback): def _sayAllWorker(self, iterator, callback, generation):
"""Process one utterance at a time (called via GLib.idle_add). """Process one utterance at a time (called via GLib.idle_add).
Arguments: Arguments:
- iterator: Utterance iterator - iterator: Utterance iterator
- callback: Progress callback - callback: Progress callback
""" """
with self._lock:
if self._stopEvent.is_set() or generation != self._speakGeneration:
return False
try: try:
context, acss = next(iterator) context, acss = next(iterator)
except StopIteration: except StopIteration:
return False return False
def onComplete(): def onComplete():
with self._lock:
if self._stopEvent.is_set() or generation != self._speakGeneration:
return False
context.currentOffset = context.endOffset context.currentOffset = context.endOffset
callback(context.copy(), speechserver.SayAllContext.COMPLETED) callback(context.copy(), speechserver.SayAllContext.COMPLETED)
GLib.idle_add(self._sayAllWorker, iterator, callback) GLib.idle_add(self._sayAllWorker, iterator, callback, generation)
return False
context.currentOffset = context.startOffset context.currentOffset = context.startOffset
callback(context.copy(), speechserver.SayAllContext.PROGRESS) callback(context.copy(), speechserver.SayAllContext.PROGRESS)
with self._lock:
self._stopEvent.clear()
generation = self._speakGeneration
self._currentFuture = self._executor.submit( self._currentFuture = self._executor.submit(
self._synthesizeAndPlay, context.utterance, acss, onComplete, generation self._synthesizeAndPlay, context.utterance, acss, onComplete, generation
) )
+29
View File
@@ -36,6 +36,10 @@ LEGACY_PLUGIN_NAME_ALIASES: Dict[str, str] = {
"ocrdesktop": "OCR", "ocrdesktop": "OCR",
} }
OBSOLETE_PLUGIN_NAMES = {
"pluginmanager",
}
LEGACY_PLUGIN_DIR_ALIASES: Dict[str, str] = { LEGACY_PLUGIN_DIR_ALIASES: Dict[str, str] = {
"OCRDesktop": "OCR", "OCRDesktop": "OCR",
} }
@@ -721,6 +725,9 @@ class PluginSystemManager:
if not name: if not name:
continue continue
name_lower = name.lower() name_lower = name.lower()
if name_lower in OBSOLETE_PLUGIN_NAMES:
logger.info(f"Dropping obsolete plugin name {name}")
continue
alias = LEGACY_PLUGIN_NAME_ALIASES.get(name_lower) alias = LEGACY_PLUGIN_NAME_ALIASES.get(name_lower)
if alias: if alias:
logger.info(f"Mapping legacy plugin name {name} to {alias}") logger.info(f"Mapping legacy plugin name {name} to {alias}")
@@ -1095,3 +1102,25 @@ class PluginSystemManager:
for pluginInfo in self.plugins: for pluginInfo in self.plugins:
if ForceAllPlugins or pluginInfo.loaded: if ForceAllPlugins or pluginInfo.loaded:
self.unloadPlugin(pluginInfo) 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,
}
@@ -1,14 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2025 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.
"""PluginManager plugin package."""
from .plugin import PluginManager
__all__ = ['PluginManager']
@@ -1,14 +0,0 @@
pluginmanager_python_sources = files([
'__init__.py',
'plugin.py'
])
python3.install_sources(
pluginmanager_python_sources,
subdir: 'cthulhu/plugins/PluginManager'
)
install_data(
'plugin.info',
install_dir: python3.get_install_dir() / 'cthulhu' / 'plugins' / 'PluginManager'
)
@@ -1,8 +0,0 @@
name = PluginManager
version = 1.0.0
description = GUI interface for managing Cthulhu plugins - enable/disable plugins with checkboxes
authors = Stormux <storm_dragon@stormux.org>
website = https://git.stormux.org/storm/cthulhu
copyright = Copyright 2025
builtin = false
hidden = false
-505
View File
@@ -1,505 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2025 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.
"""PluginManager plugin for Cthulhu - GUI interface for managing plugins."""
import logging
import os
import configparser
from pathlib import Path
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk, Gdk, Pango
from cthulhu.plugin import Plugin, cthulhu_hookimpl
from cthulhu import cthulhu
from cthulhu import debug
logger = logging.getLogger(__name__)
class PluginManager(Plugin):
"""Plugin that provides a GUI interface for managing other plugins."""
PLUGIN_COL_ENABLED = 0
PLUGIN_COL_DISPLAY = 1
PLUGIN_COL_CAN_TOGGLE = 2
PLUGIN_COL_NAME = 3
def __init__(self, *args, **kwargs):
"""Initialize the PluginManager plugin."""
super().__init__(*args, **kwargs)
self._kb_binding = None
self._dialog = None
self._plugin_treeview = None
self._plugin_model = None
self._activated = False
debug.printMessage(debug.LEVEL_INFO, "PluginManager: Plugin initialized", True)
@cthulhu_hookimpl
def activate(self, plugin=None):
"""Activate the PluginManager plugin."""
if plugin is not None and plugin is not self:
return
# Prevent duplicate activation
if self._activated:
debug.printMessage(debug.LEVEL_INFO, "PluginManager: Already activated, skipping", True)
return
try:
debug.printMessage(debug.LEVEL_INFO, "PluginManager: Plugin activation starting", True)
self._activated = True
debug.printMessage(debug.LEVEL_INFO, "PluginManager: Plugin activated successfully", True)
return True
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: ERROR activating plugin: {e}", True)
return False
@cthulhu_hookimpl
def deactivate(self, plugin=None):
"""Deactivate the PluginManager plugin."""
if plugin is not None and plugin is not self:
return
try:
debug.printMessage(debug.LEVEL_INFO, "PluginManager: Plugin deactivation starting", True)
# Close dialog if open
if self._dialog:
self._dialog.destroy()
self._dialog = None
self._activated = False
debug.printMessage(debug.LEVEL_INFO, "PluginManager: Plugin deactivated successfully", True)
return True
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: ERROR deactivating plugin: {e}", True)
return False
def _register_keybinding(self):
"""Register the Cthulhu+Shift+P keybinding for opening plugin manager."""
try:
if not self.app:
debug.printMessage(debug.LEVEL_INFO, "PluginManager: ERROR - No app reference available for keybinding", True)
return
# Register Cthulhu+Shift+P keybinding
gesture_string = "kb:cthulhu+shift+p"
description = "Open plugin manager"
self._kb_binding = self.registerGestureByString(
self._open_plugin_manager,
description,
gesture_string
)
if self._kb_binding:
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Registered keybinding {gesture_string}", True)
else:
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: ERROR - Failed to register keybinding {gesture_string}", True)
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: ERROR registering keybinding: {e}", True)
def _open_plugin_manager(self, script, inputEvent=None):
"""Open the plugin manager dialog."""
try:
debug.printMessage(debug.LEVEL_INFO, "PluginManager: Opening plugin manager dialog", True)
# Close existing dialog if open
if self._dialog:
self._dialog.destroy()
# Create new dialog
self._create_dialog()
return True
except Exception as e:
logger.error(f"PluginManager: Error opening plugin manager: {e}")
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Error opening dialog: {e}", True)
return False
def _create_dialog(self):
"""Create and show the plugin manager dialog."""
try:
# Create dialog window
self._dialog = Gtk.Dialog(
title="Cthulhu Plugin Manager",
parent=None,
flags=Gtk.DialogFlags.MODAL | Gtk.DialogFlags.DESTROY_WITH_PARENT
)
# Set dialog properties
self._dialog.set_default_size(400, 300)
self._dialog.set_border_width(10)
# Add buttons
self._dialog.add_button("Close", Gtk.ResponseType.CLOSE)
# Create main content area
content_area = self._dialog.get_content_area()
# Add title label
title_label = Gtk.Label()
title_label.set_markup("<b>Available Plugins</b>")
title_label.set_halign(Gtk.Align.CENTER)
content_area.pack_start(title_label, False, False, 10)
# Add info label about PluginManager
info_label = Gtk.Label()
info_label.set_markup("<i>Note:To apply changes, restart Cthulhu</i>")
info_label.set_halign(Gtk.Align.CENTER)
content_area.pack_start(info_label, False, False, 5)
# Create scrolled window for plugin list
scrolled = Gtk.ScrolledWindow()
scrolled.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC)
scrolled.set_size_request(-1, 200)
self._plugin_model = Gtk.TreeStore(bool, str, bool, str)
self._plugin_treeview = Gtk.TreeView(model=self._plugin_model)
self._plugin_treeview.set_headers_visible(True)
self._plugin_treeview.set_enable_search(False)
self._plugin_treeview.connect("key-press-event", self._on_plugin_list_key_press)
self._plugin_treeview.connect("row-activated", self._on_plugin_row_activated)
toggle_renderer = Gtk.CellRendererToggle()
toggle_renderer.set_activatable(True)
toggle_renderer.connect("toggled", self._on_plugin_toggled)
toggle_column = Gtk.TreeViewColumn(
"Enabled",
toggle_renderer,
active=self.PLUGIN_COL_ENABLED,
activatable=self.PLUGIN_COL_CAN_TOGGLE
)
toggle_column.add_attribute(toggle_renderer, "visible", self.PLUGIN_COL_CAN_TOGGLE)
self._plugin_treeview.append_column(toggle_column)
text_renderer = Gtk.CellRendererText()
text_renderer.set_property("ellipsize", Pango.EllipsizeMode.END)
text_column = Gtk.TreeViewColumn(
"Plugin",
text_renderer,
text=self.PLUGIN_COL_DISPLAY
)
text_column.set_expand(True)
self._plugin_treeview.append_column(text_column)
scrolled.add(self._plugin_treeview)
content_area.pack_start(scrolled, True, True, 0)
# Populate plugin list
self._populate_plugin_list()
# Connect signals
self._dialog.connect("response", self._on_dialog_response)
# Show dialog
self._dialog.show_all()
debug.printMessage(debug.LEVEL_INFO, "PluginManager: Dialog created and shown", True)
except Exception as e:
logger.error(f"PluginManager: Error creating dialog: {e}")
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Error creating dialog: {e}", True)
def _populate_plugin_list(self):
"""Populate the plugin list with available plugins."""
try:
# Get available plugins
available_plugins = self._discover_plugins()
# Get currently active plugins
active_plugins = cthulhu.cthulhuApp.settingsManager.getSetting('activePlugins') or []
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Found {len(available_plugins)} plugins", True)
# Clear existing model rows
self._plugin_model.clear()
# Add each plugin as a checkbox (except PluginManager itself)
enabled_iter = self._plugin_model.append(
None,
[False, "Enabled plugins", False, ""]
)
disabled_iter = self._plugin_model.append(
None,
[False, "Disabled plugins", False, ""]
)
for plugin_name, plugin_info in sorted(available_plugins.items()):
# Skip PluginManager to prevent users from disabling plugin management
if plugin_name == "PluginManager":
continue
display_name = plugin_info.get('name', plugin_name)
display_text = display_name
description = plugin_info.get('description')
if description:
display_text += f" - {description}"
version = plugin_info.get('version')
if version:
display_text += f" (v{version})"
is_active = plugin_name in active_plugins
parent_iter = enabled_iter if is_active else disabled_iter
self._plugin_model.append(
parent_iter,
[is_active, display_text, True, plugin_name]
)
self._plugin_treeview.collapse_all()
except Exception as e:
logger.error(f"PluginManager: Error populating plugin list: {e}")
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Error populating list: {e}", True)
def _discover_plugins(self):
"""Discover all available plugins."""
plugins = {}
try:
# Get plugin system manager
from cthulhu import plugin_system_manager
# Use existing plugin manager to get plugins
manager = plugin_system_manager.getManager()
if manager:
manager.rescanPlugins()
for plugin_info in manager.plugins:
plugin_name = plugin_info.get_module_name()
plugins[plugin_name] = {
'name': plugin_info.get_name(),
'description': plugin_info.get_description(),
'version': plugin_info.get_version(),
'path': plugin_info.get_module_dir()
}
else:
# Fallback: manually scan plugin directories
plugins = self._manual_plugin_discovery()
except Exception as e:
logger.error(f"PluginManager: Error in plugin discovery: {e}")
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Plugin discovery error: {e}", True)
# Fallback to manual discovery
plugins = self._manual_plugin_discovery()
return plugins
def _manual_plugin_discovery(self):
"""Manually discover plugins by scanning directories."""
plugins = {}
try:
# Get plugin directories
from cthulhu.plugin_system_manager import PluginType
system_dir = PluginType.SYSTEM.get_root_dir()
user_dir = PluginType.USER.get_root_dir()
for plugin_dir in [system_dir, user_dir]:
if os.path.exists(plugin_dir):
self._scan_plugin_directory(plugin_dir, plugins)
except Exception as e:
logger.error(f"PluginManager: Error in manual discovery: {e}")
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Manual discovery error: {e}", True)
return plugins
def _scan_plugin_directory(self, directory, plugins):
"""Scan a directory for plugins."""
try:
for item in os.listdir(directory):
plugin_path = os.path.join(directory, item)
if os.path.isdir(plugin_path):
# Check for plugin.py and plugin.info
plugin_py = os.path.join(plugin_path, "plugin.py")
plugin_info_file = os.path.join(plugin_path, "plugin.info")
if os.path.exists(plugin_py):
plugin_info = {'name': item, 'description': '', 'version': ''}
# Try to read plugin.info
if os.path.exists(plugin_info_file):
try:
config = configparser.ConfigParser()
config.read(plugin_info_file)
# Handle both INI-style and simple key=value format
if config.sections():
# INI-style format
for section in config.sections():
for key, value in config[section].items():
if key.lower() in ['description', 'version']:
plugin_info[key.lower()] = value
else:
# Simple key=value format
with open(plugin_info_file, 'r') as f:
for line in f:
line = line.strip()
if '=' in line and not line.startswith('#'):
key, value = line.split('=', 1)
key = key.strip().lower()
value = value.strip()
if key in ['description', 'version']:
plugin_info[key] = value
except Exception as info_e:
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Error reading {plugin_info_file}: {info_e}", True)
plugins[item] = plugin_info
except Exception as e:
logger.error(f"PluginManager: Error scanning directory {directory}: {e}")
def _on_plugin_toggled(self, renderer, path):
"""Handle plugin toggle via TreeView."""
try:
tree_iter = self._plugin_model.get_iter(path)
if not tree_iter:
return
can_toggle = self._plugin_model.get_value(tree_iter, self.PLUGIN_COL_CAN_TOGGLE)
if not can_toggle:
return
is_active = self._plugin_model.get_value(tree_iter, self.PLUGIN_COL_ENABLED)
plugin_name = self._plugin_model.get_value(tree_iter, self.PLUGIN_COL_NAME)
new_active = not is_active
self._plugin_model.set_value(tree_iter, self.PLUGIN_COL_ENABLED, new_active)
self._set_plugin_active(plugin_name, new_active)
self._rebuild_plugin_groups()
except Exception as error:
logger.error(f"PluginManager: Error toggling plugin at path {path}: {error}")
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Error toggling plugin at path {path}: {error}", True)
def _on_plugin_list_key_press(self, widget, event):
"""Toggle plugin on Space while keeping Tab navigation outside the list."""
if event.keyval != Gdk.KEY_space:
return False
selection = self._plugin_treeview.get_selection()
model, tree_iter = selection.get_selected()
if not tree_iter:
return False
path = model.get_path(tree_iter)
self._on_plugin_toggled(None, path)
return True
def _on_plugin_row_activated(self, treeview, path, column):
"""Toggle plugin when activating a row."""
self._on_plugin_toggled(None, path)
def _rebuild_plugin_groups(self):
"""Rebuild the tree so enabled/disabled groups stay accurate."""
if not self._plugin_model:
return
selection = self._plugin_treeview.get_selection()
model, tree_iter = selection.get_selected()
selected_name = None
if tree_iter:
selected_name = model.get_value(tree_iter, self.PLUGIN_COL_NAME)
self._populate_plugin_list()
if selected_name:
self._select_plugin_row(selected_name)
def _select_plugin_row(self, plugin_name):
"""Select the row for the given plugin name."""
def _walk(model, tree_iter):
while tree_iter:
name = model.get_value(tree_iter, self.PLUGIN_COL_NAME)
if name == plugin_name:
path = model.get_path(tree_iter)
self._plugin_treeview.expand_to_path(path)
self._plugin_treeview.get_selection().select_path(path)
return True
if model.iter_has_child(tree_iter):
child = model.iter_children(tree_iter)
if _walk(model, child):
return True
tree_iter = model.iter_next(tree_iter)
return False
root = self._plugin_model.get_iter_first()
_walk(self._plugin_model, root)
def _set_plugin_active(self, plugin_name, is_active):
"""Update settings to enable or disable a plugin."""
try:
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Plugin {plugin_name} toggled to {'active' if is_active else 'inactive'}", True)
# Get current active plugins
active_plugins = cthulhu.cthulhuApp.settingsManager.getSetting('activePlugins') or []
active_plugins = list(active_plugins)
if is_active and plugin_name not in active_plugins:
active_plugins.append(plugin_name)
elif not is_active and plugin_name in active_plugins:
active_plugins.remove(plugin_name)
cthulhu.cthulhuApp.settingsManager.setSetting('activePlugins', active_plugins)
if hasattr(cthulhu.cthulhuApp.settingsManager, "general") and isinstance(cthulhu.cthulhuApp.settingsManager.general, dict):
cthulhu.cthulhuApp.settingsManager.general['activePlugins'] = active_plugins
try:
active_profile = cthulhu.cthulhuApp.settingsManager.getSetting('activeProfile')
if isinstance(active_profile, (list, tuple)) and len(active_profile) > 1:
profile_name = active_profile[1]
else:
profile_name = cthulhu.cthulhuApp.settingsManager.profile or 'default'
current_general = cthulhu.cthulhuApp.settingsManager.getGeneralSettings(profile_name) or {}
current_general['activePlugins'] = active_plugins
cthulhu.cthulhuApp.settingsManager.profile = profile_name
cthulhu.cthulhuApp.settingsManager.saveProfileSettings(current_general)
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Settings saved via settings manager (profile {profile_name})", True)
except Exception as save_error:
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Error saving plugin state: {save_error}", True)
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Updated active plugins: {active_plugins}", True)
try:
from cthulhu import plugin_system_manager
manager = plugin_system_manager.getManager()
if manager:
manager.setActivePlugins(active_plugins)
debug.printMessage(debug.LEVEL_INFO, "PluginManager: Applied active plugin changes", True)
except Exception as apply_error:
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Failed to apply plugin changes: {apply_error}", True)
except Exception as error:
logger.error(f"PluginManager: Error toggling plugin {plugin_name}: {error}")
debug.printMessage(debug.LEVEL_INFO, f"PluginManager: Error toggling {plugin_name}: {error}", True)
def _on_dialog_response(self, dialog, response_id):
"""Handle dialog response (close button clicked)."""
try:
if response_id == Gtk.ResponseType.CLOSE:
dialog.destroy()
self._dialog = None
debug.printMessage(debug.LEVEL_INFO, "PluginManager: Dialog closed", True)
except Exception as e:
logger.error(f"PluginManager: Error handling dialog response: {e}")
-1
View File
@@ -8,7 +8,6 @@ subdir('HelloCthulhu')
subdir('GameMode') subdir('GameMode')
subdir('nvda2cthulhu') subdir('nvda2cthulhu')
subdir('OCR') subdir('OCR')
subdir('PluginManager')
subdir('SpeechHistory') subdir('SpeechHistory')
subdir('SimplePluginSystem') subdir('SimplePluginSystem')
subdir('hello_world') subdir('hello_world')
+25
View File
@@ -396,6 +396,31 @@ class ScriptManager:
reason 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]: def _get_script_for_app_replicant(self, app: Atspi.Accessible) -> Optional[Script]:
if not self._active: if not self._active:
return None return None
@@ -164,6 +164,16 @@ class Script(Chromium.Script):
super().onTextInserted(event) super().onTextInserted(event)
def onCaretMoved(self, event):
"""Callback for object:text-caret-moved accessibility events."""
if self._isSteamLoadingSpinnerCaretEvent(event):
msg = "STEAM: Ignoring caret event from transient loading spinner"
debug.printMessage(debug.LEVEL_INFO, msg, True)
return True
return super().onCaretMoved(event)
def onSelectionChanged(self, event): def onSelectionChanged(self, event):
"""Callback for object:selection-changed accessibility events.""" """Callback for object:selection-changed accessibility events."""
@@ -193,6 +203,34 @@ class Script(Chromium.Script):
debug.printMessage(debug.LEVEL_INFO, msg, True) debug.printMessage(debug.LEVEL_INFO, msg, True)
return True return True
def _isSteamLoadingSpinnerCaretEvent(self, event) -> bool:
if not (event and event.type.startswith("object:text-caret-moved")):
return False
source = getattr(event, "source", None)
if not source or not self.utilities.inDocumentContent(source):
return False
if AXObject.get_name(source):
return False
try:
children = list(AXObject.iter_children(source))
except Exception:
return False
if not children:
return False
for child in children:
if not AXUtilities.is_image_or_canvas(child):
continue
name = AXObject.get_name(child) or ""
if name.strip().casefold() == "steam spinner":
return True
return False
def _trySteamButtonActivation(self, keyboardEvent) -> bool: def _trySteamButtonActivation(self, keyboardEvent) -> bool:
if keyboardEvent.event_string not in ["Return", "KP_Enter"]: if keyboardEvent.event_string not in ["Return", "KP_Enter"]:
return False return False
@@ -61,13 +61,7 @@ class Utilities(web.Utilities):
boundary = args.get('boundary') boundary = args.get('boundary')
# Gecko fails to implement this boundary type. # Gecko fails to implement this boundary type.
if boundary == Atspi.TextBoundaryType.SENTENCE_START: return boundary == Atspi.TextBoundaryType.SENTENCE_START
return True
if self.isContentEditableWithEmbeddedObjects(obj):
return boundary == Atspi.TextBoundaryType.WORD_START
return True
def _treatAsLeafNode(self, obj): def _treatAsLeafNode(self, obj):
if AXUtilities.is_table_row(obj): if AXUtilities.is_table_row(obj):
+34
View File
@@ -1758,6 +1758,7 @@ class Script(default.Script):
if not document: if not document:
msg = "WEB: Locus of focus changed to non-document obj" msg = "WEB: Locus of focus changed to non-document obj"
self._madeFindAnnouncement = False self._madeFindAnnouncement = False
if not self._focusModeIsSticky:
self._inFocusMode = False self._inFocusMode = False
self._setNavigationSuspended(True, "focus left document content") self._setNavigationSuspended(True, "focus left document content")
debug.printMessage(debug.LEVEL_INFO, msg, True) debug.printMessage(debug.LEVEL_INFO, msg, True)
@@ -1779,6 +1780,19 @@ class Script(default.Script):
if self._navSuspended: if self._navSuspended:
self._setNavigationSuspended(False, "focus entered document content") self._setNavigationSuspended(False, "focus entered document content")
if self._focusModeIsSticky and not self._inFocusMode:
msg = "WEB: Restoring sticky focus mode after returning to document content"
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._inFocusMode = True
self.presentMessage(messages.MODE_FOCUS_IS_STICKY)
self.refreshKeyGrabs()
elif self._browseModeIsSticky and self._inFocusMode:
msg = "WEB: Restoring sticky browse mode after returning to document content"
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._inFocusMode = False
self.presentMessage(messages.MODE_BROWSE_IS_STICKY)
self.refreshKeyGrabs()
if self.flatReviewPresenter.is_active(): if self.flatReviewPresenter.is_active():
self.flatReviewPresenter.quit() self.flatReviewPresenter.quit()
@@ -2102,6 +2116,11 @@ class Script(default.Script):
else: else:
self._clearSyntheticWebSelection() 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: if self._lastCommandWasCaretNav:
msg = "WEB: Event ignored: Last command was caret nav" msg = "WEB: Event ignored: Last command was caret nav"
debug.printMessage(debug.LEVEL_INFO, msg, True) debug.printMessage(debug.LEVEL_INFO, msg, True)
@@ -2551,6 +2570,7 @@ class Script(default.Script):
if self._browseModeIsSticky: if self._browseModeIsSticky:
msg = "WEB: Web app descendant claimed focus, but browse mode is sticky" msg = "WEB: Web app descendant claimed focus, but browse mode is sticky"
debug.printMessage(debug.LEVEL_INFO, msg, True) debug.printMessage(debug.LEVEL_INFO, msg, True)
return True
elif AXUtilities.is_tool_tip(event.source) \ elif AXUtilities.is_tool_tip(event.source) \
and AXObject.find_ancestor(cthulhu_state.locusOfFocus, lambda x: x == event.source): and AXObject.find_ancestor(cthulhu_state.locusOfFocus, lambda x: x == event.source):
msg = "WEB: Event believed to be side effect of tooltip navigation." msg = "WEB: Event believed to be side effect of tooltip navigation."
@@ -2609,6 +2629,20 @@ class Script(default.Script):
self.utilities.setCaretContext(event.source, 0) self.utilities.setCaretContext(event.source, 0)
obj, offset = 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: if self._lastCommandWasCaretNav:
msg = "WEB: Event ignored: Last command was caret nav" msg = "WEB: Event ignored: Last command was caret nav"
debug.printMessage(debug.LEVEL_INFO, msg, True) debug.printMessage(debug.LEVEL_INFO, msg, True)
+63 -2
View File
@@ -230,6 +230,7 @@ class Utilities(script_utilities.Utilities):
self._shouldInferLabelFor = {} self._shouldInferLabelFor = {}
self._treatAsTextObject = {} self._treatAsTextObject = {}
self._treatAsDiv = {} self._treatAsDiv = {}
self.clearContentCache()
self._paths = {} self._paths = {}
self._contextPathsRolesAndNames = {} self._contextPathsRolesAndNames = {}
self._canHaveCaretContextDecision = {} self._canHaveCaretContextDecision = {}
@@ -721,8 +722,15 @@ class Utilities(script_utilities.Utilities):
nextobj, nextoffset = self.findNextCaretInOrder(obj, offset) nextobj, nextoffset = self.findNextCaretInOrder(obj, offset)
if skipSpace: if skipSpace:
while nextobj and AXText.get_character_at_offset(nextobj, nextoffset)[0].isspace(): seen = {(nextobj, nextoffset)}
while self.treatAsTextObject(nextobj) \
and AXText.get_character_at_offset(nextobj, nextoffset)[0].isspace():
nextobj, nextoffset = self.findNextCaretInOrder(nextobj, nextoffset) nextobj, nextoffset = self.findNextCaretInOrder(nextobj, nextoffset)
if (nextobj, nextoffset) in seen:
msg = "WEB: Cycle detected in nextContext skipSpace. Breaking."
debug.printMessage(debug.LEVEL_INFO, msg, True)
break
seen.add((nextobj, nextoffset))
return nextobj, nextoffset return nextobj, nextoffset
@@ -732,8 +740,15 @@ class Utilities(script_utilities.Utilities):
prevobj, prevoffset = self.findPreviousCaretInOrder(obj, offset) prevobj, prevoffset = self.findPreviousCaretInOrder(obj, offset)
if skipSpace: if skipSpace:
while prevobj and AXText.get_character_at_offset(prevobj, prevoffset)[0].isspace(): seen = {(prevobj, prevoffset)}
while self.treatAsTextObject(prevobj) \
and AXText.get_character_at_offset(prevobj, prevoffset)[0].isspace():
prevobj, prevoffset = self.findPreviousCaretInOrder(prevobj, prevoffset) prevobj, prevoffset = self.findPreviousCaretInOrder(prevobj, prevoffset)
if (prevobj, prevoffset) in seen:
msg = "WEB: Cycle detected in previousContext skipSpace. Breaking."
debug.printMessage(debug.LEVEL_INFO, msg, True)
break
seen.add((prevobj, prevoffset))
return prevobj, prevoffset return prevobj, prevoffset
@@ -1793,6 +1808,21 @@ class Utilities(script_utilities.Utilities):
self._canHaveCaretContextDecision = {} self._canHaveCaretContextDecision = {}
return rv 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): def _getLineContentsAtOffset(self, obj, offset, layoutMode=None, useCache=True):
startTime = time.time() startTime = time.time()
if not obj: if not obj:
@@ -1881,6 +1911,13 @@ class Utilities(script_utilities.Utilities):
self._debugContentsInfo(obj, offset, objects, "Line (not layout mode)") self._debugContentsInfo(obj, offset, objects, "Line (not layout mode)")
return objects 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] firstObj, firstStart, firstEnd, firstString = objects[0]
if (extents[2] == 0 and extents[3] == 0) or self.isMath(firstObj): if (extents[2] == 0 and extents[3] == 0) or self.isMath(firstObj):
extents = self.getExtents(firstObj, firstStart, firstEnd) extents = self.getExtents(firstObj, firstStart, firstEnd)
@@ -2027,6 +2064,13 @@ class Utilities(script_utilities.Utilities):
tokens = ["WEB: Previous context is: ", obj, ", ", offset] tokens = ["WEB: Previous context is: ", obj, ", ", offset]
debug.printTokens(debug.LEVEL_INFO, tokens, True) 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) contents = self.getLineContentsAtOffset(obj, offset, layoutMode, useCache)
if not contents: if not contents:
tokens = ["WEB: Could not get line contents for ", obj, ", ", offset] tokens = ["WEB: Could not get line contents for ", obj, ", ", offset]
@@ -2092,6 +2136,23 @@ class Utilities(script_utilities.Utilities):
tokens = ["WEB: Next context is: ", obj, ", ", offset] tokens = ["WEB: Next context is: ", obj, ", ", offset]
debug.printTokens(debug.LEVEL_INFO, tokens, True) 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) contents = self.getLineContentsAtOffset(obj, offset, layoutMode, useCache)
if line == contents: if line == contents:
obj, offset = self.nextContext(obj, offset, True) obj, offset = self.nextContext(obj, offset, True)
-1
View File
@@ -497,7 +497,6 @@ presentLiveRegionFromInactiveTab = False
# Plugins # Plugins
activePlugins = [ activePlugins = [
'PluginManager',
'ByeCthulhu', 'ByeCthulhu',
'Clipboard', 'Clipboard',
'HelloCthulhu', 'HelloCthulhu',
+34 -4
View File
@@ -165,6 +165,8 @@ class SpeechServer(speechserver.SpeechServer):
SpeechServer._active_servers[serverId] = self SpeechServer._active_servers[serverId] = self
self._lastKeyEchoTime = None self._lastKeyEchoTime = None
self._say_all_generation = 0
self._cancelled_say_all_generation = None
def _init(self): def _init(self):
self._client = client = speechd.SSIPClient('Cthulhu', component=self._id) self._client = client = speechd.SSIPClient('Cthulhu', component=self._id)
@@ -489,22 +491,42 @@ class SpeechServer(speechserver.SpeechServer):
self._debug_sd_values(f"Speaking '{ssml}' ") self._debug_sd_values(f"Speaking '{ssml}' ")
self._send_command(self._client.speak, ssml, **kwargs) self._send_command(self._client.speak, ssml, **kwargs)
def _say_all(self, iterator, cthulhu_callback): def _say_all(self, iterator, cthulhu_callback, generation):
"""Process another sayAll chunk. """Process another sayAll chunk.
Called by the gidle thread. Called by the gidle thread.
""" """
if generation != self._say_all_generation:
return False
if generation == self._cancelled_say_all_generation:
return False
try: try:
context, acss = next(iterator) context, acss = next(iterator)
except StopIteration: except StopIteration:
pass pass
else: else:
def dispatch_callback(context_copy, progress_type, allow_interrupted=False):
if not allow_interrupted and generation != self._say_all_generation:
return False
cthulhu_callback(context_copy, progress_type)
return False
def callback(callbackType, index_mark=None): def callback(callbackType, index_mark=None):
# This callback is called in Speech Dispatcher listener thread. # This callback is called in Speech Dispatcher listener thread.
# No subsequent Speech Dispatcher interaction is allowed here, # No subsequent Speech Dispatcher interaction is allowed here,
# so we pass the calls to the gidle thread. # so we pass the calls to the gidle thread.
t = self._CALLBACK_TYPE_MAP[callbackType] t = self._CALLBACK_TYPE_MAP[callbackType]
if generation != self._say_all_generation:
return
if generation == self._cancelled_say_all_generation:
if t == speechserver.SayAllContext.INTERRUPTED:
self._say_all_generation += 1
self._cancelled_say_all_generation = None
GLib.idle_add(dispatch_callback, context.copy(), t, True)
return
if t == speechserver.SayAllContext.PROGRESS: if t == speechserver.SayAllContext.PROGRESS:
if index_mark: if index_mark:
index = index_mark.split(':') index = index_mark.split(':')
@@ -524,14 +546,15 @@ class SpeechServer(speechserver.SpeechServer):
elif t == speechserver.SayAllContext.COMPLETED: elif t == speechserver.SayAllContext.COMPLETED:
context.currentOffset = context.endOffset context.currentOffset = context.endOffset
context.currentEndOffset = None context.currentEndOffset = None
GLib.idle_add(cthulhu_callback, context.copy(), t) GLib.idle_add(dispatch_callback, context.copy(), t)
if t == speechserver.SayAllContext.COMPLETED: if t == speechserver.SayAllContext.COMPLETED:
GLib.idle_add(self._say_all, iterator, cthulhu_callback) GLib.idle_add(self._say_all, iterator, cthulhu_callback, generation)
self._speak(context.utterance, acss, callback=callback, self._speak(context.utterance, acss, callback=callback,
event_types=list(self._CALLBACK_TYPE_MAP.keys())) event_types=list(self._CALLBACK_TYPE_MAP.keys()))
return False # to indicate, that we don't want to be called again. return False # to indicate, that we don't want to be called again.
def _cancel(self): def _cancel(self):
self._cancelled_say_all_generation = self._say_all_generation
self._send_command(self._client.cancel) self._send_command(self._client.cancel)
def _change_default_speech_rate(self, step, decrease=False): def _change_default_speech_rate(self, step, decrease=False):
@@ -657,7 +680,14 @@ class SpeechServer(speechserver.SpeechServer):
self._speak(text, acss) self._speak(text, acss)
def sayAll(self, utteranceIterator, progressCallback): def sayAll(self, utteranceIterator, progressCallback):
GLib.idle_add(self._say_all, utteranceIterator, progressCallback) self._say_all_generation += 1
self._cancelled_say_all_generation = None
GLib.idle_add(
self._say_all,
utteranceIterator,
progressCallback,
self._say_all_generation,
)
def speakCharacter(self, character, acss=None): def speakCharacter(self, character, acss=None):
self._apply_acss(acss) self._apply_acss(acss)
+79
View File
@@ -0,0 +1,79 @@
import sys
import unittest
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cthulhu import debug
class DebugStartupTimingTests(unittest.TestCase):
def setUp(self) -> None:
self._debugLevel = debug.debugLevel
self._startupTimingEnabled = debug.startupTimingEnabled
self._startupTimingStartTime = debug._startupTimingStartTime
self._startupTimingLastTime = debug._startupTimingLastTime
debug.debugLevel = debug.LEVEL_SEVERE
debug.startupTimingEnabled = False
debug._startupTimingStartTime = None
debug._startupTimingLastTime = None
def tearDown(self) -> None:
debug.debugLevel = self._debugLevel
debug.startupTimingEnabled = self._startupTimingEnabled
debug._startupTimingStartTime = self._startupTimingStartTime
debug._startupTimingLastTime = self._startupTimingLastTime
def test_startup_timing_is_silent_when_disabled_and_debug_is_not_enabled(self) -> None:
with mock.patch.object(debug, "println") as println:
debug.print_startup_timing("hidden phase")
println.assert_not_called()
def test_startup_timing_logs_when_explicitly_enabled(self) -> None:
with mock.patch.object(debug.time, "monotonic", side_effect=[10.0, 10.25]):
debug.set_startup_timing_enabled(True)
with mock.patch.object(debug, "println") as println:
debug.print_startup_timing("explicit phase")
println.assert_called_once_with(
debug.LEVEL_SEVERE,
"STARTUP TIMING: explicit phase: +0.250s, total 0.250s",
True,
False,
)
def test_startup_timing_logs_when_debug_output_is_enabled(self) -> None:
debug.debugLevel = debug.LEVEL_INFO
with mock.patch.object(debug.time, "monotonic", return_value=20.0):
with mock.patch.object(debug, "println") as println:
debug.print_startup_timing("debug phase")
println.assert_called_once_with(
debug.LEVEL_INFO,
"STARTUP TIMING: debug phase: +0.000s, total 0.000s",
True,
False,
)
def test_startup_timing_context_logs_start_and_end(self) -> None:
with mock.patch.object(debug.time, "monotonic", side_effect=[30.0, 30.1, 30.4]):
debug.set_startup_timing_enabled(True)
with mock.patch.object(debug, "println") as println:
with debug.startup_timing("wrapped phase"):
pass
self.assertEqual(2, println.call_count)
self.assertEqual(
"STARTUP TIMING: wrapped phase start: +0.100s, total 0.100s",
println.call_args_list[0].args[1],
)
self.assertEqual(
"STARTUP TIMING: wrapped phase end: +0.300s, total 0.400s",
println.call_args_list[1].args[1],
)
if __name__ == "__main__":
unittest.main()
@@ -20,7 +20,7 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
): ):
self.assertTrue(manager._active_x11_window_differs_from(cachedWindow)) self.assertTrue(manager._active_x11_window_differs_from(cachedWindow))
def test_keyboard_event_preserves_key_handling_when_unknown_x11_window_is_not_xterm(self): def test_keyboard_event_clears_stale_context_before_recovering_old_focus_window(self):
manager = input_event_manager.InputEventManager() manager = input_event_manager.InputEventManager()
staleWindow = object() staleWindow = object()
staleFocus = object() staleFocus = object()
@@ -30,6 +30,8 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
scriptManager = mock.Mock() scriptManager = mock.Mock()
keyboardEvent = mock.Mock() keyboardEvent = mock.Mock()
keyboardEvent.is_modifier_key.return_value = False keyboardEvent.is_modifier_key.return_value = False
manager._last_input_event = object()
manager._last_non_modifier_key_event = object()
with ( with (
mock.patch.object(input_event_manager.cthulhu_state, "capturingKeys", False), mock.patch.object(input_event_manager.cthulhu_state, "capturingKeys", False),
@@ -38,7 +40,8 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
mock.patch.object(input_event_manager.input_event, "KeyboardEvent", return_value=keyboardEvent), mock.patch.object(input_event_manager.input_event, "KeyboardEvent", return_value=keyboardEvent),
mock.patch.object(input_event_manager.AXUtilities, "can_be_active_window", return_value=False), mock.patch.object(input_event_manager.AXUtilities, "can_be_active_window", return_value=False),
mock.patch.object(input_event_manager.AXUtilities, "find_active_window", return_value=None), mock.patch.object(input_event_manager.AXUtilities, "find_active_window", return_value=None),
mock.patch.object(manager, "_get_top_level_window", return_value=None), mock.patch.object(manager, "_find_active_x11_atspi_window", return_value=None),
mock.patch.object(manager, "_get_top_level_window", return_value=staleWindow),
mock.patch.object(manager, "_should_pass_through_for_active_xterm", return_value=False), mock.patch.object(manager, "_should_pass_through_for_active_xterm", return_value=False),
mock.patch.object(manager, "_active_x11_window_differs_from", return_value=True), mock.patch.object(manager, "_active_x11_window_differs_from", return_value=True),
mock.patch.object(manager, "last_event_was_keyboard", return_value=False), mock.patch.object(manager, "last_event_was_keyboard", return_value=False),
@@ -53,12 +56,18 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
"Return", "Return",
) )
self.assertTrue(result) self.assertFalse(result)
scriptManager.set_active_script.assert_not_called() scriptManager.set_active_script.assert_called_once_with(
focusManager.clear_state.assert_not_called() None,
keyboardEvent.process.assert_called_once_with() "active X11 window not found in AT-SPI",
)
focusManager.clear_state.assert_called_once_with("active X11 window not found in AT-SPI")
focusManager.set_active_window.assert_not_called()
keyboardEvent.process.assert_not_called()
self.assertIsNone(manager._last_input_event)
self.assertIsNone(manager._last_non_modifier_key_event)
def test_keyboard_event_uses_active_script_app_when_cached_window_is_missing(self): def test_keyboard_event_clears_stale_script_when_cached_context_is_missing(self):
manager = input_event_manager.InputEventManager() manager = input_event_manager.InputEventManager()
staleApp = object() staleApp = object()
staleScript = mock.Mock(app=staleApp) staleScript = mock.Mock(app=staleApp)
@@ -77,6 +86,7 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
mock.patch.object(input_event_manager.input_event, "KeyboardEvent", return_value=keyboardEvent), mock.patch.object(input_event_manager.input_event, "KeyboardEvent", return_value=keyboardEvent),
mock.patch.object(input_event_manager.AXUtilities, "can_be_active_window", return_value=False), mock.patch.object(input_event_manager.AXUtilities, "can_be_active_window", return_value=False),
mock.patch.object(input_event_manager.AXUtilities, "find_active_window", return_value=None), mock.patch.object(input_event_manager.AXUtilities, "find_active_window", return_value=None),
mock.patch.object(manager, "_find_active_x11_atspi_window", return_value=None),
mock.patch.object(manager, "_get_top_level_window", return_value=None), mock.patch.object(manager, "_get_top_level_window", return_value=None),
mock.patch.object(manager, "_should_pass_through_for_active_xterm", return_value=False), mock.patch.object(manager, "_should_pass_through_for_active_xterm", return_value=False),
mock.patch.object(manager, "_active_x11_window_differs_from", return_value=True) as differs, mock.patch.object(manager, "_active_x11_window_differs_from", return_value=True) as differs,
@@ -92,8 +102,51 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
"KP_Page_Up", "KP_Page_Up",
) )
self.assertTrue(result) self.assertFalse(result)
differs.assert_called_once_with(staleApp) differs.assert_called_once_with(staleApp)
scriptManager.set_active_script.assert_called_once_with(
None,
"active X11 window not found in AT-SPI",
)
focusManager.clear_state.assert_called_once_with("active X11 window not found in AT-SPI")
keyboardEvent.process.assert_not_called()
def test_keyboard_event_recovers_focus_window_when_x11_does_not_contradict_it(self):
manager = input_event_manager.InputEventManager()
cachedWindow = object()
staleFocus = object()
focusManager = mock.Mock()
focusManager.get_active_window.return_value = cachedWindow
focusManager.get_locus_of_focus.return_value = staleFocus
scriptManager = mock.Mock()
keyboardEvent = mock.Mock()
keyboardEvent.is_modifier_key.return_value = False
with (
mock.patch.object(input_event_manager.cthulhu_state, "capturingKeys", False),
mock.patch.object(input_event_manager.focus_manager, "get_manager", return_value=focusManager),
mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager),
mock.patch.object(input_event_manager.input_event, "KeyboardEvent", return_value=keyboardEvent),
mock.patch.object(input_event_manager.AXUtilities, "can_be_active_window", return_value=False),
mock.patch.object(input_event_manager.AXUtilities, "find_active_window", return_value=None),
mock.patch.object(manager, "_find_active_x11_atspi_window", return_value=None),
mock.patch.object(manager, "_get_top_level_window", return_value=cachedWindow),
mock.patch.object(manager, "_should_pass_through_for_active_xterm", return_value=False),
mock.patch.object(manager, "_active_x11_window_differs_from", return_value=False),
mock.patch.object(manager, "last_event_was_keyboard", return_value=False),
mock.patch.object(input_event_manager.debug, "print_message"),
):
result = manager.process_keyboard_event(
mock.Mock(),
True,
36,
65293,
0,
"Return",
)
self.assertTrue(result)
focusManager.set_active_window.assert_called_once_with(cachedWindow)
scriptManager.set_active_script.assert_not_called() scriptManager.set_active_script.assert_not_called()
focusManager.clear_state.assert_not_called() focusManager.clear_state.assert_not_called()
keyboardEvent.process.assert_called_once_with() keyboardEvent.process.assert_called_once_with()
@@ -226,6 +279,52 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
activeScript.addKeyGrabs.assert_not_called() activeScript.addKeyGrabs.assert_not_called()
keyboardEvent.process.assert_not_called() keyboardEvent.process.assert_not_called()
def test_xterm_pass_through_ignores_stale_pending_self_hosted_focus(self):
manager = input_event_manager.InputEventManager()
pendingFocus = object()
focusManager = mock.Mock()
focusManager.get_active_window.return_value = None
focusManager.get_locus_of_focus.return_value = None
scriptManager = mock.Mock()
activeScript = mock.Mock(app=None)
scriptManager.get_active_script.return_value = activeScript
keyboardEvent = mock.Mock()
with (
mock.patch.object(input_event_manager.cthulhu_state, "capturingKeys", False),
mock.patch.object(input_event_manager.cthulhu_state, "pendingSelfHostedFocus", pendingFocus),
mock.patch.object(input_event_manager.focus_manager, "get_manager", return_value=focusManager),
mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager),
mock.patch.object(input_event_manager.input_event, "KeyboardEvent", return_value=keyboardEvent),
mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=True),
mock.patch.object(input_event_manager.debug, "print_message"),
mock.patch.object(input_event_manager.debug, "print_tokens"),
):
result = manager.process_keyboard_event(
mock.Mock(),
True,
90,
65438,
0,
"KP_Insert",
)
self.assertFalse(result)
activeScript.removeKeyGrabs.assert_called_once_with()
keyboardEvent.process.assert_not_called()
def test_pending_self_hosted_focus_blocks_unknown_xterm_match(self):
manager = input_event_manager.InputEventManager()
with (
mock.patch.object(manager, "_active_x11_window_xterm_match", return_value=None),
mock.patch.object(input_event_manager.debug, "print_message"),
mock.patch.object(input_event_manager.debug, "print_tokens"),
):
result = manager._should_pass_through_for_active_xterm(None, None, object())
self.assertFalse(result)
def test_xterm_grabs_restore_when_active_window_is_positively_not_xterm(self): def test_xterm_grabs_restore_when_active_window_is_positively_not_xterm(self):
manager = input_event_manager.InputEventManager() manager = input_event_manager.InputEventManager()
focusManager = mock.Mock() focusManager = mock.Mock()
@@ -264,6 +363,24 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
activeScript.addKeyGrabs.assert_called_once_with() activeScript.addKeyGrabs.assert_called_once_with()
keyboardEvent.process.assert_called_once_with() keyboardEvent.process.assert_called_once_with()
def test_unidentified_active_x11_window_is_unknown_not_non_xterm(self):
manager = input_event_manager.InputEventManager()
window = mock.Mock()
window.get_class_group_name.return_value = None
window.get_class_instance_name.return_value = None
window.get_name.return_value = None
window.get_class_group.return_value = None
window.get_pid.return_value = -1
with (
mock.patch.object(manager, "_get_active_x11_window", return_value=window),
mock.patch.object(input_event_manager.debug, "print_message"),
mock.patch.object(input_event_manager.debug, "print_tokens"),
):
result = manager._active_x11_window_xterm_match()
self.assertIsNone(result)
def test_finds_focused_atspi_window_for_active_x11_pid(self): def test_finds_focused_atspi_window_for_active_x11_pid(self):
manager = input_event_manager.InputEventManager() manager = input_event_manager.InputEventManager()
app = object() app = object()
+48
View File
@@ -1,10 +1,13 @@
import sys import sys
import threading
import unittest import unittest
from pathlib import Path from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cthulhu import piperfactory from cthulhu import piperfactory
from cthulhu import speechserver
class PiperFactoryRateMappingTests(unittest.TestCase): class PiperFactoryRateMappingTests(unittest.TestCase):
@@ -25,6 +28,51 @@ class PiperFactoryRateMappingTests(unittest.TestCase):
self.assertEqual(2.0, self.server._mapRate(-1)) self.assertEqual(2.0, self.server._mapRate(-1))
self.assertEqual(0.25, self.server._mapRate(101)) self.assertEqual(0.25, self.server._mapRate(101))
def _make_say_all_server(self):
server = piperfactory.SpeechServer.__new__(piperfactory.SpeechServer)
server._lock = threading.Lock()
server._stopEvent = threading.Event()
server._speakGeneration = 1
server._currentFuture = None
server._executor = mock.Mock()
server._audioPlayer = None
return server
def _make_say_all_context(self, utterance):
return speechserver.SayAllContext(
mock.Mock(),
utterance,
0,
len(utterance),
)
def test_stop_invalidates_pending_say_all_worker(self):
server = self._make_say_all_server()
iterator = iter([[self._make_say_all_context("first"), None]])
callback = mock.Mock()
server.stop()
result = server._sayAllWorker(iterator, callback, 1)
self.assertFalse(result)
callback.assert_not_called()
server._executor.submit.assert_not_called()
def test_stale_say_all_completion_does_not_advance(self):
server = self._make_say_all_server()
first = self._make_say_all_context("first")
second = self._make_say_all_context("second")
iterator = iter([[first, None], [second, None]])
callback = mock.Mock()
server._sayAllWorker(iterator, callback, 1)
onComplete = server._executor.submit.call_args.args[3]
server.stop()
result = onComplete()
self.assertFalse(result)
self.assertEqual(1, server._executor.submit.call_count)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
@@ -137,6 +137,21 @@ class PluginSystemManagerRegressionTests(unittest.TestCase):
self.assertEqual(instance.deactivation_count, 1) self.assertEqual(instance.deactivation_count, 1)
self.assertEqual(plugin_info.instance, None) self.assertEqual(plugin_info.instance, None)
@mock.patch("cthulhu.plugin_system_manager.dbus_service.get_remote_controller")
def test_obsolete_plugin_manager_active_entry_is_dropped(self, remote_controller):
remote_controller.return_value = mock.Mock()
with tempfile.TemporaryDirectory() as temp_dir:
manager = self._create_manager()
plugin_info = self._create_plugin_info(temp_dir, "OCR")
manager._plugins["OCR"] = plugin_info
manager._plugin_name_index["OCR"] = ["OCR"]
with mock.patch.object(manager, "syncAllPluginsActive"):
manager.setActivePlugins(["PluginManager", "OCR"])
self.assertEqual(manager.getActivePlugins(), ["OCR"])
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+148
View File
@@ -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()
@@ -7,6 +7,7 @@ from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cthulhu import speechdispatcherfactory from cthulhu import speechdispatcherfactory
from cthulhu import speechserver
import speechd import speechd
@@ -20,8 +21,24 @@ class SpeechDispatcherInterruptRegressionTests(unittest.TestCase):
server._speak = mock.Mock() server._speak = mock.Mock()
server._client = mock.Mock() server._client = mock.Mock()
server._client.char = mock.Mock() server._client.char = mock.Mock()
server._say_all_generation = 0
server._cancelled_say_all_generation = None
server._CALLBACK_TYPE_MAP = {
speechd.CallbackType.BEGIN: speechserver.SayAllContext.PROGRESS,
speechd.CallbackType.CANCEL: speechserver.SayAllContext.INTERRUPTED,
speechd.CallbackType.END: speechserver.SayAllContext.COMPLETED,
speechd.CallbackType.INDEX_MARK: speechserver.SayAllContext.PROGRESS,
}
return server return server
def _make_say_all_context(self, utterance):
return speechserver.SayAllContext(
mock.Mock(),
utterance,
0,
len(utterance),
)
def test_interrupting_string_speech_cancels_active_output_first(self): def test_interrupting_string_speech_cancels_active_output_first(self):
server = self._make_server() server = self._make_server()
@@ -47,6 +64,41 @@ class SpeechDispatcherInterruptRegressionTests(unittest.TestCase):
server._cancel.assert_not_called() server._cancel.assert_not_called()
server._speak.assert_called_once_with("next", None) server._speak.assert_called_once_with("next", None)
def test_stop_invalidates_pending_say_all_continuation(self):
server = self._make_server()
first = self._make_say_all_context("first")
second = self._make_say_all_context("second")
callback = mock.Mock()
iterator = iter([[first, None], [second, None]])
server.sayAll(iterator, callback)
server._say_all(iterator, callback, 1)
speechd_callback = server._speak.call_args.kwargs["callback"]
server.stop()
speechd_callback(speechd.CallbackType.END)
self.assertEqual(1, server._speak.call_count)
callback.assert_not_called()
def test_interrupting_speech_invalidates_pending_say_all_continuation(self):
server = self._make_server()
first = self._make_say_all_context("first")
second = self._make_say_all_context("second")
callback = mock.Mock()
iterator = iter([[first, None], [second, None]])
server.sayAll(iterator, callback)
server._say_all(iterator, callback, 1)
speechd_callback = server._speak.call_args.kwargs["callback"]
server.speak("new speech", interrupt=True)
speechd_callback(speechd.CallbackType.END)
self.assertEqual(2, server._speak.call_count)
self.assertEqual("new speech", server._speak.call_args.args[0])
callback.assert_not_called()
def test_send_command_logs_and_recovers_from_command_errors(self): def test_send_command_logs_and_recovers_from_command_errors(self):
server = self._make_server() server = self._make_server()
server.reset = mock.Mock() server.reset = mock.Mock()
+39
View File
@@ -96,6 +96,45 @@ class SteamReturnActivationTests(unittest.TestCase):
class SteamVirtualizedListMutationTests(unittest.TestCase): class SteamVirtualizedListMutationTests(unittest.TestCase):
def test_caret_moved_ignores_transient_loading_spinner(self):
testScript = steam_script.Script.__new__(steam_script.Script)
source = object()
spinner = object()
event = mock.Mock(type="object:text-caret-moved", source=source)
testScript.utilities = mock.Mock()
testScript.utilities.inDocumentContent.return_value = True
with (
mock.patch.object(steam_script.AXObject, "get_name", side_effect=["", "Steam Spinner"]),
mock.patch.object(steam_script.AXObject, "iter_children", return_value=iter([spinner])),
mock.patch.object(steam_script.AXUtilities, "is_image_or_canvas", return_value=True),
mock.patch.object(steam_script.debug, "printMessage"),
mock.patch.object(steam_script.Chromium.Script, "onCaretMoved") as chromiumCaretMoved,
):
self.assertTrue(testScript.onCaretMoved(event))
chromiumCaretMoved.assert_not_called()
def test_caret_moved_defers_to_chromium_without_loading_spinner(self):
testScript = steam_script.Script.__new__(steam_script.Script)
source = object()
child = object()
event = mock.Mock(type="object:text-caret-moved", source=source)
testScript.utilities = mock.Mock()
testScript.utilities.inDocumentContent.return_value = True
with (
mock.patch.object(steam_script.AXObject, "get_name", side_effect=["", "Not A Spinner"]),
mock.patch.object(steam_script.AXObject, "iter_children", return_value=iter([child])),
mock.patch.object(steam_script.AXUtilities, "is_image_or_canvas", return_value=True),
mock.patch.object(steam_script.Chromium.Script, "onCaretMoved", return_value="handled") as chromiumCaretMoved,
):
self.assertEqual(testScript.onCaretMoved(event), "handled")
chromiumCaretMoved.assert_called_once_with(event)
def test_children_added_skips_generic_web_cache_dump_for_virtualized_list_churn(self): def test_children_added_skips_generic_web_cache_dump_for_virtualized_list_churn(self):
testScript = steam_script.Script.__new__(steam_script.Script) testScript = steam_script.Script.__new__(steam_script.Script)
source = object() source = object()
@@ -39,7 +39,7 @@ color-calculation-max = 3
copy-to-clipboard = false copy-to-clipboard = false
[profiles.default.plugins] [profiles.default.plugins]
active-plugins = ["PluginManager", "OCR"] active-plugins = ["Clipboard", "OCR"]
plugin-sources = [] plugin-sources = []
""" """
@@ -64,7 +64,7 @@ class LegacyTomlSchemaMigrationTests(unittest.TestCase):
settings.GENERAL_KEYBOARD_LAYOUT_DESKTOP, settings.GENERAL_KEYBOARD_LAYOUT_DESKTOP,
) )
self.assertEqual(general["cthulhuModifierKeys"], settings.DESKTOP_MODIFIER_KEYS) self.assertEqual(general["cthulhuModifierKeys"], settings.DESKTOP_MODIFIER_KEYS)
self.assertEqual(general["activePlugins"], ["PluginManager", "OCR"]) self.assertEqual(general["activePlugins"], ["Clipboard", "OCR"])
self.assertEqual(general["aiProvider"], settings.AI_PROVIDER_OLLAMA) self.assertEqual(general["aiProvider"], settings.AI_PROVIDER_OLLAMA)
self.assertFalse(general["aiAssistantEnabled"]) self.assertFalse(general["aiAssistantEnabled"])
self.assertEqual(general["ocrLanguageCode"], "eng") self.assertEqual(general["ocrLanguageCode"], "eng")
@@ -82,7 +82,7 @@ class LegacyTomlSchemaMigrationTests(unittest.TestCase):
savedSettings = settingsPath.read_text(encoding="utf-8") savedSettings = settingsPath.read_text(encoding="utf-8")
self.assertIn('profile = ["Default", "default"]', savedSettings) self.assertIn('profile = ["Default", "default"]', savedSettings)
self.assertIn('activePlugins = ["PluginManager", "OCR"]', savedSettings) self.assertIn('activePlugins = ["Clipboard", "OCR"]', savedSettings)
self.assertNotIn("format-version = 2", savedSettings) self.assertNotIn("format-version = 2", savedSettings)
self.assertNotIn("[profiles.default.metadata]", savedSettings) self.assertNotIn("[profiles.default.metadata]", savedSettings)
+457
View File
@@ -10,6 +10,9 @@ os.environ.setdefault("GSETTINGS_BACKEND", "memory")
gi.require_version("Gdk", "3.0") gi.require_version("Gdk", "3.0")
gi.require_version("Gtk", "3.0") gi.require_version("Gtk", "3.0")
gi.require_version("Atspi", "2.0")
from gi.repository import Atspi
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
@@ -23,6 +26,7 @@ if soundGeneratorModule is not None and not hasattr(soundGeneratorModule, "Sound
from cthulhu import messages from cthulhu import messages
from cthulhu.scripts.web import script as web_script from cthulhu.scripts.web import script as web_script
from cthulhu.scripts.web import script_utilities as web_script_utilities from cthulhu.scripts.web import script_utilities as web_script_utilities
from cthulhu.scripts.toolkits.Gecko import script_utilities as gecko_script_utilities
class WebKeyGrabRegressionTests(unittest.TestCase): class WebKeyGrabRegressionTests(unittest.TestCase):
@@ -34,14 +38,33 @@ class WebKeyGrabRegressionTests(unittest.TestCase):
testScript._lastMouseButtonContext = ("old", 7) testScript._lastMouseButtonContext = ("old", 7)
testScript._madeFindAnnouncement = True testScript._madeFindAnnouncement = True
testScript._inFocusMode = True testScript._inFocusMode = True
testScript._focusModeIsSticky = False
testScript._browseModeIsSticky = False
testScript._navSuspended = False testScript._navSuspended = False
testScript.removeKeyGrabs = mock.Mock() testScript.removeKeyGrabs = mock.Mock()
testScript.refreshKeyGrabs = mock.Mock() testScript.refreshKeyGrabs = mock.Mock()
testScript._setNavigationSuspended = mock.Mock() testScript._setNavigationSuspended = mock.Mock()
testScript.presentMessage = mock.Mock()
testScript.utilities = mock.Mock() testScript.utilities = mock.Mock()
testScript.utilities.isZombie.return_value = False testScript.utilities.isZombie.return_value = False
testScript.utilities.isDocument.return_value = False testScript.utilities.isDocument.return_value = False
testScript.utilities.getTopLevelDocumentForObject.return_value = None testScript.utilities.getTopLevelDocumentForObject.return_value = None
testScript.utilities.inFindContainer.return_value = False
testScript.utilities.getCaretContext.return_value = (None, -1)
testScript.utilities.queryNonEmptyText.return_value = None
testScript.utilities.isContentEditableWithEmbeddedObjects.return_value = False
testScript.utilities.isAnchor.return_value = False
testScript.utilities.lastInputEventWasPageNav.return_value = False
testScript.utilities.isFocusedWithMathChild.return_value = False
testScript.utilities.caretMovedToSamePageFragment.return_value = False
testScript.utilities.lastInputEventWasLineNav.return_value = False
testScript.utilities.shouldInterruptForLocusOfFocusChange.return_value = False
testScript.flatReviewPresenter = mock.Mock()
testScript.flatReviewPresenter.is_active.return_value = False
testScript.updateBraille = mock.Mock()
testScript.speechGenerator = mock.Mock()
testScript.speechGenerator.generateSpeech.return_value = []
testScript._saveFocusedObjectInfo = mock.Mock()
return testScript return testScript
def test_window_deactivate_does_not_drop_key_grabs(self): def test_window_deactivate_does_not_drop_key_grabs(self):
@@ -99,6 +122,44 @@ class WebKeyGrabRegressionTests(unittest.TestCase):
self.assertFalse(result) self.assertFalse(result)
testScript.refreshKeyGrabs.assert_called_once_with() testScript.refreshKeyGrabs.assert_called_once_with()
def test_non_document_focus_preserves_sticky_focus_mode(self):
testScript = self._make_partial_script()
testScript._focusModeIsSticky = True
oldFocus = object()
newFocus = object()
with mock.patch("cthulhu.scripts.web.script.AXObject.is_dead", return_value=False):
result = web_script.Script.locus_of_focus_changed(testScript, None, oldFocus, newFocus)
self.assertFalse(result)
self.assertTrue(testScript._inFocusMode)
def test_document_focus_restores_sticky_focus_after_suspension(self):
testScript = self._make_partial_script()
testScript._inFocusMode = False
testScript._focusModeIsSticky = True
testScript._navSuspended = True
oldFocus = object()
newFocus = object()
document = object()
testScript.utilities.getTopLevelDocumentForObject.side_effect = (
lambda obj: document if obj is newFocus else None
)
with (
mock.patch("cthulhu.scripts.web.script.AXObject.is_dead", return_value=False),
mock.patch("cthulhu.scripts.web.script.AXUtilities.is_unknown_or_redundant", return_value=False),
mock.patch("cthulhu.scripts.web.script.AXUtilities.is_heading", return_value=False),
mock.patch("cthulhu.scripts.web.script.speech.speak"),
mock.patch("cthulhu.scripts.web.script.cthulhu.emitRegionChanged"),
):
result = web_script.Script.locus_of_focus_changed(testScript, None, oldFocus, newFocus)
self.assertTrue(result)
self.assertTrue(testScript._inFocusMode)
testScript.presentMessage.assert_called_with(messages.MODE_FOCUS_IS_STICKY)
testScript.refreshKeyGrabs.assert_called_once_with()
class WebActiveWindowRegressionTests(unittest.TestCase): class WebActiveWindowRegressionTests(unittest.TestCase):
def test_sanity_check_recovers_missing_script_app_from_active_window(self): def test_sanity_check_recovers_missing_script_app_from_active_window(self):
@@ -272,6 +333,34 @@ class WebFocusSpeechInterruptionRegressionTests(unittest.TestCase):
speak.assert_called_once_with(["focus speech"], interrupt=False) 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): class WebDescriptionChangeRegressionTests(unittest.TestCase):
def _make_script(self): def _make_script(self):
testScript = web_script.Script.__new__(web_script.Script) testScript = web_script.Script.__new__(web_script.Script)
@@ -395,6 +484,52 @@ class WebDynamicContentRecoveryRegressionTests(unittest.TestCase):
setLocusOfFocus.assert_called_once_with(event, source, False) setLocusOfFocus.assert_called_once_with(event, source, False)
testScript.utilities.setCaretContext.assert_called_once_with(source, 0) 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()
event = mock.Mock(detail1=1, source=source)
testScript._browseModeIsSticky = True
testScript.utilities.getDocumentForObject.return_value = "document"
testScript.utilities.isWebAppDescendant.return_value = True
with (
mock.patch.object(web_script.cthulhu_state, "locusOfFocus", object()),
mock.patch.object(web_script.cthulhu, "setLocusOfFocus") as setLocusOfFocus,
):
result = web_script.Script.onFocusedChanged(testScript, event)
self.assertTrue(result)
setLocusOfFocus.assert_not_called()
class WebContextReplicantRegressionTests(unittest.TestCase): class WebContextReplicantRegressionTests(unittest.TestCase):
def test_same_object_replicant_can_recover_before_old_focus_is_dead(self): def test_same_object_replicant_can_recover_before_old_focus_is_dead(self):
@@ -561,3 +696,325 @@ class WebSelectionRegressionTests(unittest.TestCase):
result = web_script_utilities.Utilities.allSelectedText(utilities, obj) result = web_script_utilities.Utilities.allSelectedText(utilities, obj)
self.assertEqual(result, ["Dark", 0, 4]) 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 WebSkipSpaceCaretNavigationRegressionTests(unittest.TestCase):
def _make_utilities(self):
utilities = web_script_utilities.Utilities.__new__(web_script_utilities.Utilities)
utilities.getCaretContext = mock.Mock(return_value=(None, -1))
utilities.treatAsTextObject = mock.Mock(return_value=True)
return utilities
def test_next_context_does_not_skip_named_whole_object_as_space(self):
utilities = self._make_utilities()
start = object()
wholeObject = object()
after = object()
utilities.treatAsTextObject.return_value = False
utilities.findNextCaretInOrder = mock.Mock(
side_effect=[(wholeObject, 0), (after, 0)]
)
with mock.patch.object(
web_script_utilities.AXText,
"get_character_at_offset",
return_value=(" ", 0, 1),
) as getCharacter:
result = web_script_utilities.Utilities.nextContext(
utilities,
start,
0,
skipSpace=True,
)
self.assertEqual(result, (wholeObject, 0))
utilities.treatAsTextObject.assert_called_once_with(wholeObject)
getCharacter.assert_not_called()
utilities.findNextCaretInOrder.assert_called_once_with(start, 0)
def test_previous_context_does_not_skip_named_whole_object_as_space(self):
utilities = self._make_utilities()
start = object()
wholeObject = object()
before = object()
utilities.treatAsTextObject.return_value = False
utilities.findPreviousCaretInOrder = mock.Mock(
side_effect=[(wholeObject, 0), (before, 0)]
)
with mock.patch.object(
web_script_utilities.AXText,
"get_character_at_offset",
return_value=(" ", 0, 1),
) as getCharacter:
result = web_script_utilities.Utilities.previousContext(
utilities,
start,
0,
skipSpace=True,
)
self.assertEqual(result, (wholeObject, 0))
utilities.treatAsTextObject.assert_called_once_with(wholeObject)
getCharacter.assert_not_called()
utilities.findPreviousCaretInOrder.assert_called_once_with(start, 0)
def test_next_context_breaks_skip_space_cycles(self):
utilities = self._make_utilities()
start = object()
first = object()
second = object()
utilities.findNextCaretInOrder = mock.Mock(
side_effect=[(first, 0), (second, 0), (first, 0)]
)
with mock.patch.object(
web_script_utilities.AXText,
"get_character_at_offset",
return_value=(" ", 0, 1),
):
result = web_script_utilities.Utilities.nextContext(
utilities,
start,
0,
skipSpace=True,
)
self.assertEqual(result, (first, 0))
self.assertEqual(utilities.findNextCaretInOrder.call_args_list, [
mock.call(start, 0),
mock.call(first, 0),
mock.call(second, 0),
])
def test_previous_context_breaks_skip_space_cycles(self):
utilities = self._make_utilities()
start = object()
first = object()
second = object()
utilities.findPreviousCaretInOrder = mock.Mock(
side_effect=[(first, 0), (second, 0), (first, 0)]
)
with mock.patch.object(
web_script_utilities.AXText,
"get_character_at_offset",
return_value=(" ", 0, 1),
):
result = web_script_utilities.Utilities.previousContext(
utilities,
start,
0,
skipSpace=True,
)
self.assertEqual(result, (first, 0))
self.assertEqual(utilities.findPreviousCaretInOrder.call_args_list, [
mock.call(start, 0),
mock.call(first, 0),
mock.call(second, 0),
])
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),
],
)
class WebGeckoBrokenTextRecoveryRegressionTests(unittest.TestCase):
def test_gecko_limits_broken_text_recovery_to_sentences(self):
utilities = gecko_script_utilities.Utilities.__new__(gecko_script_utilities.Utilities)
utilities.isContentEditableWithEmbeddedObjects = mock.Mock(return_value=True)
obj = object()
self.assertTrue(
gecko_script_utilities.Utilities._attemptBrokenTextRecovery(
utilities,
obj,
boundary=Atspi.TextBoundaryType.SENTENCE_START,
)
)
self.assertFalse(
gecko_script_utilities.Utilities._attemptBrokenTextRecovery(
utilities,
obj,
boundary=Atspi.TextBoundaryType.LINE_START,
)
)
self.assertFalse(
gecko_script_utilities.Utilities._attemptBrokenTextRecovery(
utilities,
obj,
boundary=Atspi.TextBoundaryType.WORD_START,
)
)