feat: suppress stale AT-SPI churn from compositor state

2026-04-09 09:27:45 -04:00
parent 80d53ebcc3
commit 309822a0ca
2 changed files with 256 additions and 0 deletions
@@ -0,0 +1,108 @@
+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.listener = mock.Mock()
+        self.listenerPatch = mock.patch.object(
+            event_manager.Atspi.EventListener,
+            "new",
+            return_value=self.listener,
+        )
+        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
+
+    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_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_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])
+
+
+if __name__ == "__main__":
+    unittest.main()