leocrm/tests/test_hooks.py

"""Tests for the WordPress-style hooks/filters system."""

from __future__ import annotations

import asyncio
import pytest

from app.core.hooks import (
    HookRegistry,
    get_hook_registry,
    do_action,
    apply_filters,
    reset_hook_registry_for_testing,
)


@pytest.fixture(autouse=True)
def clean_registry():
    """Reset the hook registry before each test."""
    reset_hook_registry_for_testing()
    yield
    reset_hook_registry_for_testing()


class TestHookRegistry:
    def test_singleton_identity(self):
        """HookRegistry is a singleton."""
        reg1 = get_hook_registry()
        reg2 = get_hook_registry()
        assert reg1 is reg2

    def test_register_action(self):
        """Actions can be registered and listed."""
        reg = get_hook_registry()
        called = []
        reg.register_action("test.action", lambda: called.append(True))
        assert reg.has_action("test.action")
        assert "test.action" in reg.list_actions()

    def test_register_filter(self):
        """Filters can be registered and listed."""
        reg = get_hook_registry()
        reg.register_filter("test.filter", lambda v: v + "!")
        assert reg.has_filter("test.filter")
        assert "test.filter" in reg.list_filters()

    @pytest.mark.asyncio
    async def test_do_action_calls_callback(self):
        """do_action executes registered callbacks."""
        reg = get_hook_registry()
        called = []
        reg.register_action("test.action", lambda: called.append("yes"))
        await do_action("test.action")
        assert called == ["yes"]

    @pytest.mark.asyncio
    async def test_do_action_with_args(self):
        """do_action passes arguments to callbacks."""
        reg = get_hook_registry()
        received = []
        reg.register_action("test.action", lambda x, y: received.append((x, y)))
        await do_action("test.action", 1, 2)
        assert received == [(1, 2)]

    @pytest.mark.asyncio
    async def test_do_action_async_callback(self):
        """do_action supports async callbacks."""
        reg = get_hook_registry()
        called = []

        async def async_cb():
            called.append("async")

        reg.register_action("test.action", async_cb)
        await do_action("test.action")
        assert called == ["async"]

    @pytest.mark.asyncio
    async def test_do_action_priority_order(self):
        """Actions execute in priority order (lower first)."""
        reg = get_hook_registry()
        order = []
        reg.register_action("test.action", lambda: order.append("low"), priority=20)
        reg.register_action("test.action", lambda: order.append("high"), priority=5)
        reg.register_action("test.action", lambda: order.append("mid"), priority=10)
        await do_action("test.action")
        assert order == ["high", "mid", "low"]

    @pytest.mark.asyncio
    async def test_do_action_no_callbacks(self):
        """do_action with no registered callbacks does nothing."""
        await do_action("nonexistent.action")

    @pytest.mark.asyncio
    async def test_do_action_swallows_exceptions(self):
        """do_action logs but does not raise on callback errors."""
        reg = get_hook_registry()
        called = []
        reg.register_action("test.action", lambda: (_ for _ in ()).throw(ValueError("boom")))
        reg.register_action("test.action", lambda: called.append("after_error"))
        await do_action("test.action")
        assert called == ["after_error"]

    @pytest.mark.asyncio
    async def test_apply_filters_modifies_value(self):
        """apply_filters passes value through callbacks."""
        reg = get_hook_registry()
        reg.register_filter("test.filter", lambda v: v.upper())
        result = await apply_filters("test.filter", "hello")
        assert result == "HELLO"

    @pytest.mark.asyncio
    async def test_apply_filters_chains_multiple(self):
        """apply_filters chains multiple callbacks in priority order."""
        reg = get_hook_registry()
        reg.register_filter("test.filter", lambda v: v + " B", priority=20)
        reg.register_filter("test.filter", lambda v: v + " A", priority=10)
        result = await apply_filters("test.filter", "start")
        assert result == "start A B"

    @pytest.mark.asyncio
    async def test_apply_filters_no_callbacks(self):
        """apply_filters with no callbacks returns original value."""
        result = await apply_filters("nonexistent.filter", "original")
        assert result == "original"

    @pytest.mark.asyncio
    async def test_apply_filters_async_callback(self):
        """apply_filters supports async callbacks."""
        reg = get_hook_registry()

        async def async_upper(v: str) -> str:
            return v.upper()

        reg.register_filter("test.filter", async_upper)
        result = await apply_filters("test.filter", "hello")
        assert result == "HELLO"

    def test_unregister_specific_callback(self):
        """unregister removes a specific callback."""
        reg = get_hook_registry()
        cb1 = lambda: None
        cb2 = lambda: None
        reg.register_action("test.action", cb1)
        reg.register_action("test.action", cb2)
        assert reg.has_action("test.action")
        reg.unregister("test.action", cb1)
        assert reg.has_action("test.action")
        reg.unregister("test.action", cb2)
        assert not reg.has_action("test.action")

    def test_unregister_all_for_plugin(self):
        """unregister_all_for_plugin removes hooks owned by a plugin instance."""
        reg = get_hook_registry()

        class FakePlugin:
            class manifest:
                name = "fake_plugin"

            def __init__(self):
                self.manifest = type("m", (), {"name": "fake_plugin"})()

            def my_action(self):
                pass

            def my_filter(self, v):
                return v
        plugin = FakePlugin()
        reg.register_action("test.action", plugin.my_action)
        reg.register_filter("test.filter", plugin.my_filter)
        assert reg.has_action("test.action")
        assert reg.has_filter("test.filter")
        reg.unregister_all_for_plugin("fake_plugin")
        assert not reg.has_action("test.action")
        assert not reg.has_filter("test.filter")

    @pytest.mark.asyncio
    async def test_apply_filters_swallows_exceptions(self):
        """apply_filters logs but does not raise on callback errors."""
        reg = get_hook_registry()
        reg.register_filter("test.filter", lambda v: (_ for _ in ()).throw(ValueError("boom")))
        reg.register_filter("test.filter", lambda v: v + "!")
        result = await apply_filters("test.filter", "test")
        # First filter errored, second still ran
        assert result == "test!"

    def test_reset_for_testing(self):
        """_reset_for_testing clears all state."""
        reg = get_hook_registry()
        reg.register_action("test.action", lambda: None)
        reg.register_filter("test.filter", lambda v: v)
        reg._reset_for_testing()
        assert not reg.list_actions()
        assert not reg.list_filters()