"""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()