CMDC EventBus 事件断言族 — 替代手写 receive/after 模板。
必须先调 CMDCTest.EventCapture.start_capture/1 让当前进程订阅事件。
API
assert_event_emitted/3— 期望收到指定事件refute_event_emitted/3— 期望不收到指定事件assert_event_count/4— 期望恰好收到 N 个指定事件
匹配语义:事件首元 atom 等于目标 atom 即匹配;payload 可选传 map 做子集匹配。
不匹配事件的处理
函数内部 receive 到的不匹配事件会重新 push 回当前进程 mailbox, 确保后续断言能继续看到它们。顺序会按 push 时序重排到 mailbox 尾部, 对绝大多数测试断言无影响。
Quick Start
import CMDCTest.Assertions
assert_event_emitted(session, :agent_end)
assert_event_emitted(session, :tool_blocked, payload: %{tool: "shell"})
assert_event_emitted(session, :token, timeout: 1_000)
refute_event_emitted(session, :approval_required, timeout: 200)
events = assert_event_count(session, :stream_chunk, 5, timeout: 1_000)
Summary
Functions
断言恰好收到 N 个指定事件(在 timeout 内)。
断言当前进程在 timeout 内收到指定事件,返回匹配的 event term。
断言当前进程在 timeout 内不收到指定事件。
Functions
@spec assert_event_count(CMDC.session(), atom(), non_neg_integer(), keyword()) :: [ term() ]
断言恰好收到 N 个指定事件(在 timeout 内)。
返回收到的事件列表(按时序排列)。
@spec assert_event_emitted(CMDC.session(), atom(), keyword()) :: term()
断言当前进程在 timeout 内收到指定事件,返回匹配的 event term。
选项
:timeout— 最大等待时间(默认 500ms):payload— payload map 子集匹配(默认:_不校验)
@spec refute_event_emitted(CMDC.session(), atom(), keyword()) :: :ok
断言当前进程在 timeout 内不收到指定事件。
选项
:timeout— 等待时间(默认 200ms):payload— payload map 子集匹配