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

Continue web refactor work.

Browse Source
This commit is contained in:
Storm Dragon
2026-07-26 17:36:43 -04:00
parent 243854b761
commit f92fd33089
6 changed files with 335 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cthulhu/scripts/toolkits/Chromium/script.py
+14 -5
View File
@@ -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."""
src/cthulhu/scripts/toolkits/Gecko/script.py
+14 -5
View File
@@ -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."""
src/cthulhu/scripts/web/event_router.py
+121
View File
@@ -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
src/cthulhu/scripts/web/meson.build
+1
View File
@@ -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',
src/cthulhu/scripts/web/script.py
+3
View File
@@ -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)
tests/test_web_event_router_contract.py
+181
View File
@@ -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()