From f92fd330891bce02c53aa9a18a3aaa1e6b6285c1 Mon Sep 17 00:00:00 2001 From: Storm Dragon Date: Sun, 26 Jul 2026 17:36:43 -0400 Subject: [PATCH] Continue web refactor work. --- .../scripts/toolkits/Chromium/script.py | 19 +- src/cthulhu/scripts/toolkits/Gecko/script.py | 19 +- src/cthulhu/scripts/web/event_router.py | 121 ++++++++++++ src/cthulhu/scripts/web/meson.build | 3 +- src/cthulhu/scripts/web/script.py | 3 + tests/test_web_event_router_contract.py | 181 ++++++++++++++++++ 6 files changed, 335 insertions(+), 11 deletions(-) create mode 100644 src/cthulhu/scripts/web/event_router.py create mode 100644 tests/test_web_event_router_contract.py diff --git a/src/cthulhu/scripts/toolkits/Chromium/script.py b/src/cthulhu/scripts/toolkits/Chromium/script.py index 7da3440..e0da14e 100644 --- a/src/cthulhu/scripts/toolkits/Chromium/script.py +++ b/src/cthulhu/scripts/toolkits/Chromium/script.py @@ -39,6 +39,7 @@ from cthulhu.ax_object import AXObject from cthulhu.ax_utilities import AXUtilities from cthulhu.scripts import default from cthulhu.scripts import web +from cthulhu.scripts.web.event_router import RouteResult from .braille_generator import BrailleGenerator from .script_utilities import Utilities from .speech_generator import SpeechGenerator @@ -152,12 +153,20 @@ class Script(web.Script): def onCheckedChanged(self, event): """Callback for object:state-changed:checked accessibility events.""" - if super().onCheckedChanged(event): - return + def handle_shared_web(routed_event: object) -> RouteResult: + handled = web.Script.onCheckedChanged(self, routed_event) + return RouteResult.HANDLED if handled else RouteResult.CONTINUE - msg = "CHROMIUM: Passing along event to default script" - debug.printMessage(debug.LEVEL_INFO, msg, True) - default.Script.onCheckedChanged(self, event) + self._eventRouter.route_event( + event, + application_handler=None, + toolkit_handler=None, + web_handler=handle_shared_web, + default_handler=lambda routed_event: default.Script.onCheckedChanged( + self, + routed_event, + ), + ) def onColumnReordered(self, event): """Callback for object:column-reordered accessibility events.""" diff --git a/src/cthulhu/scripts/toolkits/Gecko/script.py b/src/cthulhu/scripts/toolkits/Gecko/script.py index f856d8f..032d9aa 100644 --- a/src/cthulhu/scripts/toolkits/Gecko/script.py +++ b/src/cthulhu/scripts/toolkits/Gecko/script.py @@ -42,6 +42,7 @@ from cthulhu.ax_object import AXObject from cthulhu.ax_utilities import AXUtilities from cthulhu.scripts import default from cthulhu.scripts import web +from cthulhu.scripts.web.event_router import RouteResult from .script_utilities import Utilities @@ -125,12 +126,20 @@ class Script(web.Script): def onCheckedChanged(self, event): """Callback for object:state-changed:checked accessibility events.""" - if super().onCheckedChanged(event): - return + def handle_shared_web(routed_event: object) -> RouteResult: + handled = web.Script.onCheckedChanged(self, routed_event) + return RouteResult.HANDLED if handled else RouteResult.CONTINUE - msg = "GECKO: Passing along event to default script" - debug.printMessage(debug.LEVEL_INFO, msg, True) - default.Script.onCheckedChanged(self, event) + self._eventRouter.route_event( + event, + application_handler=None, + toolkit_handler=None, + web_handler=handle_shared_web, + default_handler=lambda routed_event: default.Script.onCheckedChanged( + self, + routed_event, + ), + ) def onColumnReordered(self, event): """Callback for object:column-reordered accessibility events.""" diff --git a/src/cthulhu/scripts/web/event_router.py b/src/cthulhu/scripts/web/event_router.py new file mode 100644 index 0000000..f126bd0 --- /dev/null +++ b/src/cthulhu/scripts/web/event_router.py @@ -0,0 +1,121 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2026 Stormux +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the +# Free Software Foundation, Inc., Franklin Street, Fifth Floor, +# Boston MA 02110-1301 USA. + +"""Defines the ownership contract for routing web accessibility events. + +Handlers run in application, toolkit, and shared-web order. A handler returns +``HANDLED`` to claim the event, ``CONTINUE`` to offer it to the next layer, or +``DEFER_DEFAULT`` to skip directly to the default script. Invalid results and +handler exceptions fail safely by reaching the default handler exactly once. + +This module defines the Phase 2 routing boundary. Existing event families must +be migrated individually; creating a router does not activate parallel routing. +""" + +from __future__ import annotations + +import enum +from collections.abc import Callable + +from cthulhu import debug + + +class RouteResult(enum.Enum): + """A routing layer's explicit decision for one accessibility event.""" + + HANDLED = enum.auto() + CONTINUE = enum.auto() + DEFER_DEFAULT = enum.auto() + + +class RoutingLayer(enum.Enum): + """The layer which ultimately owns an accessibility event.""" + + APPLICATION = enum.auto() + TOOLKIT = enum.auto() + SHARED_WEB = enum.auto() + DEFAULT = enum.auto() + + +RouteHandler = Callable[[object], RouteResult] +DefaultHandler = Callable[[object], object] + + +class WebEventRouter: + """Routes one event to at most one owning presentation layer.""" + + def route_event( + self, + event: object, + *, + application_handler: RouteHandler | None, + toolkit_handler: RouteHandler | None, + web_handler: RouteHandler | None, + default_handler: DefaultHandler, + ) -> RoutingLayer: + """Route ``event`` in application-first order and return its owner.""" + + handlers = ( + (RoutingLayer.APPLICATION, application_handler), + (RoutingLayer.TOOLKIT, toolkit_handler), + (RoutingLayer.SHARED_WEB, web_handler), + ) + for layer, handler in handlers: + if handler is None: + continue + + try: + result = handler(event) + except Exception as error: # pylint: disable=broad-except + message = ( + f"WEB EVENT ROUTER: {layer.name} raised " + f"{type(error).__name__}; deferring to DEFAULT" + ) + debug.printMessage(debug.LEVEL_WARNING, message, True) + return self._route_to_default(event, default_handler) + + if not isinstance(result, RouteResult): + message = ( + f"WEB EVENT ROUTER: {layer.name} returned invalid result " + f"{type(result).__name__}; deferring to DEFAULT" + ) + debug.printMessage(debug.LEVEL_WARNING, message, True) + return self._route_to_default(event, default_handler) + + message = f"WEB EVENT ROUTER: {layer.name} -> {result.name}" + debug.printMessage(debug.LEVEL_INFO, message, True) + if result is RouteResult.HANDLED: + return layer + if result is RouteResult.DEFER_DEFAULT: + return self._route_to_default(event, default_handler) + + return self._route_to_default(event, default_handler) + + @staticmethod + def _route_to_default( + event: object, + default_handler: DefaultHandler, + ) -> RoutingLayer: + debug.printMessage( + debug.LEVEL_INFO, + "WEB EVENT ROUTER: DEFAULT -> HANDLED", + True, + ) + default_handler(event) + return RoutingLayer.DEFAULT diff --git a/src/cthulhu/scripts/web/meson.build b/src/cthulhu/scripts/web/meson.build index bc74e1a..5823f84 100644 --- a/src/cthulhu/scripts/web/meson.build +++ b/src/cthulhu/scripts/web/meson.build @@ -2,6 +2,7 @@ web_python_sources = files([ '__init__.py', 'bookmarks.py', 'braille_generator.py', + 'event_router.py', 'script.py', 'script_utilities.py', 'sound_generator.py', @@ -12,4 +13,4 @@ web_python_sources = files([ python3.install_sources( web_python_sources, subdir: 'cthulhu/scripts/web' -) \ No newline at end of file +) diff --git a/src/cthulhu/scripts/web/script.py b/src/cthulhu/scripts/web/script.py index fb6de01..df08dc0 100644 --- a/src/cthulhu/scripts/web/script.py +++ b/src/cthulhu/scripts/web/script.py @@ -64,6 +64,7 @@ from cthulhu.ax_utilities_event import AXUtilitiesEvent, TextEventReason from .bookmarks import Bookmarks from .braille_generator import BrailleGenerator +from .event_router import WebEventRouter from .sound_generator import SoundGenerator from .speech_generator import SpeechGenerator from .tutorial_generator import TutorialGenerator @@ -80,6 +81,8 @@ def _log_tokens(tokens, reason=None, timestamp=True, stack=False): class Script(default.Script): + _eventRouter = WebEventRouter() + def __init__(self, app): super().__init__(app) diff --git a/tests/test_web_event_router_contract.py b/tests/test_web_event_router_contract.py new file mode 100644 index 0000000..7e47efd --- /dev/null +++ b/tests/test_web_event_router_contract.py @@ -0,0 +1,181 @@ +import sys +import unittest +from pathlib import Path +from unittest import mock + +import gi + + +gi.require_version("Gdk", "3.0") +gi.require_version("Gtk", "3.0") + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from cthulhu.scripts.web.event_router import RouteResult +from cthulhu.scripts.web.event_router import RoutingLayer +from cthulhu.scripts.web.event_router import WebEventRouter +from cthulhu.scripts import default +from cthulhu.scripts import web +from cthulhu.scripts.toolkits.Chromium import script as chromium_script +from cthulhu.scripts.toolkits.Gecko import script as gecko_script + + +class WebEventRouterContractTests(unittest.TestCase): + def setUp(self): + self.event = object() + self.application_handler = mock.Mock(return_value=RouteResult.CONTINUE) + self.toolkit_handler = mock.Mock(return_value=RouteResult.CONTINUE) + self.web_handler = mock.Mock(return_value=RouteResult.CONTINUE) + self.default_handler = mock.Mock() + self.router = WebEventRouter() + + def route_event(self): + return self.router.route_event( + self.event, + application_handler=self.application_handler, + toolkit_handler=self.toolkit_handler, + web_handler=self.web_handler, + default_handler=self.default_handler, + ) + + def test_application_handler_owns_steam_event(self): + self.application_handler.return_value = RouteResult.HANDLED + + self.assertEqual(self.route_event(), RoutingLayer.APPLICATION) + + self.application_handler.assert_called_once_with(self.event) + self.toolkit_handler.assert_not_called() + self.web_handler.assert_not_called() + self.default_handler.assert_not_called() + + def test_toolkit_handler_owns_chromium_quirk(self): + self.toolkit_handler.return_value = RouteResult.HANDLED + + self.assertEqual(self.route_event(), RoutingLayer.TOOLKIT) + + self.application_handler.assert_called_once_with(self.event) + self.toolkit_handler.assert_called_once_with(self.event) + self.web_handler.assert_not_called() + self.default_handler.assert_not_called() + + def test_shared_web_handler_owns_document_event(self): + self.web_handler.return_value = RouteResult.HANDLED + + self.assertEqual(self.route_event(), RoutingLayer.SHARED_WEB) + + self.application_handler.assert_called_once_with(self.event) + self.toolkit_handler.assert_called_once_with(self.event) + self.web_handler.assert_called_once_with(self.event) + self.default_handler.assert_not_called() + + def test_unhandled_browser_ui_reaches_default_once(self): + self.assertEqual(self.route_event(), RoutingLayer.DEFAULT) + + self.application_handler.assert_called_once_with(self.event) + self.toolkit_handler.assert_called_once_with(self.event) + self.web_handler.assert_called_once_with(self.event) + self.default_handler.assert_called_once_with(self.event) + + def test_explicit_default_deferral_skips_remaining_handlers(self): + self.toolkit_handler.return_value = RouteResult.DEFER_DEFAULT + + self.assertEqual(self.route_event(), RoutingLayer.DEFAULT) + + self.application_handler.assert_called_once_with(self.event) + self.toolkit_handler.assert_called_once_with(self.event) + self.web_handler.assert_not_called() + self.default_handler.assert_called_once_with(self.event) + + def test_invalid_route_result_logs_and_reaches_default_once(self): + self.toolkit_handler.return_value = object() + + with mock.patch( + "cthulhu.scripts.web.event_router.debug.printMessage" + ) as print_message: + self.assertEqual(self.route_event(), RoutingLayer.DEFAULT) + + self.web_handler.assert_not_called() + self.default_handler.assert_called_once_with(self.event) + messages = [call.args[1] for call in print_message.call_args_list] + self.assertTrue(any("invalid result" in message for message in messages)) + self.assertTrue(any("TOOLKIT" in message for message in messages)) + + def test_handler_exception_logs_and_reaches_default_once(self): + self.toolkit_handler.side_effect = RuntimeError("broken handler") + + with mock.patch( + "cthulhu.scripts.web.event_router.debug.printMessage" + ) as print_message: + self.assertEqual(self.route_event(), RoutingLayer.DEFAULT) + + self.web_handler.assert_not_called() + self.default_handler.assert_called_once_with(self.event) + messages = [call.args[1] for call in print_message.call_args_list] + self.assertTrue(any("raised RuntimeError" in message for message in messages)) + self.assertTrue(any("TOOLKIT" in message for message in messages)) + + +class CheckedStateRoutingContractTests(unittest.TestCase): + def test_shared_web_owns_checked_state_without_default_fallback(self): + for toolkit_script in (chromium_script, gecko_script): + with self.subTest(toolkit=toolkit_script.__name__): + test_script = toolkit_script.Script.__new__(toolkit_script.Script) + test_script._eventRouter = WebEventRouter() + event = object() + with ( + mock.patch.object( + web.Script, + "onCheckedChanged", + return_value=True, + ) as web_handler, + mock.patch.object( + default.Script, + "onCheckedChanged", + ) as default_handler, + mock.patch( + "cthulhu.scripts.web.event_router.debug.printMessage" + ) as print_message, + ): + toolkit_script.Script.onCheckedChanged(test_script, event) + + web_handler.assert_called_once_with(test_script, event) + default_handler.assert_not_called() + messages = [call.args[1] for call in print_message.call_args_list] + self.assertIn( + "WEB EVENT ROUTER: SHARED_WEB -> HANDLED", + messages, + ) + + def test_unhandled_checked_state_reaches_default_once(self): + for toolkit_script in (chromium_script, gecko_script): + with self.subTest(toolkit=toolkit_script.__name__): + test_script = toolkit_script.Script.__new__(toolkit_script.Script) + test_script._eventRouter = WebEventRouter() + event = object() + with ( + mock.patch.object( + web.Script, + "onCheckedChanged", + return_value=False, + ) as web_handler, + mock.patch.object( + default.Script, + "onCheckedChanged", + ) as default_handler, + mock.patch( + "cthulhu.scripts.web.event_router.debug.printMessage" + ) as print_message, + ): + toolkit_script.Script.onCheckedChanged(test_script, event) + + web_handler.assert_called_once_with(test_script, event) + default_handler.assert_called_once_with(test_script, event) + messages = [call.args[1] for call in print_message.call_args_list] + self.assertIn( + "WEB EVENT ROUTER: DEFAULT -> HANDLED", + messages, + ) + + +if __name__ == "__main__": + unittest.main()