195 lines
7.0 KiB
Python
195 lines
7.0 KiB
Python
|
"""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()
|