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

fix: normalize workspace state signals

Browse Source
This commit is contained in:
Hunter Jozwiak
2026-04-09 09:13:44 -04:00
parent d0aef1331d
commit 3671b0d6b9
4 changed files with 149 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_compositor_state_adapter_regressions.py
+34 -29
View File
@@ -25,11 +25,11 @@ class FakeWorkspaceBackend:
def is_available(self, session_type: str | None = None) -> bool:
return self.available
def activate(self, adapter=None) -> None:
self.activate_calls.append(adapter)
def activate(self, emit_signal=None) -> None:
self.activate_calls.append(emit_signal)
def deactivate(self, adapter=None) -> None:
self.deactivate_calls.append(adapter)
def deactivate(self, emit_signal=None) -> None:
self.deactivate_calls.append(emit_signal)
class CompositorStateAdapterRegressionTests(unittest.TestCase):
@@ -50,7 +50,8 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
adapter.activate()
self.assertEqual(unavailableBackend.activate_calls, [])
self.assertEqual(selectedBackend.activate_calls, [adapter])
self.assertEqual(len(selectedBackend.activate_calls), 1)
self.assertTrue(callable(selectedBackend.activate_calls[0]))
self.assertEqual(adapter.get_snapshot().backend_name, "selected")
def test_default_workspace_backends_include_wayland_then_null_backend(self) -> None:
@@ -116,8 +117,9 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
adapter.activate()
selected_backend.activate.assert_called_once_with(adapter)
selected_backend.activate.assert_called_once()
fallback_backend.activate.assert_not_called()
self.assertTrue(callable(selected_backend.activate.call_args.args[0]))
self.assertEqual(adapter.get_snapshot().backend_name, "wayland-shared-protocols")
def test_activate_is_idempotent_and_deactivates_previous_backend(self) -> None:
@@ -127,8 +129,9 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
adapter.activate()
adapter.activate()
self.assertEqual(backend.activate_calls, [adapter, adapter])
self.assertEqual(backend.deactivate_calls, [adapter])
self.assertEqual(len(backend.activate_calls), 2)
self.assertTrue(all(callable(call) for call in backend.activate_calls))
self.assertEqual(len(backend.deactivate_calls), 1)
self.assertEqual(adapter.get_snapshot().backend_name, "selected")
def test_sync_accessible_context_emits_focus_context_changed_when_active_window_token_changes(self) -> None:
@@ -181,7 +184,7 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
self.assertEqual(snapshot.active_window_token, "4242:Terminal")
self.assertEqual(cthulhu_state.compositorSnapshot.active_window_token, "4242:Terminal")
def test_workspace_backend_normalizes_transition_completion(self) -> None:
def test_workspace_backend_normalizes_initial_done_and_handoff(self) -> None:
adapter = compositor_state_adapter.CompositorStateAdapter(workspace_backends=[])
events = []
adapter.add_listener(events.append)
@@ -189,28 +192,36 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
first_workspace = mock.Mock()
second_workspace = mock.Mock()
first_workspace_token = f"workspace:{id(first_workspace)}"
second_workspace_token = f"workspace:{id(second_workspace)}"
adapter._snapshot = compositor_state_types.DesktopContextSnapshot(
session_type="wayland",
backend_name="wayland-shared-protocols",
)
workspace.activate(adapter)
workspace._handle_workspace_name(first_workspace, "Workspace 1")
workspace._handle_workspace_coordinates(first_workspace, (0, 0))
workspace.activate(adapter._handle_workspace_signal)
workspace._handle_workspace_state(first_workspace, "active", True)
workspace._handle_workspace_manager_done()
self.assertEqual(adapter.get_snapshot().active_workspace_token, first_workspace_token)
self.assertEqual(adapter.get_snapshot().active_workspace_name, "Workspace 1")
self.assertEqual(adapter.get_snapshot().active_workspace_coordinates, (0, 0))
self.assertEqual(
[event.type for event in events],
[
compositor_state_types.DESKTOP_TRANSITION_STARTED,
compositor_state_types.PAUSE_ATSPI_CHURN,
compositor_state_types.WORKSPACE_STATE_CHANGED,
compositor_state_types.DESKTOP_TRANSITION_FINISHED,
compositor_state_types.RESUME_ATSPI_CHURN,
compositor_state_types.FLUSH_STALE_ATSPI_EVENTS,
],
)
self.assertEqual(events[0].snapshot.active_workspace_ids, frozenset({first_workspace_token}))
self.assertTrue(events[0].snapshot.workspace_transition_pending)
self.assertEqual(events[2].snapshot.active_workspace_ids, frozenset({first_workspace_token}))
self.assertFalse(adapter.get_snapshot().workspace_transition_pending)
self.assertEqual(cthulhu_state.compositorSnapshot.active_workspace_token, first_workspace_token)
self.assertEqual(adapter.get_snapshot().active_workspace_ids, frozenset({first_workspace_token}))
self.assertEqual(cthulhu_state.compositorSnapshot.active_workspace_ids, frozenset({first_workspace_token}))
events.clear()
workspace._handle_workspace_name(second_workspace, "Workspace 2")
workspace._handle_workspace_coordinates(second_workspace, (1, 0))
workspace._handle_workspace_id(second_workspace, "ws-2")
workspace._handle_workspace_state(first_workspace, "active", False)
workspace._handle_workspace_state(second_workspace, "active", True)
workspace._handle_workspace_manager_done()
@@ -229,17 +240,11 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
],
)
self.assertTrue(events[0].snapshot.workspace_transition_pending)
self.assertEqual(events[0].snapshot.active_workspace_token, first_workspace_token)
self.assertFalse(events[2].payload["is_transition_pending"])
self.assertEqual(events[2].snapshot.active_workspace_token, second_workspace_token)
self.assertEqual(events[2].snapshot.active_workspace_name, "Workspace 2")
self.assertEqual(events[2].snapshot.active_workspace_coordinates, (1, 0))
self.assertIs(events[2].payload["workspace_handle"], second_workspace)
self.assertEqual(adapter.get_snapshot().active_workspace_token, second_workspace_token)
self.assertEqual(adapter.get_snapshot().active_workspace_name, "Workspace 2")
self.assertEqual(adapter.get_snapshot().active_workspace_coordinates, (1, 0))
self.assertEqual(events[0].snapshot.active_workspace_ids, frozenset())
self.assertEqual(events[2].snapshot.active_workspace_ids, frozenset({"ws-2"}))
self.assertEqual(adapter.get_snapshot().active_workspace_ids, frozenset({"ws-2"}))
self.assertFalse(adapter.get_snapshot().workspace_transition_pending)
self.assertEqual(cthulhu_state.compositorSnapshot.active_workspace_token, second_workspace_token)
self.assertEqual(cthulhu_state.compositorSnapshot.active_workspace_ids, frozenset({"ws-2"}))
self.assertFalse(cthulhu_state.pauseAtspiChurn)
def test_event_manager_startup_resyncs_adapter_after_focus_recovery(self) -> None: