Replace event manager with Orca 50 model

This commit is contained in:
2026-04-11 03:12:21 -04:00
parent b641d6d8a4
commit bff131db18
6 changed files with 2793 additions and 1913 deletions
@@ -8,11 +8,9 @@ from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cthulhu import cthulhu_state
from cthulhu import cthulhu
from cthulhu import compositor_state_adapter
from cthulhu import compositor_state_types
from cthulhu import compositor_state_wayland
from cthulhu.wayland_protocols import ext_workspace_v1
class FakeWorkspaceBackend:
@@ -136,6 +134,8 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
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:
@@ -309,29 +309,5 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
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:
adapter = mock.Mock()
adapter.sync_accessible_context = mock.Mock(return_value=None)
listener = mock.Mock()
window = object()
focusedObject = object()
with (
mock.patch.object(cthulhu.event_manager.Atspi.EventListener, "new", return_value=listener),
mock.patch.object(cthulhu.event_manager.AXUtilities, "can_be_active_window", return_value=False),
mock.patch.object(cthulhu.event_manager.AXUtilities, "find_active_window", return_value=window),
mock.patch.object(cthulhu.event_manager.AXUtilities, "get_focused_object", return_value=focusedObject),
mock.patch.object(cthulhu.event_manager.cthulhu, "setActiveWindow") as setActiveWindow,
mock.patch.object(cthulhu.event_manager.cthulhu, "setLocusOfFocus") as setLocusOfFocus,
):
manager = cthulhu.event_manager.EventManager(mock.Mock())
manager.set_compositor_state_adapter(adapter)
manager._sync_focus_on_startup()
setActiveWindow.assert_called_once_with(window, alsoSetLocusOfFocus=True, notifyScript=False)
setLocusOfFocus.assert_called_once_with(None, focusedObject, notifyScript=True, force=True)
adapter.sync_accessible_context.assert_called_once_with("event-manager-startup")
if __name__ == "__main__":
unittest.main()
File diff suppressed because it is too large Load Diff
@@ -1,209 +0,0 @@
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()
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
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_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])
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_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()
@@ -1,193 +0,0 @@
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
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 EventManagerRelevanceGateRegressionTests(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.steamAppPatch = mock.patch.object(event_manager.EventManager, "_isSteamApp", return_value=False)
self.steamNotificationPatch = mock.patch.object(
event_manager.EventManager,
"_isSteamNotificationEvent",
return_value=False,
)
self.steamAppPatch.start()
self.steamNotificationPatch.start()
self.addCleanup(self.steamAppPatch.stop)
self.addCleanup(self.steamNotificationPatch.stop)
self.originalActiveScript = cthulhu_state.activeScript
self.originalLocusOfFocus = cthulhu_state.locusOfFocus
self.addCleanup(self._restore_state)
self.manager = event_manager.EventManager(mock.Mock(), asyncMode=False)
self.manager._active = True
cthulhu_state.activeScript = None
cthulhu_state.locusOfFocus = None
def _restore_state(self) -> None:
cthulhu_state.activeScript = self.originalActiveScript
cthulhu_state.locusOfFocus = self.originalLocusOfFocus
def test_unfocused_web_link_name_change_is_dropped_when_focus_is_stable_elsewhere(self) -> None:
app = object()
focus = object()
source = object()
event = FakeEvent(
"object:property-change:accessible-name",
source=source,
any_data="Diablo II: Resurrected Infernal Edition $39.99",
)
cthulhu_state.activeScript = mock.Mock(app=app)
cthulhu_state.locusOfFocus = focus
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="Chromium"),
mock.patch.object(event_manager.AXObject, "get_role", return_value=event_manager.Atspi.Role.LINK),
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, "is_focused", return_value=False),
mock.patch.object(event_manager.AXUtilities, "is_notification", return_value=False),
mock.patch.object(event_manager.AXUtilities, "is_alert", return_value=False),
):
self.assertTrue(self.manager._ignore(event))
def test_focused_web_link_name_change_is_kept(self) -> None:
app = object()
source = object()
event = FakeEvent("object:property-change:accessible-name", source=source, any_data="Cookie Clicker")
cthulhu_state.activeScript = mock.Mock(app=app)
cthulhu_state.locusOfFocus = source
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="Chromium"),
mock.patch.object(event_manager.AXObject, "get_role", return_value=event_manager.Atspi.Role.LINK),
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, "is_focused", return_value=True),
mock.patch.object(event_manager.AXUtilities, "is_notification", return_value=False),
mock.patch.object(event_manager.AXUtilities, "is_alert", return_value=False),
):
self.assertFalse(self.manager._ignore(event))
def test_live_region_name_change_is_kept_even_when_unfocused(self) -> None:
app = object()
focus = object()
source = object()
event = FakeEvent("object:property-change:accessible-name", source=source, any_data="sale updated")
cthulhu_state.activeScript = mock.Mock(app=app)
cthulhu_state.locusOfFocus = focus
def get_attribute(obj, name):
if name == "live":
return "polite"
return ""
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="Chromium"),
mock.patch.object(event_manager.AXObject, "get_role", return_value=event_manager.Atspi.Role.LINK),
mock.patch.object(event_manager.AXObject, "get_attribute", side_effect=get_attribute),
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, "is_focused", return_value=False),
mock.patch.object(event_manager.AXUtilities, "is_notification", return_value=False),
mock.patch.object(event_manager.AXUtilities, "is_alert", return_value=False),
):
self.assertFalse(self.manager._ignore(event))
def test_unfocused_focus_loss_is_dropped_when_focus_is_stable_elsewhere(self) -> None:
app = object()
focus = object()
source = object()
event = FakeEvent("object:state-changed:focused", source=source, detail1=0)
cthulhu_state.activeScript = mock.Mock(app=app)
cthulhu_state.locusOfFocus = focus
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="Chromium"),
mock.patch.object(event_manager.AXObject, "get_role", return_value=event_manager.Atspi.Role.LINK),
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, "is_notification", return_value=False),
mock.patch.object(event_manager.AXUtilities, "is_alert", return_value=False),
):
self.assertTrue(self.manager._ignore(event))
def test_repeated_web_children_changed_burst_is_collapsed(self) -> None:
app = object()
focus = object()
source = object()
firstEvent = FakeEvent("object:children-changed:add", source=source, any_data=object())
secondEvent = FakeEvent("object:children-changed:add", source=source, any_data=object())
cthulhu_state.activeScript = mock.Mock(app=app)
cthulhu_state.locusOfFocus = focus
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="Chromium"),
mock.patch.object(event_manager.AXObject, "get_role", return_value=event_manager.Atspi.Role.SECTION),
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),
mock.patch.object(event_manager.AXUtilities, "is_notification", return_value=False),
mock.patch.object(event_manager.AXUtilities, "is_alert", return_value=False),
):
self.assertFalse(self.manager._ignore(firstEvent))
self.assertTrue(self.manager._ignore(secondEvent))