86 lines
3.1 KiB
Python
86 lines
3.1 KiB
Python
"""Dispatcher 端到端集成测试：InMemoryEventStore + Dispatcher + fake handler 全链路。
|
||
|
||
验证 EventStore.append → pattern notify → Dispatcher all_queue → pump glob →
|
||
per-subscription queue → consumer handler 的完整链路（单元测试 test_dispatcher.py 不覆盖
|
||
EventStore 与 Dispatcher 的接合）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
|
||
from app.modules.runtime.events import Event, InMemoryEventStore
|
||
from app.services.dispatcher import Dispatcher
|
||
|
||
|
||
async def test_end_to_end_inmemory_store_to_handler() -> None:
|
||
"""append time.reunion → handler 收到；append run.started → handler 不收到。"""
|
||
store = InMemoryEventStore()
|
||
dispatcher = Dispatcher(store)
|
||
received: list[Event] = []
|
||
|
||
async def handler(event: Event) -> None:
|
||
received.append(event)
|
||
|
||
dispatcher.register("time.*", handler)
|
||
await dispatcher.start()
|
||
try:
|
||
await store.append(Event(type="time.reunion", source="time_sense"))
|
||
await store.append(Event(type="run.started", source="run"))
|
||
await asyncio.sleep(0.05)
|
||
assert [e.type for e in received] == ["time.reunion"]
|
||
finally:
|
||
await dispatcher.stop()
|
||
|
||
|
||
async def test_end_to_end_multiple_handlers_same_event() -> None:
|
||
"""handler_A(time.*) + handler_B(time.reunion)：time.reunion 两者都收；time.sentiment 仅 A 收。"""
|
||
store = InMemoryEventStore()
|
||
dispatcher = Dispatcher(store)
|
||
received_a: list[Event] = []
|
||
received_b: list[Event] = []
|
||
|
||
async def handler_a(event: Event) -> None:
|
||
received_a.append(event)
|
||
|
||
async def handler_b(event: Event) -> None:
|
||
received_b.append(event)
|
||
|
||
dispatcher.register("time.*", handler_a)
|
||
dispatcher.register("time.reunion", handler_b)
|
||
await dispatcher.start()
|
||
try:
|
||
await store.append(Event(type="time.reunion", source="time_sense"))
|
||
await store.append(Event(type="time.sentiment", source="time_sense"))
|
||
await asyncio.sleep(0.05)
|
||
assert sorted(e.type for e in received_a) == ["time.reunion", "time.sentiment"]
|
||
assert [e.type for e in received_b] == ["time.reunion"]
|
||
finally:
|
||
await dispatcher.stop()
|
||
|
||
|
||
async def test_handler_exception_does_not_block_dispatch() -> None:
|
||
"""handler raise 后，后续 append 的事件仍被其它 handler 收到（at-most-once 隔离）。"""
|
||
store = InMemoryEventStore()
|
||
dispatcher = Dispatcher(store)
|
||
received_b: list[Event] = []
|
||
|
||
async def handler_a(_event: Event) -> None:
|
||
raise RuntimeError("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:
|
||
await store.append(Event(type="time.reunion", source="time_sense"))
|
||
await asyncio.sleep(0.02)
|
||
await store.append(Event(type="time.sentiment", source="time_sense"))
|
||
await asyncio.sleep(0.05)
|
||
# handler_a 两次都 raise（被吞），handler_b 两次都收到
|
||
assert [e.type for e in received_b] == ["time.reunion", "time.sentiment"]
|
||
finally:
|
||
await dispatcher.stop()
|