2 Commits

Author SHA1 Message Date
Storm Dragon 0576b6f79f Release candidate. 2026-07-17 13:39:18 -04:00
Storm Dragon 40aeadb68a Tighter integration with i38lock. Found and squished a bug with sound sometimes not loading. 2026-07-17 13:38:01 -04:00
12 changed files with 537 additions and 4 deletions
+12
View File
@@ -257,6 +257,7 @@ from . import keybindings
from . import learn_mode_presenter from . import learn_mode_presenter
from . import logger from . import logger
from . import mako_notification_monitor from . import mako_notification_monitor
from . import screen_lock_monitor
from . import messages from . import messages
from . import notification_presenter from . import notification_presenter
from . import focus_manager from . import focus_manager
@@ -724,6 +725,7 @@ def start() -> None:
# 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"): with debug.startup_timing("startup idle services"):
cthulhuApp.getMakoNotificationMonitor().start() cthulhuApp.getMakoNotificationMonitor().start()
cthulhuApp.getScreenLockMonitor().start()
GObject.idle_add(_start_dbus_service) GObject.idle_add(_start_dbus_service)
debug.print_startup_timing("AT-SPI main loop start") debug.print_startup_timing("AT-SPI main loop start")
@@ -772,11 +774,13 @@ def shutdown(script: Optional[Any] = None, inputEvent: Optional[Any] = None) ->
signal.signal(signal.SIGALRM, settings.timeoutCallback) signal.signal(signal.SIGALRM, settings.timeoutCallback)
signal.alarm(settings.timeoutTime) signal.alarm(settings.timeoutTime)
if cthulhu_state.activeScript is not None:
cthulhu_state.activeScript.presentationInterrupt() cthulhu_state.activeScript.presentationInterrupt()
cthulhuApp.getSignalManager().emitSignal('stop-application-completed') cthulhuApp.getSignalManager().emitSignal('stop-application-completed')
sound_theme_manager.getManager().playStopSound(wait=True, timeoutSeconds=1) sound_theme_manager.getManager().playStopSound(wait=True, timeoutSeconds=1)
cthulhuApp.getPluginSystemManager().unloadAllPlugins(ForceAllPlugins=True) cthulhuApp.getPluginSystemManager().unloadAllPlugins(ForceAllPlugins=True)
cthulhuApp.getScreenLockMonitor().stop()
cthulhuApp.getMakoNotificationMonitor().stop() cthulhuApp.getMakoNotificationMonitor().stop()
# Deactivate the event manager first so that it clears its queue and will not # Deactivate the event manager first so that it clears its queue and will not
@@ -991,6 +995,9 @@ class Cthulhu(GObject.Object):
with debug.startup_timing("notification monitor construction"): 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("screen lock monitor construction"):
self.screenLockMonitor: screen_lock_monitor.ScreenLockMonitor = \
screen_lock_monitor.ScreenLockMonitor(self)
with debug.startup_timing("compat API registration"): with debug.startup_timing("compat API registration"):
self.createCompatAPI() self.createCompatAPI()
with debug.startup_timing("plugin manager construction"): with debug.startup_timing("plugin manager construction"):
@@ -1009,6 +1016,9 @@ class Cthulhu(GObject.Object):
def getMakoNotificationMonitor(self) -> mako_notification_monitor.MakoNotificationMonitor: def getMakoNotificationMonitor(self) -> mako_notification_monitor.MakoNotificationMonitor:
return self.makoNotificationMonitor return self.makoNotificationMonitor
def getScreenLockMonitor(self) -> screen_lock_monitor.ScreenLockMonitor:
return self.screenLockMonitor
def getDynamicApiManager(self) -> DynamicApiManager: def getDynamicApiManager(self) -> DynamicApiManager:
return self.dynamicApiManager return self.dynamicApiManager
@@ -1061,6 +1071,7 @@ class Cthulhu(GObject.Object):
"speech_server_present": speechServer is not None, "speech_server_present": speechServer is not None,
"speech_server": speechServer.__class__.__name__ if speechServer else "", "speech_server": speechServer.__class__.__name__ if speechServer else "",
}, },
"sound": sound.get_debug_snapshot(),
"event_manager": self.eventManager.get_debug_snapshot(), "event_manager": self.eventManager.get_debug_snapshot(),
"script_manager": self.scriptManager.get_debug_snapshot(), "script_manager": self.scriptManager.get_debug_snapshot(),
"input_event_manager": input_event_manager.get_manager().get_debug_snapshot(), "input_event_manager": input_event_manager.get_manager().get_debug_snapshot(),
@@ -1097,6 +1108,7 @@ class Cthulhu(GObject.Object):
self.getDynamicApiManager().registerAPI('Cmdnames', cmdnames) self.getDynamicApiManager().registerAPI('Cmdnames', cmdnames)
self.getDynamicApiManager().registerAPI('NotificationPresenter', notification_presenter) self.getDynamicApiManager().registerAPI('NotificationPresenter', notification_presenter)
self.getDynamicApiManager().registerAPI('MakoNotificationMonitor', self.makoNotificationMonitor) self.getDynamicApiManager().registerAPI('MakoNotificationMonitor', self.makoNotificationMonitor)
self.getDynamicApiManager().registerAPI('ScreenLockMonitor', self.screenLockMonitor)
self.getDynamicApiManager().registerAPI('CthulhuState', cthulhu_state) self.getDynamicApiManager().registerAPI('CthulhuState', cthulhu_state)
self.getDynamicApiManager().registerAPI('CthulhuPlatform', cthulhu_platform) self.getDynamicApiManager().registerAPI('CthulhuPlatform', cthulhu_platform)
self.getDynamicApiManager().registerAPI('Settings', settings) self.getDynamicApiManager().registerAPI('Settings', settings)
+1 -1
View File
@@ -23,5 +23,5 @@
# Forked from Orca screen reader. # Forked from Orca screen reader.
# Cthulhu project: https://git.stormux.org/storm/cthulhu # Cthulhu project: https://git.stormux.org/storm/cthulhu
version = "2026.05.25" version = "2026.07.17"
codeName = "master" codeName = "master"
+4
View File
@@ -64,6 +64,10 @@ compositorSnapshot: Optional[Any] = None
# #
pauseAtspiChurn: bool = False pauseAtspiChurn: bool = False
# Whether a trusted screen locker has requested secure-input isolation.
#
screenLockActive: bool = False
# The desktop-context token that should be prioritized next. # The desktop-context token that should be prioritized next.
# #
prioritizedDesktopContextToken: Optional[str] = None prioritizedDesktopContextToken: Optional[str] = None
+68 -1
View File
@@ -41,7 +41,9 @@ import time
from typing import TYPE_CHECKING, Optional, Dict, List, Tuple, Any, Union from typing import TYPE_CHECKING, Optional, Dict, List, Tuple, Any, Union
from . import cthulhu from . import cthulhu
from . import braille
from . import debug from . import debug
from . import focus_manager
from . import input_event from . import input_event
from . import input_event_manager from . import input_event_manager
from . import cthulhu_state from . import cthulhu_state
@@ -132,6 +134,9 @@ class EventManager:
def _sync_focus_on_startup(self) -> bool: def _sync_focus_on_startup(self) -> bool:
"""Initialize active window and focus when startup missed focus events.""" """Initialize active window and focus when startup missed focus events."""
if cthulhu_state.screenLockActive:
return False
focus = cthulhu_state.locusOfFocus focus = cthulhu_state.locusOfFocus
if focus and not AXObject.is_dead(focus): if focus and not AXObject.is_dead(focus):
return False return False
@@ -216,8 +221,49 @@ class EventManager:
"prioritized_context_token_present": self._prioritizedContextToken is not None, "prioritized_context_token_present": self._prioritizedContextToken is not None,
"key_handling_active": self._keyHandlingActive, "key_handling_active": self._keyHandlingActive,
"input_event_manager_present": self._inputEventManager is not None, "input_event_manager_present": self._inputEventManager is not None,
"screen_lock_active": cthulhu_state.screenLockActive,
} }
def set_screen_lock_active(self, active: bool, schedule_focus_resync: bool = True) -> None:
"""Isolates screen-reader state while I38Lock owns secure input."""
if cthulhu_state.screenLockActive == active:
return
cthulhu_state.screenLockActive = active
reason = (
"I38Lock secure screen lock active"
if active
else "I38Lock secure screen lock released"
)
input_event_manager.get_manager().set_secure_input_active(active, reason)
self._discard_queued_events_for_screen_lock()
active_script = self.app.scriptManager.get_active_script()
if active and active_script is not None:
active_script.presentationInterrupt()
if active:
braille.clear()
self.app.scriptManager.set_active_script(None, reason)
manager = focus_manager.get_manager()
if manager is not None:
manager.clear_state(reason)
cthulhu_state.pendingSelfHostedFocus = None
if not active and schedule_focus_resync:
GLib.idle_add(self._sync_focus_on_startup)
def _discard_queued_events_for_screen_lock(self) -> None:
"""Drops queued accessibility events at both lock-state boundaries."""
self._gidleLock.acquire()
try:
self._eventQueue = queue.Queue(0)
self._prioritizedEvent = None
finally:
self._gidleLock.release()
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."""
@@ -604,6 +650,16 @@ class EventManager:
def _ignore(self, event: Atspi.Event) -> bool: def _ignore(self, event: Atspi.Event) -> bool:
"""Returns True if this event should be ignored.""" """Returns True if this event should be ignored."""
if cthulhu_state.screenLockActive:
debug.print_log(
debug.LEVEL_INFO,
"EVENT MANAGER",
"Ignoring - secure screen lock is active",
"screen-lock-active",
True,
)
return True
app = AXObject.get_application(event.source) app = AXObject.get_application(event.source)
debug.printMessage(debug.LEVEL_INFO, '') debug.printMessage(debug.LEVEL_INFO, '')
tokens = ["EVENT MANAGER:", event.type, "from", app] tokens = ["EVENT MANAGER:", event.type, "from", app]
@@ -965,6 +1021,11 @@ class EventManager:
- e: an at-spi event. - e: an at-spi event.
""" """
if cthulhu_state.screenLockActive:
msg = "EVENT MANAGER: Dropping event during secure screen lock."
debug.printMessage(debug.LEVEL_INFO, msg, True)
return
if debug.debugEventQueue: if debug.debugEventQueue:
if self._enqueueCount: if self._enqueueCount:
msg = f"EVENT MANAGER: _enqueue entered before exiting (count={self._enqueueCount})" msg = f"EVENT MANAGER: _enqueue entered before exiting (count={self._enqueueCount})"
@@ -1098,6 +1159,9 @@ class EventManager:
return True return True
def _onNoFocus(self) -> bool: def _onNoFocus(self) -> bool:
if cthulhu_state.screenLockActive:
return False
if not self._isNoFocus(): if not self._isNoFocus():
return False return False
@@ -1300,7 +1364,7 @@ class EventManager:
- event: an instance of BrailleEvent or a KeyboardEvent - event: an instance of BrailleEvent or a KeyboardEvent
""" """
if not cthulhu_state.activeScript: if cthulhu_state.screenLockActive or not cthulhu_state.activeScript:
return return
if not isinstance(event, input_event.BrailleEvent): if not isinstance(event, input_event.BrailleEvent):
@@ -1658,6 +1722,9 @@ class EventManager:
- e: an at-spi event. - e: an at-spi event.
""" """
if cthulhu_state.screenLockActive:
return
debug.printObjectEvent(debug.LEVEL_INFO, event, timestamp=True) debug.printObjectEvent(debug.LEVEL_INFO, event, timestamp=True)
eType = event.type eType = event.type
+18
View File
@@ -73,6 +73,7 @@ class InputEventManager:
self._mapped_keysyms: List[int] = [] self._mapped_keysyms: List[int] = []
self._grabbed_bindings: Dict[int, keybindings.KeyBinding] = {} self._grabbed_bindings: Dict[int, keybindings.KeyBinding] = {}
self._paused: bool = False self._paused: bool = False
self._secure_input_active: bool = False
self._wnck = None self._wnck = None
self._did_attempt_wnck_load: bool = False self._did_attempt_wnck_load: bool = False
self._scriptWithSuspendedGrabsForXterm = None self._scriptWithSuspendedGrabsForXterm = None
@@ -119,6 +120,7 @@ class InputEventManager:
"mapped_keysyms": len(self._mapped_keysyms), "mapped_keysyms": len(self._mapped_keysyms),
"grabbed_bindings": len(self._grabbed_bindings), "grabbed_bindings": len(self._grabbed_bindings),
"paused": self._paused, "paused": self._paused,
"secure_input_active": self._secure_input_active,
"suspended_xterm_script_present": self._scriptWithSuspendedGrabsForXterm is not None, "suspended_xterm_script_present": self._scriptWithSuspendedGrabsForXterm is not None,
"last_input_event_present": self._last_input_event 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, "last_non_modifier_key_event_present": self._last_non_modifier_key_event is not None,
@@ -188,6 +190,18 @@ class InputEventManager:
debug.print_message(debug.LEVEL_INFO, msg, True) debug.print_message(debug.LEVEL_INFO, msg, True)
self._paused = pause self._paused = pause
def set_secure_input_active(self, active: bool, reason: str = "") -> None:
"""Suppresses ordinary keyboard processing while a trusted lock screen is active."""
if self._secure_input_active == active:
return
self._secure_input_active = active
self._last_input_event = None
self._last_non_modifier_key_event = None
msg = f"INPUT EVENT MANAGER: Secure input active: {active}. {reason}"
debug.print_message(debug.LEVEL_INFO, msg, True)
def check_grabbed_bindings(self) -> None: def check_grabbed_bindings(self) -> None:
"""Checks the grabbed key bindings.""" """Checks the grabbed key bindings."""
@@ -753,6 +767,10 @@ class InputEventManager:
) -> bool: ) -> bool:
"""Processes this Atspi keyboard event.""" """Processes this Atspi keyboard event."""
if self._secure_input_active:
msg = "INPUT EVENT MANAGER: Ignoring keyboard event during secure screen lock."
debug.print_message(debug.LEVEL_INFO, msg, True)
return False
if self._paused: if self._paused:
msg = "INPUT EVENT MANAGER: Keyboard event processing is paused." msg = "INPUT EVENT MANAGER: Keyboard event processing is paused."
debug.print_message(debug.LEVEL_INFO, msg, True) debug.print_message(debug.LEVEL_INFO, msg, True)
+1
View File
@@ -87,6 +87,7 @@ cthulhu_python_sources = files([
'script.py', 'script.py',
'script_manager.py', 'script_manager.py',
'script_utilities.py', 'script_utilities.py',
'screen_lock_monitor.py',
'settings.py', 'settings.py',
'settings_manager.py', 'settings_manager.py',
'signal_manager.py', 'signal_manager.py',
+155
View File
@@ -0,0 +1,155 @@
#!/usr/bin/env python3
#
# Copyright (c) 2026 Stormux
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
"""Tracks the lifetime of I38Lock's secure accessibility-feedback worker."""
from __future__ import annotations
import os
from typing import Any, Optional
from gi.repository import Gio, GLib
from . import debug
class ScreenLockMonitor:
"""Enters secure screen-reader mode while I38Lock owns its session-bus name."""
BUS_NAME = "org.stormux.I38Lock1"
TRUSTED_EXECUTABLE = "i38lock"
def __init__(self, app: Any) -> None:
self._app = app
self._connection: Optional[Gio.DBusConnection] = None
self._watch_id: int = 0
self._is_running: bool = False
self._is_active: bool = False
def start(self) -> None:
"""Starts watching the session bus for I38Lock."""
if self._is_running:
return
try:
self._connection = Gio.bus_get_sync(Gio.BusType.SESSION, None)
except Exception as error:
msg = f"SCREEN LOCK MONITOR: Failed to connect to session bus: {error}"
debug.printMessage(debug.LEVEL_WARNING, msg, True)
return
self._watch_id = Gio.bus_watch_name_on_connection(
self._connection,
self.BUS_NAME,
Gio.BusNameWatcherFlags.NONE,
self._on_name_appeared,
self._on_name_vanished,
)
self._is_running = True
debug.printMessage(debug.LEVEL_INFO, "SCREEN LOCK MONITOR: Started", True)
def stop(self) -> None:
"""Stops watching I38Lock and releases any active secure state."""
if self._watch_id:
Gio.bus_unwatch_name(self._watch_id)
self._watch_id = 0
if self._is_active:
self._set_active(False, schedule_focus_resync=False)
self._connection = None
self._is_running = False
debug.printMessage(debug.LEVEL_INFO, "SCREEN LOCK MONITOR: Stopped", True)
def is_running(self) -> bool:
"""Returns whether the D-Bus name watcher is running."""
return self._is_running
def is_active(self) -> bool:
"""Returns whether I38Lock currently owns its secure-state bus name."""
return self._is_active
def _set_active(self, active: bool, schedule_focus_resync: bool = True) -> None:
if self._is_active == active:
return
self._is_active = active
self._app.eventManager.set_screen_lock_active(
active,
schedule_focus_resync=schedule_focus_resync,
)
def _get_name_owner_pid(
self,
connection: Gio.DBusConnection,
name_owner: str,
) -> Optional[int]:
try:
result = connection.call_sync(
"org.freedesktop.DBus",
"/org/freedesktop/DBus",
"org.freedesktop.DBus",
"GetConnectionUnixProcessID",
GLib.Variant("(s)", (name_owner,)),
GLib.VariantType("(u)"),
Gio.DBusCallFlags.NONE,
500,
None,
)
return result.unpack()[0]
except Exception as error:
msg = f"SCREEN LOCK MONITOR: Failed to get owner PID for {name_owner}: {error}"
debug.printMessage(debug.LEVEL_WARNING, msg, True)
return None
def _is_trusted_owner(self, connection: Gio.DBusConnection, name_owner: str) -> bool:
pid = self._get_name_owner_pid(connection, name_owner)
if pid is None:
return False
try:
executable = os.path.basename(os.readlink(f"/proc/{pid}/exe"))
except OSError as error:
msg = f"SCREEN LOCK MONITOR: Failed to inspect owner PID {pid}: {error}"
debug.printMessage(debug.LEVEL_WARNING, msg, True)
return False
if executable != self.TRUSTED_EXECUTABLE:
msg = f"SCREEN LOCK MONITOR: Ignoring untrusted owner executable {executable}"
debug.printMessage(debug.LEVEL_WARNING, msg, True)
return False
return True
def _on_name_appeared(
self,
connection: Gio.DBusConnection,
name: str,
name_owner: str,
_user_data: Optional[Any] = None,
) -> None:
msg = f"SCREEN LOCK MONITOR: {name} appeared as {name_owner}"
debug.printMessage(debug.LEVEL_INFO, msg, True)
if not self._is_trusted_owner(connection, name_owner):
return
self._set_active(True)
def _on_name_vanished(
self,
_connection: Gio.DBusConnection,
name: str,
_user_data: Optional[Any] = None,
) -> None:
msg = f"SCREEN LOCK MONITOR: {name} vanished"
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._set_active(False)
+36
View File
@@ -144,6 +144,37 @@ class Player:
self._stopWorkerLocked("Sound system shutdown") self._stopWorkerLocked("Sound system shutdown")
self._initialized = False self._initialized = False
def get_debug_snapshot(self) -> dict[str, Any]:
"""Returns read-only sound worker state for runtime diagnostics."""
with self._workerLock:
process = self._workerProcess
workerExitStatus: Any = ""
workerAlive = False
workerPid = 0
if process is not None:
workerPid = process.pid
workerExitStatus = process.poll()
workerAlive = workerExitStatus is None
with self._responseLock:
pendingResponseCount = len(self._pendingResponses)
return {
"gstreamer_available": _gstreamerAvailable,
"sound_system_available": isSoundSystemAvailable(),
"failure_reason": getSoundSystemFailureReason() or "",
"initialized": self._initialized,
"worker_present": process is not None,
"worker_alive": workerAlive,
"worker_pid": workerPid,
"worker_exit_status": workerExitStatus if workerExitStatus is not None else "",
"worker_sink": self._workerSink or "",
"worker_restart_required": self._workerRestartRequired,
"worker_restart_reason": self._workerRestartReason or "",
"pending_responses": pendingResponseCount,
}
def _playIcon(self, icon: Icon, interrupt: bool = True) -> None: def _playIcon(self, icon: Icon, interrupt: bool = True) -> None:
if not icon.isValid(): if not icon.isValid():
return return
@@ -555,6 +586,7 @@ def _buildSoundHelperEnvironment() -> dict[str, str]:
environment = os.environ.copy() environment = os.environ.copy()
if pythonPathEntries: if pythonPathEntries:
environment["PYTHONPATH"] = os.pathsep.join(pythonPathEntries) environment["PYTHONPATH"] = os.pathsep.join(pythonPathEntries)
environment.setdefault("ORC_CODE", "backup")
return environment return environment
@@ -571,6 +603,10 @@ def getPlayer() -> Player:
return _player return _player
def get_debug_snapshot() -> dict[str, Any]:
return _player.get_debug_snapshot()
def play(item: Any, interrupt: bool = True) -> None: def play(item: Any, interrupt: bool = True) -> None:
_player.play(item, interrupt) _player.play(item, interrupt)
+3 -1
View File
@@ -540,7 +540,9 @@ def speak(content: Union[str, List[Any]], acss: Optional[Any] = None, interrupt:
if element.isValid(): if element.isValid():
player = sound.getPlayer() player = sound.getPlayer()
if player: if player:
player.play(element, interrupt=interrupt) # Role/state icons are short UI cues. Speech can be queued, but
# sound cues should not pile up behind stale or previous sounds.
player.play(element, interrupt=True)
elif toSpeak: elif toSpeak:
newVoice = ACSS(acss) newVoice = ACSS(acss)
newItemsToSpeak = [] newItemsToSpeak = []
@@ -1,6 +1,7 @@
import sys import sys
import unittest import unittest
import importlib import importlib
import tempfile
from pathlib import Path from pathlib import Path
from unittest import mock from unittest import mock
@@ -27,6 +28,19 @@ class PresentationInterruptSoundRegressionTests(unittest.TestCase):
echoServer.stop.assert_called_once_with() echoServer.stop.assert_called_once_with()
soundPlayer.stop.assert_called_once_with() soundPlayer.stop.assert_called_once_with()
def test_speech_icons_interrupt_previous_sound_when_speech_is_queued(self):
soundPlayer = mock.Mock()
with tempfile.TemporaryDirectory() as temporaryDirectory:
soundPath = Path(temporaryDirectory) / "button.wav"
soundPath.write_bytes(b"RIFF")
icon = speech.Icon(temporaryDirectory, "button.wav")
with mock.patch.object(speech.sound, "getPlayer", return_value=soundPlayer):
speech.speak([icon], interrupt=False)
soundPlayer.play.assert_called_once_with(icon, interrupt=True)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+211
View File
@@ -0,0 +1,211 @@
import sys
import types
import unittest
from unittest import mock
import cthulhu as cthulhu_package
from cthulhu import cthulhu_state
stubCthulhu = types.ModuleType("cthulhu.cthulhu")
stubCthulhu.cthulhuApp = mock.Mock()
stubCthulhu.cthulhuApp.getPluginSystemManager.return_value = None
previousCthulhuModule = sys.modules.get("cthulhu.cthulhu")
sys.modules.setdefault("cthulhu.cthulhu", stubCthulhu)
from cthulhu import event_manager
from cthulhu import input_event_manager
from cthulhu.screen_lock_monitor import ScreenLockMonitor
if previousCthulhuModule is None:
sys.modules.pop("cthulhu.cthulhu", None)
if getattr(cthulhu_package, "cthulhu", None) is stubCthulhu:
delattr(cthulhu_package, "cthulhu")
else:
sys.modules["cthulhu.cthulhu"] = previousCthulhuModule
class ScreenLockMonitorTests(unittest.TestCase):
def setUp(self):
self.app = mock.Mock()
self.monitor = ScreenLockMonitor(self.app)
def test_name_appearance_enters_secure_lock_mode(self):
with mock.patch.object(self.monitor, "_is_trusted_owner", return_value=True):
self.monitor._on_name_appeared(mock.Mock(), self.monitor.BUS_NAME, ":1.42")
self.app.eventManager.set_screen_lock_active.assert_called_once_with(
True,
schedule_focus_resync=True,
)
self.assertTrue(self.monitor.is_active())
def test_untrusted_name_owner_does_not_enter_secure_lock_mode(self):
with mock.patch.object(self.monitor, "_is_trusted_owner", return_value=False):
self.monitor._on_name_appeared(mock.Mock(), self.monitor.BUS_NAME, ":1.42")
self.app.eventManager.set_screen_lock_active.assert_not_called()
self.assertFalse(self.monitor.is_active())
def test_name_disappearance_leaves_secure_lock_mode(self):
with mock.patch.object(self.monitor, "_is_trusted_owner", return_value=True):
self.monitor._on_name_appeared(mock.Mock(), self.monitor.BUS_NAME, ":1.42")
self.monitor._on_name_vanished(mock.Mock(), self.monitor.BUS_NAME)
self.assertEqual(
self.app.eventManager.set_screen_lock_active.call_args_list,
[
mock.call(True, schedule_focus_resync=True),
mock.call(False, schedule_focus_resync=True),
],
)
self.assertFalse(self.monitor.is_active())
def test_start_watches_i38lock_bus_name(self):
connection = mock.Mock()
with (
mock.patch("cthulhu.screen_lock_monitor.Gio.bus_get_sync", return_value=connection),
mock.patch(
"cthulhu.screen_lock_monitor.Gio.bus_watch_name_on_connection",
return_value=23,
) as watch_name,
):
self.monitor.start()
watch_name.assert_called_once_with(
connection,
self.monitor.BUS_NAME,
mock.ANY,
self.monitor._on_name_appeared,
self.monitor._on_name_vanished,
)
self.assertTrue(self.monitor.is_running())
def test_stop_releases_secure_lock_without_focus_resync(self):
with mock.patch.object(self.monitor, "_is_trusted_owner", return_value=True):
self.monitor._on_name_appeared(mock.Mock(), self.monitor.BUS_NAME, ":1.42")
with mock.patch("cthulhu.screen_lock_monitor.Gio.bus_unwatch_name"):
self.monitor.stop()
self.assertEqual(
self.app.eventManager.set_screen_lock_active.call_args_list,
[
mock.call(True, schedule_focus_resync=True),
mock.call(False, schedule_focus_resync=False),
],
)
self.assertFalse(self.monitor.is_active())
def test_i38lock_process_owner_is_trusted(self):
connection = mock.Mock()
result = mock.Mock()
result.unpack.return_value = (1234,)
connection.call_sync.return_value = result
with mock.patch("cthulhu.screen_lock_monitor.os.readlink", return_value="/usr/bin/i38lock"):
self.assertTrue(self.monitor._is_trusted_owner(connection, ":1.42"))
def test_non_i38lock_process_owner_is_not_trusted(self):
connection = mock.Mock()
result = mock.Mock()
result.unpack.return_value = (1234,)
connection.call_sync.return_value = result
with mock.patch("cthulhu.screen_lock_monitor.os.readlink", return_value="/usr/bin/python"):
self.assertFalse(self.monitor._is_trusted_owner(connection, ":1.42"))
class SecureScreenLockStateTests(unittest.TestCase):
def setUp(self):
self.original_screen_lock_active = cthulhu_state.screenLockActive
cthulhu_state.screenLockActive = False
def tearDown(self):
cthulhu_state.screenLockActive = self.original_screen_lock_active
def test_entering_secure_lock_clears_context_and_pauses_input(self):
app = mock.Mock()
active_script = mock.Mock()
app.scriptManager.get_active_script.return_value = active_script
manager = event_manager.EventManager(app, asyncMode=True)
keyboard_manager = mock.Mock()
focus_manager = mock.Mock()
with (
mock.patch.object(input_event_manager, "get_manager", return_value=keyboard_manager),
mock.patch("cthulhu.event_manager.focus_manager.get_manager", return_value=focus_manager),
mock.patch("cthulhu.event_manager.braille.clear") as clear_braille,
):
manager.set_screen_lock_active(True)
self.assertTrue(cthulhu_state.screenLockActive)
active_script.presentationInterrupt.assert_called_once_with()
app.scriptManager.set_active_script.assert_called_once_with(
None,
"I38Lock secure screen lock active",
)
focus_manager.clear_state.assert_called_once_with(
"I38Lock secure screen lock active"
)
keyboard_manager.set_secure_input_active.assert_called_once_with(
True,
"I38Lock secure screen lock active",
)
clear_braille.assert_called_once_with()
def test_object_events_are_dropped_without_inspecting_locked_content(self):
app = mock.Mock()
manager = event_manager.EventManager(app, asyncMode=True)
cthulhu_state.screenLockActive = True
event = mock.Mock()
with (
mock.patch.object(manager, "_addToQueue") as add_to_queue,
mock.patch("cthulhu.event_manager.AXObject.get_application") as get_application,
):
manager._enqueue(event)
add_to_queue.assert_not_called()
get_application.assert_not_called()
def test_unlock_resumes_input_and_schedules_focus_resync(self):
app = mock.Mock()
manager = event_manager.EventManager(app, asyncMode=True)
keyboard_manager = mock.Mock()
focus_manager = mock.Mock()
cthulhu_state.screenLockActive = True
with (
mock.patch.object(input_event_manager, "get_manager", return_value=keyboard_manager),
mock.patch("cthulhu.event_manager.focus_manager.get_manager", return_value=focus_manager),
mock.patch("cthulhu.event_manager.GLib.idle_add") as idle_add,
):
manager.set_screen_lock_active(False)
self.assertFalse(cthulhu_state.screenLockActive)
keyboard_manager.set_secure_input_active.assert_called_once_with(
False,
"I38Lock secure screen lock released",
)
idle_add.assert_called_once_with(manager._sync_focus_on_startup)
def test_secure_input_blocks_key_echo_before_event_creation(self):
manager = input_event_manager.InputEventManager()
manager.set_secure_input_active(True, "test lock")
with mock.patch("cthulhu.input_event_manager.input_event.KeyboardEvent") as event_class:
consumed = manager.process_keyboard_event(
mock.Mock(),
True,
25,
112,
0,
"p",
)
self.assertFalse(consumed)
event_class.assert_not_called()
if __name__ == "__main__":
unittest.main()
+13
View File
@@ -2,6 +2,7 @@ import sys
import types import types
import unittest import unittest
import importlib import importlib
import os
from pathlib import Path from pathlib import Path
from unittest import mock from unittest import mock
@@ -68,6 +69,18 @@ class SoundSinkTests(unittest.TestCase):
for module in reloadedModules: for module in reloadedModules:
importlib.reload(module) importlib.reload(module)
def test_sound_helper_environment_defaults_to_orc_backup(self):
with mock.patch.dict(os.environ, {}, clear=True):
environment = sound._buildSoundHelperEnvironment()
self.assertEqual("backup", environment["ORC_CODE"])
def test_sound_helper_environment_preserves_explicit_orc_code(self):
with mock.patch.dict(os.environ, {"ORC_CODE": "debug"}, clear=True):
environment = sound._buildSoundHelperEnvironment()
self.assertEqual("debug", environment["ORC_CODE"])
class PlayerRecoveryTests(unittest.TestCase): class PlayerRecoveryTests(unittest.TestCase):
def test_worker_diagnostic_marks_restart_required(self): def test_worker_diagnostic_marks_restart_required(self):