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
+13 -1
View File
@@ -257,6 +257,7 @@ from . import keybindings
from . import learn_mode_presenter
from . import logger
from . import mako_notification_monitor
from . import screen_lock_monitor
from . import messages
from . import notification_presenter
from . import focus_manager
@@ -724,6 +725,7 @@ def start() -> None:
# Start D-Bus remote controller service after ATSPI is ready
with debug.startup_timing("startup idle services"):
cthulhuApp.getMakoNotificationMonitor().start()
cthulhuApp.getScreenLockMonitor().start()
GObject.idle_add(_start_dbus_service)
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.alarm(settings.timeoutTime)
cthulhu_state.activeScript.presentationInterrupt()
if cthulhu_state.activeScript is not None:
cthulhu_state.activeScript.presentationInterrupt()
cthulhuApp.getSignalManager().emitSignal('stop-application-completed')
sound_theme_manager.getManager().playStopSound(wait=True, timeoutSeconds=1)
cthulhuApp.getPluginSystemManager().unloadAllPlugins(ForceAllPlugins=True)
cthulhuApp.getScreenLockMonitor().stop()
cthulhuApp.getMakoNotificationMonitor().stop()
# 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"):
self.makoNotificationMonitor: mako_notification_monitor.MakoNotificationMonitor = \
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"):
self.createCompatAPI()
with debug.startup_timing("plugin manager construction"):
@@ -1009,6 +1016,9 @@ class Cthulhu(GObject.Object):
def getMakoNotificationMonitor(self) -> mako_notification_monitor.MakoNotificationMonitor:
return self.makoNotificationMonitor
def getScreenLockMonitor(self) -> screen_lock_monitor.ScreenLockMonitor:
return self.screenLockMonitor
def getDynamicApiManager(self) -> DynamicApiManager:
return self.dynamicApiManager
@@ -1061,6 +1071,7 @@ class Cthulhu(GObject.Object):
"speech_server_present": speechServer is not None,
"speech_server": speechServer.__class__.__name__ if speechServer else "",
},
"sound": sound.get_debug_snapshot(),
"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(),
@@ -1097,6 +1108,7 @@ class Cthulhu(GObject.Object):
self.getDynamicApiManager().registerAPI('Cmdnames', cmdnames)
self.getDynamicApiManager().registerAPI('NotificationPresenter', notification_presenter)
self.getDynamicApiManager().registerAPI('MakoNotificationMonitor', self.makoNotificationMonitor)
self.getDynamicApiManager().registerAPI('ScreenLockMonitor', self.screenLockMonitor)
self.getDynamicApiManager().registerAPI('CthulhuState', cthulhu_state)
self.getDynamicApiManager().registerAPI('CthulhuPlatform', cthulhu_platform)
self.getDynamicApiManager().registerAPI('Settings', settings)
+1 -1
View File
@@ -23,5 +23,5 @@
# Forked from Orca screen reader.
# Cthulhu project: https://git.stormux.org/storm/cthulhu
version = "2026.05.25"
version = "2026.07.17"
codeName = "master"
+4
View File
@@ -64,6 +64,10 @@ compositorSnapshot: Optional[Any] = None
#
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.
#
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 . import cthulhu
from . import braille
from . import debug
from . import focus_manager
from . import input_event
from . import input_event_manager
from . import cthulhu_state
@@ -132,6 +134,9 @@ class EventManager:
def _sync_focus_on_startup(self) -> bool:
"""Initialize active window and focus when startup missed focus events."""
if cthulhu_state.screenLockActive:
return False
focus = cthulhu_state.locusOfFocus
if focus and not AXObject.is_dead(focus):
return False
@@ -216,8 +221,49 @@ class EventManager:
"prioritized_context_token_present": self._prioritizedContextToken is not None,
"key_handling_active": self._keyHandlingActive,
"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:
"""Stores the compositor state adapter used for startup wiring."""
@@ -604,6 +650,16 @@ class EventManager:
def _ignore(self, event: Atspi.Event) -> bool:
"""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)
debug.printMessage(debug.LEVEL_INFO, '')
tokens = ["EVENT MANAGER:", event.type, "from", app]
@@ -965,6 +1021,11 @@ class EventManager:
- 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 self._enqueueCount:
msg = f"EVENT MANAGER: _enqueue entered before exiting (count={self._enqueueCount})"
@@ -1098,6 +1159,9 @@ class EventManager:
return True
def _onNoFocus(self) -> bool:
if cthulhu_state.screenLockActive:
return False
if not self._isNoFocus():
return False
@@ -1300,7 +1364,7 @@ class EventManager:
- event: an instance of BrailleEvent or a KeyboardEvent
"""
if not cthulhu_state.activeScript:
if cthulhu_state.screenLockActive or not cthulhu_state.activeScript:
return
if not isinstance(event, input_event.BrailleEvent):
@@ -1658,6 +1722,9 @@ class EventManager:
- e: an at-spi event.
"""
if cthulhu_state.screenLockActive:
return
debug.printObjectEvent(debug.LEVEL_INFO, event, timestamp=True)
eType = event.type
+18
View File
@@ -73,6 +73,7 @@ class InputEventManager:
self._mapped_keysyms: List[int] = []
self._grabbed_bindings: Dict[int, keybindings.KeyBinding] = {}
self._paused: bool = False
self._secure_input_active: bool = False
self._wnck = None
self._did_attempt_wnck_load: bool = False
self._scriptWithSuspendedGrabsForXterm = None
@@ -119,6 +120,7 @@ class InputEventManager:
"mapped_keysyms": len(self._mapped_keysyms),
"grabbed_bindings": len(self._grabbed_bindings),
"paused": self._paused,
"secure_input_active": self._secure_input_active,
"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,
@@ -188,6 +190,18 @@ class InputEventManager:
debug.print_message(debug.LEVEL_INFO, msg, True)
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:
"""Checks the grabbed key bindings."""
@@ -753,6 +767,10 @@ class InputEventManager:
) -> bool:
"""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:
msg = "INPUT EVENT MANAGER: Keyboard event processing is paused."
debug.print_message(debug.LEVEL_INFO, msg, True)
+1
View File
@@ -87,6 +87,7 @@ cthulhu_python_sources = files([
'script.py',
'script_manager.py',
'script_utilities.py',
'screen_lock_monitor.py',
'settings.py',
'settings_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._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:
if not icon.isValid():
return
@@ -555,6 +586,7 @@ def _buildSoundHelperEnvironment() -> dict[str, str]:
environment = os.environ.copy()
if pythonPathEntries:
environment["PYTHONPATH"] = os.pathsep.join(pythonPathEntries)
environment.setdefault("ORC_CODE", "backup")
return environment
@@ -571,6 +603,10 @@ def getPlayer() -> Player:
return _player
def get_debug_snapshot() -> dict[str, Any]:
return _player.get_debug_snapshot()
def play(item: Any, interrupt: bool = True) -> None:
_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():
player = sound.getPlayer()
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:
newVoice = ACSS(acss)
newItemsToSpeak = []
@@ -1,6 +1,7 @@
import sys
import unittest
import importlib
import tempfile
from pathlib import Path
from unittest import mock
@@ -27,6 +28,19 @@ class PresentationInterruptSoundRegressionTests(unittest.TestCase):
echoServer.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__":
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 unittest
import importlib
import os
from pathlib import Path
from unittest import mock
@@ -68,6 +69,18 @@ class SoundSinkTests(unittest.TestCase):
for module in reloadedModules:
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):
def test_worker_diagnostic_marks_restart_required(self):