import sys import types import unittest from pathlib import Path from unittest import mock sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from cthulhu import cthulhu_state from cthulhu import compositor_state_types stubCthulhu = types.ModuleType("cthulhu.cthulhu") stubCthulhu.cthulhuApp = mock.Mock() sys.modules.setdefault("cthulhu.cthulhu", stubCthulhu) from cthulhu import event_manager class FakeEvent: def __init__(self, event_type, source="same", detail1=0, detail2=0, any_data=None): self.type = event_type self.source = source self.detail1 = detail1 self.detail2 = detail2 self.any_data = any_data 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() fakeAtspi = types.SimpleNamespace( Accessible=event_manager.Atspi.Accessible, EventListener=types.SimpleNamespace(new=mock.Mock(return_value=self.listener)), Role=event_manager.Atspi.Role, ) self.listenerPatch = mock.patch.object(event_manager, "Atspi", fakeAtspi) self.listenerPatch.start() self.addCleanup(self.listenerPatch.stop) self.manager = event_manager.EventManager(mock.Mock(), asyncMode=False) self.manager._active = True self.manager._context_token_for_event = mock.Mock(side_effect=lambda event: event.source) 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() self.manager.set_compositor_state_adapter(adapter) adapter.add_listener.assert_called_once_with(self.manager._handle_compositor_signal) def test_set_compositor_state_adapter_updates_active_input_manager(self) -> None: adapter = mock.Mock() self.manager._inputEventManager = mock.Mock() self.manager.set_compositor_state_adapter(adapter) self.manager._inputEventManager.set_compositor_state_adapter.assert_called_once_with( adapter ) def test_pause_signal_updates_churn_state_and_resume_clears_it(self) -> None: snapshot = compositor_state_types.DesktopContextSnapshot(session_type="wayland") self.manager._handle_compositor_signal( compositor_state_types.CompositorStateEvent( compositor_state_types.PAUSE_ATSPI_CHURN, reason="workspace-transition", snapshot=snapshot, payload={"context_token": "current"}, ) ) self.assertTrue(cthulhu_state.pauseAtspiChurn) self.assertTrue(self.manager._churnSuppressed) self.manager._handle_compositor_signal( compositor_state_types.CompositorStateEvent( compositor_state_types.RESUME_ATSPI_CHURN, reason="workspace-transition", snapshot=snapshot, payload={"context_token": "current"}, ) ) self.assertFalse(cthulhu_state.pauseAtspiChurn) self.assertFalse(self.manager._churnSuppressed) def test_confirmed_empty_i3_workspace_clears_stale_web_context(self) -> None: staleScript = mock.Mock() defaultScript = mock.Mock() focusManager = mock.Mock() self.manager.app.scriptManager.get_active_script.return_value = staleScript self.manager.app.scriptManager.get_default_script.return_value = defaultScript snapshot = compositor_state_types.DesktopContextSnapshot( session_type="x11", backend_name="i3-ipc", active_workspace_ids=frozenset({"9"}), focused_workspace_empty=True, ) with mock.patch.object( event_manager.focus_manager, "get_manager", return_value=focusManager, ): self.manager._handle_compositor_signal( compositor_state_types.CompositorStateEvent( compositor_state_types.DESKTOP_TRANSITION_FINISHED, reason="workspace-focus", snapshot=snapshot, ) ) staleScript.presentationInterrupt.assert_called_once_with() focusManager.clear_state.assert_called_once_with("confirmed empty i3 workspace") self.manager.app.scriptManager.set_active_script.assert_called_once_with( defaultScript, "focus: confirmed-empty-i3-workspace", ) self.assertTrue(self.manager._desktopContextConfirmedEmpty) def test_atspi_event_is_ignored_while_i3_workspace_is_confirmed_empty(self) -> None: self.manager._desktopContextConfirmedEmpty = True event = FakeEvent("object:state-changed:focused", source=object(), detail1=1) self.manager._get_scriptForEvent = mock.Mock() with ( mock.patch.object(event_manager.debug, "printObjectEvent"), mock.patch.object(event_manager.debug, "printMessage"), ): self.manager._processObjectEvent(event) self.manager._get_scriptForEvent.assert_not_called() def test_nonempty_i3_workspace_schedules_focus_resync_after_empty_workspace(self) -> None: self.manager._desktopContextConfirmedEmpty = True snapshot = compositor_state_types.DesktopContextSnapshot( session_type="x11", backend_name="i3-ipc", active_workspace_ids=frozenset({"8"}), focused_window_id=0x200000C, focused_workspace_empty=False, ) with mock.patch.object(event_manager.GLib, "idle_add") as idleAdd: self.manager._handle_compositor_signal( compositor_state_types.CompositorStateEvent( compositor_state_types.WORKSPACE_STATE_CHANGED, reason="window-focus", snapshot=snapshot, ) ) self.assertFalse(self.manager._desktopContextConfirmedEmpty) idleAdd.assert_called_once_with(self.manager._sync_focus_on_startup) def test_stale_context_event_is_obsolete_while_churn_is_paused(self) -> None: self.manager._churnSuppressed = True self.manager._prioritizedContextToken = "current" event = FakeEvent("object:children-changed:add", source="stale") self.assertTrue(self.manager._is_obsolete_by_context(event)) def test_flush_signal_removes_stale_events_from_queue(self) -> None: self.manager._churnSuppressed = False self.manager._prioritizedContextToken = "current" staleEvent = FakeEvent("object:children-changed:add", source="stale") currentEvent = FakeEvent("object:children-changed:add", source="current") self.manager._eventQueue.put(staleEvent) self.manager._eventQueue.put(currentEvent) self.manager._handle_compositor_signal( compositor_state_types.CompositorStateEvent( compositor_state_types.FLUSH_STALE_ATSPI_EVENTS, reason="resume", snapshot=compositor_state_types.DesktopContextSnapshot(session_type="wayland"), payload={"context_token": "current"}, ) ) self.assertEqual(list(self.manager._eventQueue.queue), [currentEvent]) def test_flush_classifies_outside_queue_lock_and_preserves_nested_event(self) -> None: staleEvent = FakeEvent("object:children-changed:add", source="stale") currentEvent = FakeEvent("object:children-changed:add", source="current") nestedEvent = FakeEvent("object:state-changed:focused", source="nested", detail1=1) self.manager._eventQueue.put(staleEvent) self.manager._eventQueue.put(currentEvent) self.manager._ignore = mock.Mock(return_value=False) self.manager._prioritizeSelfHostedFocusedEvent = mock.Mock(return_value=False) self.manager._queuePrintln = mock.Mock() self.manager._inFlood = mock.Mock(return_value=False) self.manager._shouldSuspendEventsFor = mock.Mock(return_value=False) self.manager._dequeue = mock.Mock(return_value=False) script = types.SimpleNamespace(eventCache={}) def classify_event(event) -> bool: self.assertFalse(self.manager._gidleLock.locked()) if event is staleEvent: self.manager._enqueue(nestedEvent) return True return False self.manager._event_is_from_stale_context = mock.Mock(side_effect=classify_event) self.manager._should_preserve_during_suppression = mock.Mock(return_value=False) with ( mock.patch.object(event_manager.AXObject, "get_application", return_value=object()), mock.patch.object(event_manager.GLib, "idle_add", return_value=1), mock.patch.object( event_manager.cthulhu.cthulhuApp.scriptManager, "get_script", return_value=script, ), mock.patch.object( event_manager.AXUtilities, "get_application_toolkit_name", return_value="VCL", ), ): self.manager._flush_stale_atspi_events() self.assertEqual(list(self.manager._eventQueue.queue), [currentEvent, nestedEvent]) self.manager._dequeue.assert_not_called() def test_flush_restores_retained_and_unexamined_events_after_exception(self) -> None: firstEvent = FakeEvent("object:children-changed:add", source="first") secondEvent = FakeEvent("object:children-changed:add", source="second") self.manager._eventQueue.put(firstEvent) self.manager._eventQueue.put(secondEvent) self.manager._event_is_from_stale_context = mock.Mock( side_effect=RuntimeError("property lookup failed"), ) with self.assertRaisesRegex(RuntimeError, "property lookup failed"): self.manager._flush_stale_atspi_events() self.assertEqual(list(self.manager._eventQueue.queue), [firstEvent, secondEvent]) self.assertFalse(self.manager._flushingStaleAtspiEvents) def test_stale_background_event_does_not_activate_script_during_suppression(self) -> None: script = mock.Mock() script.isActivatableEvent.return_value = True script.forceScriptActivation.return_value = False self.manager._churnSuppressed = True self.manager._prioritizedContextToken = "current" event = FakeEvent("object:state-changed:showing", source="old") result, reason = self.manager._isActivatableEvent(event, script) self.assertFalse(result) self.assertIn("compositor-prioritized context", reason) def test_focus_event_syncs_accessible_context_back_into_adapter(self) -> None: adapter = mock.Mock() script = mock.Mock() source = object() event = FakeEvent("object:state-changed:focused", source=source, detail1=1) self.manager._compositorStateAdapter = adapter self.manager._get_scriptForEvent = mock.Mock(return_value=script) self.manager._isActivatableEvent = mock.Mock(return_value=(False, "already active")) self.manager._inFlood = mock.Mock(return_value=False) with ( mock.patch.object(event_manager.debug, "printObjectEvent"), mock.patch.object(event_manager.debug, "printDetails"), mock.patch.object(event_manager.debug, "printMessage"), mock.patch.object(event_manager.debug, "printTokens"), mock.patch.object(event_manager.AXUtilities, "get_desktop", return_value=object()), mock.patch.object(event_manager.AXUtilities, "is_defunct", return_value=False), mock.patch.object(event_manager.AXUtilities, "is_iconified", return_value=False), mock.patch.object(event_manager.AXUtilities, "is_frame", return_value=False), mock.patch.object(event_manager.AXObject, "is_dead", return_value=False), ): self.manager._processObjectEvent(event) 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_quick_access_name_changed_survives_without_active_script(self) -> None: app = object() event = FakeEvent("object:property-change:accessible-name", source="quick-access") cthulhu_state.activeScript = None 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"), ): self.manager._isSteamOverlayNavigationEvent = mock.Mock(return_value=True) self.assertFalse(self.manager._ignore(event)) def test_steam_overlay_navigation_requires_quick_access_context(self) -> None: app = object() event = FakeEvent("object:property-change:accessible-name", source="not-quick-access") self.manager._isSteamApp = mock.Mock(return_value=True) self.manager._activeX11WindowIsSteamSelectionContext = mock.Mock(return_value=True) self.manager._eventHasSteamQuickAccessContext = mock.Mock(return_value=False) self.assertFalse(self.manager._isSteamOverlayNavigationEvent(event, app)) def test_steam_overlay_navigation_rejects_children_changed_churn(self) -> None: app = object() event = FakeEvent("object:children-changed:add", source="quick-access", any_data=object()) self.manager._isSteamApp = mock.Mock(return_value=True) self.manager._activeX11WindowIsSteamSelectionContext = mock.Mock(return_value=True) self.manager._eventHasSteamQuickAccessContext = mock.Mock(return_value=True) self.assertFalse(self.manager._isSteamOverlayNavigationEvent(event, app)) 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()