Tighter integration with i38lock. Found and squished a bug with sound sometimes not loading.

2026-07-17 13:38:01 -04:00
parent 0858ec441a
commit 40aeadb68a
11 changed files with 536 additions and 3 deletions
@@ -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()
@@ -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()
@@ -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):