cthulhu/tests/test_web_event_router_contract.py

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,
                )


class ExpandedStateRoutingContractTests(unittest.TestCase):
    def test_shared_web_owns_expanded_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,
                        "onExpandedChanged",
                        return_value=True,
                    ) as web_handler,
                    mock.patch.object(
                        default.Script,
                        "onExpandedChanged",
                    ) as default_handler,
                    mock.patch(
                        "cthulhu.scripts.web.event_router.debug.printMessage"
                    ) as print_message,
                ):
                    toolkit_script.Script.onExpandedChanged(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_expanded_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,
                        "onExpandedChanged",
                        return_value=False,
                    ) as web_handler,
                    mock.patch.object(
                        default.Script,
                        "onExpandedChanged",
                    ) as default_handler,
                    mock.patch(
                        "cthulhu.scripts.web.event_router.debug.printMessage"
                    ) as print_message,
                ):
                    toolkit_script.Script.onExpandedChanged(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,
                )


class TableReorderRoutingContractTests(unittest.TestCase):
    def test_shared_web_owns_table_reorder_without_default_fallback(self):
        for toolkit_script in (chromium_script, gecko_script):
            for handler_name in ("onColumnReordered", "onRowReordered"):
                with self.subTest(
                    toolkit=toolkit_script.__name__,
                    handler=handler_name,
                ):
                    test_script = toolkit_script.Script.__new__(toolkit_script.Script)
                    test_script._eventRouter = WebEventRouter()
                    event = object()
                    with (
                        mock.patch.object(
                            web.Script,
                            handler_name,
                            return_value=True,
                        ) as web_handler,
                        mock.patch.object(
                            default.Script,
                            handler_name,
                        ) as default_handler,
                        mock.patch(
                            "cthulhu.scripts.web.event_router.debug.printMessage"
                        ) as print_message,
                    ):
                        getattr(toolkit_script.Script, handler_name)(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_table_reorder_reaches_default_once(self):
        for toolkit_script in (chromium_script, gecko_script):
            for handler_name in ("onColumnReordered", "onRowReordered"):
                with self.subTest(
                    toolkit=toolkit_script.__name__,
                    handler=handler_name,
                ):
                    test_script = toolkit_script.Script.__new__(toolkit_script.Script)
                    test_script._eventRouter = WebEventRouter()
                    event = object()
                    with (
                        mock.patch.object(
                            web.Script,
                            handler_name,
                            return_value=False,
                        ) as web_handler,
                        mock.patch.object(
                            default.Script,
                            handler_name,
                        ) as default_handler,
                        mock.patch(
                            "cthulhu.scripts.web.event_router.debug.printMessage"
                        ) as print_message,
                    ):
                        getattr(toolkit_script.Script, handler_name)(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()