124 lines
4.7 KiB
Python
124 lines
4.7 KiB
Python
"""Dispatcher 单元测试：注册、glob 分发、异常隔离、生命周期。
|
||
|
||
单元测试直接驱动 Dispatcher（start 后手动塞 all_queue 验证 pump + consumer 行为），
|
||
不依赖 EventStore.append 的真实 notify 链路；端到端验证见 test_dispatcher_integration.py。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
|
||
import pytest
|
||
|
||
from app.modules.runtime.events import Event, InMemoryEventStore
|
||
from app.services.dispatcher import Dispatcher
|
||
|
||
|
||
async def test_register_before_start_stores_subscription() -> None:
|
||
"""register 在 start 前调用 → subscription 入表（pattern + handler 透传）。"""
|
||
dispatcher = Dispatcher(InMemoryEventStore())
|
||
|
||
async def handler(_event: Event) -> None:
|
||
pass
|
||
|
||
dispatcher.register("time.*", handler)
|
||
assert len(dispatcher._subscriptions) == 1
|
||
assert dispatcher._subscriptions[0].pattern == "time.*"
|
||
assert dispatcher._subscriptions[0].handler is handler
|
||
|
||
|
||
async def test_register_after_start_raises() -> None:
|
||
"""start 后 register 抛 RuntimeError（约束：handler 必须在 lifespan startup 前注册）。"""
|
||
dispatcher = Dispatcher(InMemoryEventStore())
|
||
|
||
async def handler(_event: Event) -> None:
|
||
pass
|
||
|
||
await dispatcher.start()
|
||
try:
|
||
with pytest.raises(RuntimeError, match="已启动"):
|
||
dispatcher.register("time.*", handler)
|
||
finally:
|
||
await dispatcher.stop()
|
||
|
||
|
||
async def test_pump_glob_match() -> None:
|
||
"""_pump 按 glob 分发：time.* subscription 收 time.reunion 不收 run.started；* subscription 全收。"""
|
||
dispatcher = Dispatcher(InMemoryEventStore())
|
||
received_time: list[Event] = []
|
||
received_all: list[Event] = []
|
||
|
||
async def time_handler(event: Event) -> None:
|
||
received_time.append(event)
|
||
|
||
async def all_handler(event: Event) -> None:
|
||
received_all.append(event)
|
||
|
||
dispatcher.register("time.*", time_handler)
|
||
dispatcher.register("*", all_handler)
|
||
await dispatcher.start()
|
||
try:
|
||
# 直接 put 到 all_queue（模拟 EventStore append 经 pattern notify 投递到 '*' 订阅）
|
||
assert dispatcher._all_queue is not None
|
||
dispatcher._all_queue.put_nowait(Event(type="time.reunion", source="time_sense"))
|
||
dispatcher._all_queue.put_nowait(Event(type="run.started", source="run"))
|
||
# 给 pump + consumer task 时间处理（asyncio 调度）
|
||
await asyncio.sleep(0.05)
|
||
assert [e.type for e in received_time] == ["time.reunion"]
|
||
assert [e.type for e in received_all] == ["time.reunion", "run.started"]
|
||
finally:
|
||
await dispatcher.stop()
|
||
|
||
|
||
async def test_consumer_isolates_handler_exception() -> None:
|
||
"""handler A raise → 被 try/except 吞（test 不崩）+ handler B 仍收后续事件（含触发异常的事件）。"""
|
||
dispatcher = Dispatcher(InMemoryEventStore())
|
||
received_b: list[Event] = []
|
||
|
||
async def handler_a(event: Event) -> None:
|
||
if event.type == "time.boom":
|
||
raise ValueError("handler A 故障")
|
||
|
||
async def handler_b(event: Event) -> None:
|
||
received_b.append(event)
|
||
|
||
dispatcher.register("time.*", handler_a)
|
||
dispatcher.register("time.*", handler_b)
|
||
await dispatcher.start()
|
||
try:
|
||
assert dispatcher._all_queue is not None
|
||
dispatcher._all_queue.put_nowait(Event(type="time.boom", source="test"))
|
||
dispatcher._all_queue.put_nowait(Event(type="time.reunion", source="test"))
|
||
await asyncio.sleep(0.05)
|
||
# handler A 对 time.boom raise（被 logger.exception 吞），handler B 两事件都收到
|
||
assert [e.type for e in received_b] == ["time.boom", "time.reunion"]
|
||
finally:
|
||
await dispatcher.stop()
|
||
|
||
|
||
async def test_start_stop_lifecycle() -> None:
|
||
"""start 后 pump + consumer task 存在 + all_queue 在 store 订阅表；stop 后 task cancelled + all_queue 从订阅表移除。"""
|
||
store = InMemoryEventStore()
|
||
dispatcher = Dispatcher(store)
|
||
|
||
async def handler(_event: Event) -> None:
|
||
pass
|
||
|
||
dispatcher.register("time.*", handler)
|
||
await dispatcher.start()
|
||
try:
|
||
assert dispatcher._pump_task is not None
|
||
assert not dispatcher._pump_task.cancelled()
|
||
assert len(dispatcher._consumer_tasks) == 1
|
||
assert dispatcher._all_queue is not None
|
||
assert dispatcher._all_queue in store._pattern_subscribers["*"]
|
||
finally:
|
||
all_queue = dispatcher._all_queue
|
||
await dispatcher.stop()
|
||
# stop 后状态清空
|
||
assert dispatcher._pump_task is None
|
||
assert dispatcher._consumer_tasks == []
|
||
assert dispatcher._all_queue is None
|
||
# all_queue 已从 store 的 pattern 订阅列表移除
|
||
assert all_queue not in store._pattern_subscribers.get("*", [])
|