cmdc_test

CMDC 集成方 / 第三方 plugin 作者 / cmdc 子库测试 helpers 一站式包。

cmdc 主库自身的 test/support/ 目录在 hex 发布时不会打包,导致集成方写 contract test 时都各自重写 mock provider / plugin runner / 事件断言族。 本子库把这些常见测试基础设施统一发布到 hex.pm。

安装

defp deps do
[
{:cmdc, "~> 0.5"},
{:cmdc_test, "~> 0.2", only: :test}
]
end

核心模块

模块用途
CMDCTest.MockProviderBuilder 式 mock LLM provider,配合 CMDC.Config.provider_fn 注入
CMDCTest.Plugin + Plugin.Spyrun_hook/3 单元测 Plugin;Spy plugin 集成路径 inject anonymous handler
CMDCTest.EventCaptureEventBus 订阅 + 清理
CMDCTest.Assertionsassert_event_emitted / assert_event_count / refute_event_emitted
CMDCTest.RAG.FixturesRAG / GraphRAG tool JSON、plugin event、evidence 与状态 fixture
CMDCTest.RAG.MockBackendfake Arcana search/answer backend
CMDCTest.RAG.MockGraphBackendfake GraphStore backend
CMDCTest.RAG.MockMaintenanceBackendfake reembed / GraphRAG maintenance backend
CMDCTest.RAG.MockPipelineRunnerfake pipeline runner,返回 grounding 与 pipeline summary
CMDCTest.RAG.MockStatusBackendfake knowledge index status backend
CMDCTest.RAG.Policycollection ACL / pipeline / graph profile 测试 user_data helper
CMDCTest.RAG.Assertionscitation、grounding、pipeline、GraphRAG evidence 断言

Quick Start

1. Mock LLM Provider

defmodule MyAgentTest do
use ExUnit.Case
alias CMDCTest.MockProvider
test "Agent 收到 prompt 后 LLM 回复" do
provider =
MockProvider.new()
|> MockProvider.respond("Hello from mock!")
{:ok, session} =
CMDC.create_agent(
model: "mock:test",
config: %{provider_fn: MockProvider.to_provider_fn(provider)}
)
CMDC.prompt(session, "hi")
{:ok, reply} = CMDC.collect_reply(session, timeout: 2_000)
assert reply == "Hello from mock!"
end
end

2. 单元测试 Plugin

import CMDCTest.Plugin
test "SecurityGuard 拦截危险 shell 命令" do
assert {:ok, {:block_tool, reason, _state}} =
run_hook(
CMDC.Plugin.Builtin.SecurityGuard,
{:before_tool, "shell", %{"cmd" => "rm -rf /"}}
)
assert reason =~ "dangerous"
end

3. 集成路径 hook 替换(Spy plugin)

alias CMDCTest.Plugin.Spy
test "Spy 拦截 before_tool 收集所有工具调用" do
test_pid = self()
{:ok, session} =
CMDC.create_agent(
model: "mock:test",
plugins: [
{Spy,
handler: fn
{:before_tool, name, args}, state, _ctx ->
send(test_pid, {:tool_intercepted, name, args})
{:continue, state}
_, state, _ -> {:continue, state}
end}
]
)
CMDC.prompt(session, "go")
assert_receive {:tool_intercepted, "shell", _}, 2_000
end

4. 事件断言族

import CMDCTest.Assertions
alias CMDCTest.EventCapture
test "Agent 执行后发出 agent_end 事件" do
{:ok, session} = CMDC.create_agent(...)
:ok = EventCapture.start_capture(session)
CMDC.prompt(session, "go")
# 等待事件
assert_event_emitted(session, :agent_end, timeout: 2_000)
# payload 子集匹配
assert_event_emitted(session, :tool_blocked, payload: %{tool: "shell"})
# 事件不应出现
refute_event_emitted(session, :approval_required, timeout: 200)
# 数事件次数
events = assert_event_count(session, :stream_chunk, 5, timeout: 1_000)
end

5. RAG / GraphRAG 测试支撑

import CMDCTest.RAG.Assertions
alias CMDCTest.RAG.{Fixtures, Policy}
test "RAG 输出必须有 citation 和 GraphRAG evidence" do
json = Fixtures.graph_search_tool_json()
assert_citations(json, min: 1, collection: "policies")
assert_graph_evidence(json, min_entities: 1, min_relationships: 1)
end
test "Agent user_data 使用 fake RAG backend" do
user_data = Policy.user_data(collections: ["policies"])
assert user_data.cmdc_rag_arcana[:backend] == CMDCTest.RAG.MockBackend
assert user_data.cmdc_rag_arcana[:graph_backend] == CMDCTest.RAG.MockGraphBackend
end

RAG helpers 是 shape-compatible contract,不依赖真实 Arcana DB、GraphStore、 embedding 或 LLM。集成方可以把这些模块填进 cmdc_rag_arcana config,用于 CI、Gateway event contract、AgentOps Trace Viewer 和 Eval Gate 测试。

License

Apache 2.0