Clean lint surface and drop PyWayland backend

Remove the direct PyWayland workspace backend and local protocol wrapper from the Orca 50 rebase branch.

Keep the null/external workspace signal path, clear E402/F401 for this branch, and remove the Arch python-pywayland dependency.
This commit is contained in:
2026-04-11 21:17:21 -04:00
parent 09c03ad06a
commit 642fe6da66
73 changed files with 114 additions and 787 deletions
@@ -1,6 +1,4 @@
import importlib
import sys
import types
import unittest
from pathlib import Path
from unittest import mock
@@ -10,7 +8,6 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cthulhu import cthulhu_state
from cthulhu import compositor_state_adapter
from cthulhu import compositor_state_types
from cthulhu import compositor_state_wayland
class FakeWorkspaceBackend:
@@ -52,7 +49,7 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
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:
def test_default_workspace_backends_include_null_backend(self) -> None:
adapter = compositor_state_adapter.CompositorStateAdapter()
backend_types = [type(backend) for backend in adapter._workspaceBackends]
@@ -60,117 +57,15 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
self.assertEqual(
backend_types,
[
compositor_state_wayland.WaylandSharedProtocolsBackend,
compositor_state_wayland.NullWorkspaceBackend,
compositor_state_adapter.NullWorkspaceBackend,
],
)
def test_wayland_backend_is_unavailable_without_wayland_session(self) -> None:
backend = compositor_state_wayland.WaylandSharedProtocolsBackend()
with mock.patch.dict(compositor_state_wayland.os.environ, {}, clear=True):
self.assertFalse(backend.is_available("x11"))
self.assertFalse(backend.is_available("wayland"))
def test_wayland_backend_installs_dispatch_watch_and_processes_runtime_events(self) -> None:
fakeDisplay = mock.Mock()
fakeRegistry = mock.Mock()
fakeDisplay.connect = mock.Mock()
fakeDisplay.get_registry = mock.Mock(return_value=fakeRegistry)
fakeDisplay.get_fd = mock.Mock(return_value=17)
fakeDisplay.roundtrip = mock.Mock()
fakeDisplay.dispatch = mock.Mock(side_effect=[1, 0])
fakeProtocols = mock.Mock()
fakeProtocols.INTERFACE_NAME = "ext_workspace_manager_v1"
fakeProtocols.INTERFACE_VERSION = 1
fakeProtocols.ACTIVE_STATE_VALUE = 1
fakeProtocols.has_runtime_support.return_value = True
fakeProtocols.get_display_class.return_value = mock.Mock(return_value=fakeDisplay)
emitSignal = mock.Mock()
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.GLib, "io_add_watch", return_value=41) as ioAddWatch,
mock.patch.object(compositor_state_wayland.GLib, "source_remove") as sourceRemove,
):
backend.activate(emitSignal)
ioAddWatch.assert_called_once()
watchCallback = ioAddWatch.call_args.args[3]
self.assertTrue(watchCallback(17, compositor_state_wayland.GLib.IO_IN))
self.assertEqual(fakeDisplay.dispatch.call_count, 2)
backend.deactivate()
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:
from cthulhu.wayland_protocols import ext_workspace_v1
fakeClientModule = types.ModuleType("pywayland.client")
class FakeDisplay:
pass
fakeClientModule.Display = FakeDisplay
def fake_import(moduleName: str, package=None):
if moduleName in ("pywayland.client", "pywayland.client.display"):
return fakeClientModule
raise ImportError(moduleName)
try:
with mock.patch("importlib.import_module", side_effect=fake_import):
importlib.reload(ext_workspace_v1)
self.assertTrue(ext_workspace_v1.has_runtime_support())
self.assertIs(ext_workspace_v1.get_display_class(), FakeDisplay)
self.assertIsNotNone(ext_workspace_v1.ExtWorkspaceManagerV1)
self.assertIsNotNone(ext_workspace_v1.ExtWorkspaceHandleV1)
self.assertEqual(
ext_workspace_v1.ExtWorkspaceManagerV1.interface_name,
"ext_workspace_manager_v1",
)
self.assertEqual(
ext_workspace_v1.ExtWorkspaceHandleV1.interface_name,
"ext_workspace_handle_v1",
)
finally:
importlib.reload(ext_workspace_v1)
def test_wayland_backend_is_selected_when_available(self) -> None:
def test_injected_workspace_backend_is_selected_when_available(self) -> None:
adapter = compositor_state_adapter.CompositorStateAdapter()
selected_backend = mock.Mock(name="selected-backend")
fallback_backend = mock.Mock(name="fallback-backend")
selected_backend.name = "wayland-shared-protocols"
selected_backend.name = "external-workspace-signals"
selected_backend.is_available.return_value = True
fallback_backend.name = "null"
fallback_backend.is_available.return_value = True
@@ -181,7 +76,7 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
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")
self.assertEqual(adapter.get_snapshot().backend_name, "external-workspace-signals")
def test_activate_is_idempotent_and_deactivates_previous_backend(self) -> None:
backend = FakeWorkspaceBackend(True, "selected")
@@ -245,24 +140,31 @@ 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_initial_done_and_handoff(self) -> None:
def test_external_workspace_signal_normalizes_transition_and_handoff(self) -> None:
adapter = compositor_state_adapter.CompositorStateAdapter(workspace_backends=[])
events = []
adapter.add_listener(events.append)
workspace = compositor_state_wayland.WaylandSharedProtocolsBackend()
first_workspace = mock.Mock()
second_workspace = mock.Mock()
first_workspace_token = f"workspace:{id(first_workspace)}"
adapter._snapshot = compositor_state_types.DesktopContextSnapshot(
session_type="wayland",
backend_name="wayland-shared-protocols",
backend_name="external-workspace-signals",
)
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()
adapter._handle_workspace_signal(
compositor_state_types.DESKTOP_TRANSITION_STARTED,
["workspace-1"],
"transition-started",
)
adapter._handle_workspace_signal(
compositor_state_types.WORKSPACE_STATE_CHANGED,
["workspace-1"],
"workspace-changed",
)
adapter._handle_workspace_signal(
compositor_state_types.DESKTOP_TRANSITION_FINISHED,
["workspace-1"],
"transition-finished",
)
self.assertEqual(
[event.type for event in events],
@@ -275,18 +177,29 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
compositor_state_types.FLUSH_STALE_ATSPI_EVENTS,
],
)
self.assertEqual(events[0].snapshot.active_workspace_ids, frozenset({first_workspace_token}))
self.assertEqual(events[0].snapshot.active_workspace_ids, frozenset({"workspace-1"}))
self.assertTrue(events[0].snapshot.workspace_transition_pending)
self.assertEqual(events[2].snapshot.active_workspace_ids, frozenset({first_workspace_token}))
self.assertEqual(events[2].snapshot.active_workspace_ids, frozenset({"workspace-1"}))
self.assertFalse(adapter.get_snapshot().workspace_transition_pending)
self.assertEqual(adapter.get_snapshot().active_workspace_ids, frozenset({first_workspace_token}))
self.assertEqual(cthulhu_state.compositorSnapshot.active_workspace_ids, frozenset({first_workspace_token}))
self.assertEqual(adapter.get_snapshot().active_workspace_ids, frozenset({"workspace-1"}))
self.assertEqual(cthulhu_state.compositorSnapshot.active_workspace_ids, frozenset({"workspace-1"}))
events.clear()
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()
adapter._handle_workspace_signal(
compositor_state_types.DESKTOP_TRANSITION_STARTED,
[],
"transition-started",
)
adapter._handle_workspace_signal(
compositor_state_types.WORKSPACE_STATE_CHANGED,
["ws-2"],
"workspace-changed",
)
adapter._handle_workspace_signal(
compositor_state_types.DESKTOP_TRANSITION_FINISHED,
["ws-2"],
"transition-finished",
)
event_types = [event.type for event in events]
@@ -7,7 +7,6 @@ from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cthulhu import cthulhu
from cthulhu import settings
from cthulhu import cthulhu_gui_prefs
from cthulhu import settings_manager