CMDCTest.Assertions (cmdc_test v0.1.0)

Copy Markdown View Source

CMDC EventBus 事件断言族 — 替代手写 receive/after 模板。

必须先调 CMDCTest.EventCapture.start_capture/1 让当前进程订阅事件。

API

匹配语义:事件首元 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

assert_event_count(session, event_atom, n, opts \\ [])

@spec assert_event_count(CMDC.session(), atom(), non_neg_integer(), keyword()) :: [
  term()
]

断言恰好收到 N 个指定事件(在 timeout 内)。

返回收到的事件列表(按时序排列)。

assert_event_emitted(session, event_atom, opts \\ [])

@spec assert_event_emitted(CMDC.session(), atom(), keyword()) :: term()

断言当前进程在 timeout 内收到指定事件,返回匹配的 event term。

选项

  • :timeout — 最大等待时间(默认 500ms)
  • :payload — payload map 子集匹配(默认 :_ 不校验)

refute_event_emitted(session, event_atom, opts \\ [])

@spec refute_event_emitted(CMDC.session(), atom(), keyword()) :: :ok

断言当前进程在 timeout 内收到指定事件。

选项

  • :timeout — 等待时间(默认 200ms)
  • :payload — payload map 子集匹配