78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
"""事件存储测试：订阅推送、sequence 单调、run 隔离。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
|
||
import pytest
|
||
|
||
from app.modules.runtime.events import Event, InMemoryEventStore
|
||
|
||
|
||
async def test_subscribe_receives_appended_event() -> None:
|
||
store = InMemoryEventStore()
|
||
queue = store.subscribe("run_1")
|
||
await store.append(
|
||
Event(type="tool.started", source="tool", run_id="run_1", payload={"tool": "calc"})
|
||
)
|
||
event = await asyncio.wait_for(queue.get(), timeout=1.0)
|
||
assert event.type == "tool.started"
|
||
assert event.run_id == "run_1"
|
||
|
||
|
||
async def test_sequence_increments_per_run() -> None:
|
||
store = InMemoryEventStore()
|
||
first = await store.append(Event(type="tool.started", source="tool", run_id="r1"))
|
||
second = await store.append(Event(type="tool.completed", source="tool", run_id="r1"))
|
||
assert first.sequence == 1
|
||
assert second.sequence == 2
|
||
|
||
|
||
async def test_subscribe_isolated_by_run() -> None:
|
||
store = InMemoryEventStore()
|
||
queue_a = store.subscribe("run_a")
|
||
await store.append(Event(type="x", source="tool", run_id="run_b"))
|
||
with pytest.raises(asyncio.TimeoutError):
|
||
await asyncio.wait_for(queue_a.get(), timeout=0.3)
|
||
|
||
|
||
async def test_unsubscribe_stops_delivery() -> None:
|
||
store = InMemoryEventStore()
|
||
queue = store.subscribe("run_1")
|
||
store.unsubscribe("run_1", queue)
|
||
await store.append(Event(type="x", source="tool", run_id="run_1"))
|
||
with pytest.raises(asyncio.TimeoutError):
|
||
await asyncio.wait_for(queue.get(), timeout=0.3)
|
||
|
||
|
||
async def test_subscribe_by_pattern_glob_match() -> None:
|
||
"""subscribe_by_pattern('time.*') 仅收 type 匹配的事件，不收其它 type。"""
|
||
store = InMemoryEventStore()
|
||
queue = store.subscribe_by_pattern("time.*")
|
||
await store.append(Event(type="time.reunion", source="time_sense"))
|
||
await store.append(Event(type="run.started", source="run"))
|
||
# time.reunion 命中，立即可取（append 内 put_nowait 同步执行）
|
||
received = queue.get_nowait()
|
||
assert received.type == "time.reunion"
|
||
# run.started 不命中，队列应空
|
||
assert queue.empty()
|
||
|
||
|
||
async def test_unsubscribe_by_pattern_stops_delivery() -> None:
|
||
"""unsubscribe_by_pattern 后不再收到事件。"""
|
||
store = InMemoryEventStore()
|
||
queue = store.subscribe_by_pattern("time.*")
|
||
store.unsubscribe_by_pattern("time.*", queue)
|
||
await store.append(Event(type="time.reunion", source="time_sense"))
|
||
assert queue.empty()
|
||
|
||
|
||
async def test_subscribe_by_pattern_wildcard_all() -> None:
|
||
"""subscribe_by_pattern('*') 收所有 type（Dispatcher 用此订阅全量）。"""
|
||
store = InMemoryEventStore()
|
||
queue = store.subscribe_by_pattern("*")
|
||
await store.append(Event(type="time.reunion", source="time_sense"))
|
||
await store.append(Event(type="run.started", source="run"))
|
||
assert queue.get_nowait().type == "time.reunion"
|
||
assert queue.get_nowait().type == "run.started"
|