把工作流当作事件驱动系统来测试最容易。运行工作流、收集它流出的事件、断言最终结果，并在状态重要时检查 context 状态。

workflows.testing 模块内置了一个用于此目的的小型运行器：

from workflows.testing import WorkflowTestRunner

端到端测试

WorkflowTestRunner 会启动工作流、排空它的事件流、等待最终结果，并把全部内容打包在一个对象里返回：

import pytest

from workflows import Context, Workflow, step
from workflows.events import Event, StartEvent, StopEvent
from workflows.testing import WorkflowTestRunner

class Progress(Event):
    message: str

class Done(Event):
    value: str

class ExampleWorkflow(Workflow):
    @step
    async def start(self, ctx: Context, ev: StartEvent) -> Done:
        ctx.write_event_to_stream(Progress(message="started"))
        return Done(value=ev.topic.upper())

    @step
    async def finish(self, ev: Done) -> StopEvent:
        return StopEvent(result=ev.value)

@pytest.mark.asyncio
async def test_workflow_streams_progress_and_returns_result() -> None:
    result = await WorkflowTestRunner(ExampleWorkflow()).run(
        start_event=StartEvent(topic="docs")
    )

    assert result.result == "DOCS"
    assert result.event_types[Progress] == 1
    assert any(
        isinstance(ev, Progress) and ev.message == "started"
        for ev in result.collected
    )

返回的对象包含：

字段	含义
result	等待工作流 handler 结束后返回的最终值。
collected	所有未被排除的流式事件。
event_types	按事件类统计的已收集事件数量。
ctx	最终的 Context，可用于状态断言或快照。

内部事件

默认情况下，运行器会暴露内部事件，包括 StepStateChanged。当你想断言执行形态时这很有用：

from workflows.events import StepStateChanged

@pytest.mark.asyncio
async def test_step_state_events_are_emitted() -> None:
    result = await WorkflowTestRunner(ExampleWorkflow()).run(
        start_event=StartEvent(topic="docs")
    )

    assert result.event_types[StepStateChanged] > 0

如果某个测试只关心用户事件，可以把内部事件关掉：

result = await WorkflowTestRunner(ExampleWorkflow()).run(
    start_event=StartEvent(topic="docs"),
    expose_internal=False,
)

或者保留内部事件可用，但把嘈杂的那些从收集列表中排除：

result = await WorkflowTestRunner(ExampleWorkflow()).run(
    start_event=StartEvent(topic="docs"),
    exclude_events=[StepStateChanged]
)

状态断言

当工作流会写入 ctx.store 时，用返回的 context 来做断言：

from pydantic import BaseModel, Field

class CounterState(BaseModel):
    count: int = Field(default=0)

class CounterWorkflow(Workflow):
    @step
    async def count(self, ctx: Context[CounterState], ev: StartEvent) -> StopEvent:
        async with ctx.store.edit_state() as state:
            state.count += 1
        return StopEvent(result="done")

@pytest.mark.asyncio
async def test_workflow_updates_state() -> None:
    result = await WorkflowTestRunner(CounterWorkflow()).run()

    state = await result.ctx.store.get_state()
    assert state.count == 1

对于需要持久化的工作流代码，更推荐在一次真实的快照与恢复之后再断言行为：

workflow = CounterWorkflow()
first = await WorkflowTestRunner(workflow).run()

ctx_dict = first.ctx.to_dict()
restored_workflow = CounterWorkflow()
restored = Context.from_dict(restored_workflow, ctx_dict)

resumed = await WorkflowTestRunner(restored_workflow).run(ctx=restored)

这能抓住单元测试常漏掉的那类错误：无法序列化的状态、恢复时不可导入的事件，以及不宜重复执行的副作用。