storm/cthulhu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden Wayland compositor runtime checks

Browse Source
This commit is contained in:
Hunter Jozwiak
2026-04-09 12:01:09 -04:00
parent 8fc77c5a2f
commit 5bb5f3d711
3 changed files with 93 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/cthulhu/compositor_state_wayland.py
View File

@@ -17,6 +17,7 @@ from typing import Any
from gi.repository import GLib
from . import debug
from .compositor_state_types import (
DESKTOP_TRANSITION_FINISHED,
DESKTOP_TRANSITION_STARTED,
@@ -98,7 +99,9 @@ class WaylandSharedProtocolsBackend:
self._bind_listener(self._registry, "global_remove", self._handle_registry_global_remove)
self._roundtrip()
self._install_dispatch_watch()
except Exception:
except Exception as error:
msg = f"COMPOSITOR STATE: Wayland backend activation failed: {error}"
debug.printMessage(debug.LEVEL_WARNING, msg, True)
self.deactivate()
def deactivate(self, emit_signal: Any = None) -> None:

28
tests/test_compositor_state_adapter_regressions.py
View File

@@ -110,6 +110,31 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
sourceRemove.assert_called_once_with(41)
def test_wayland_backend_logs_activation_failure_reason(self) -> None:
fakeDisplay = mock.Mock()
fakeDisplay.connect.side_effect = ValueError("Unable to connect to display")
fakeProtocols = mock.Mock()
fakeProtocols.has_runtime_support.return_value = True
fakeProtocols.get_display_class.return_value = mock.Mock(return_value=fakeDisplay)
backend = compositor_state_wayland.WaylandSharedProtocolsBackend(
environment={"WAYLAND_DISPLAY": "wayland-0"},
protocols=fakeProtocols,
)
with (
mock.patch.object(compositor_state_wayland, "get_session_type", return_value="wayland"),
mock.patch.object(compositor_state_wayland.debug, "printMessage") as printMessage,
):
backend.activate(mock.Mock())
printMessage.assert_any_call(
compositor_state_wayland.debug.LEVEL_WARNING,
mock.ANY,
True,
)
self.assertIn("Unable to connect to display", printMessage.call_args_list[-1].args[1])
self.assertIsNone(backend._display)
def test_local_ext_workspace_wrapper_supports_base_pywayland_without_distro_protocol_module(self) -> None:
fakeClientModule = types.ModuleType("pywayland.client")
@@ -234,7 +259,8 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
backend_name="wayland-shared-protocols",
)
workspace.activate(adapter._handle_workspace_signal)
with mock.patch.object(workspace, "is_available", return_value=False):
workspace.activate(adapter._handle_workspace_signal)
workspace._handle_workspace_state(first_workspace, "active", True)
workspace._handle_workspace_manager_done()

62
tests/test_event_manager_compositor_context_regressions.py
View File

@@ -27,6 +27,12 @@ class FakeEvent:
class EventManagerCompositorContextRegressionTests(unittest.TestCase):
def setUp(self) -> None:
self.originalPauseAtspiChurn = cthulhu_state.pauseAtspiChurn
self.originalPrioritizedDesktopContextToken = cthulhu_state.prioritizedDesktopContextToken
self.originalCompositorSnapshot = cthulhu_state.compositorSnapshot
self.originalActiveScript = cthulhu_state.activeScript
self.addCleanup(self._restore_cthulhu_state)
self.listener = mock.Mock()
self.listenerPatch = mock.patch.object(
event_manager.Atspi.EventListener,
@@ -42,6 +48,13 @@ class EventManagerCompositorContextRegressionTests(unittest.TestCase):
cthulhu_state.pauseAtspiChurn = False
cthulhu_state.prioritizedDesktopContextToken = None
cthulhu_state.compositorSnapshot = None
cthulhu_state.activeScript = None
def _restore_cthulhu_state(self) -> None:
cthulhu_state.pauseAtspiChurn = self.originalPauseAtspiChurn
cthulhu_state.prioritizedDesktopContextToken = self.originalPrioritizedDesktopContextToken
cthulhu_state.compositorSnapshot = self.originalCompositorSnapshot
cthulhu_state.activeScript = self.originalActiveScript
def test_set_compositor_state_adapter_registers_compositor_listener(self) -> None:
adapter = mock.Mock()
@@ -142,6 +155,55 @@ class EventManagerCompositorContextRegressionTests(unittest.TestCase):
adapter.sync_accessible_context.assert_called_once_with("object:state-changed:focused")
def test_steam_children_changed_burst_is_suppressed_before_flood_threshold(self) -> None:
app = object()
cthulhu_state.activeScript = mock.Mock(app=app)
firstEvent = FakeEvent("object:children-changed:add", source="steam-context", any_data=object())
secondEvent = FakeEvent("object:children-changed:add", source="steam-context", any_data=object())
with (
mock.patch.object(event_manager.time, "monotonic", side_effect=[100.0, 100.05]),
mock.patch.object(event_manager.debug, "printMessage"),
mock.patch.object(event_manager.debug, "printTokens"),
mock.patch.object(event_manager.debug, "print_log"),
mock.patch.object(event_manager.AXObject, "get_application", return_value=app),
mock.patch.object(event_manager.AXObject, "get_name", return_value="steamwebhelper"),
mock.patch.object(event_manager.AXObject, "get_role", return_value=mock.Mock()),
mock.patch.object(event_manager.AXObject, "is_dead", return_value=False),
mock.patch.object(event_manager.AXUtilities, "has_no_state", return_value=False),
mock.patch.object(event_manager.AXUtilities, "is_defunct", return_value=False),
mock.patch.object(event_manager.AXUtilities, "manages_descendants", return_value=False),
mock.patch.object(event_manager.AXUtilities, "is_image", return_value=False),
mock.patch.object(event_manager.AXUtilities, "is_menu_item", return_value=False),
):
self.manager._isSteamApp = mock.Mock(return_value=True)
self.manager._isSteamNotificationEvent = mock.Mock(return_value=False)
self.assertFalse(self.manager._ignore(firstEvent))
self.assertTrue(self.manager._ignore(secondEvent))
def test_steam_focus_lost_burst_is_ignored_but_focus_gain_is_preserved(self) -> None:
app = object()
cthulhu_state.activeScript = mock.Mock(app=app)
focusLost = FakeEvent("object:state-changed:focused", source="steam-context", detail1=0)
focusGained = FakeEvent("object:state-changed:focused", source="steam-context", detail1=1)
with (
mock.patch.object(event_manager.debug, "printMessage"),
mock.patch.object(event_manager.debug, "printTokens"),
mock.patch.object(event_manager.debug, "print_log"),
mock.patch.object(event_manager.AXObject, "get_application", return_value=app),
mock.patch.object(event_manager.AXObject, "get_name", return_value="steamwebhelper"),
mock.patch.object(event_manager.AXObject, "get_role", return_value=mock.Mock()),
mock.patch.object(event_manager.AXUtilities, "has_no_state", return_value=False),
mock.patch.object(event_manager.AXUtilities, "is_defunct", return_value=False),
):
self.manager._isSteamApp = mock.Mock(return_value=True)
self.manager._isSteamNotificationEvent = mock.Mock(return_value=False)
self.assertTrue(self.manager._ignore(focusLost))
self.assertFalse(self.manager._ignore(focusGained))
if __name__ == "__main__":
unittest.main()