---
title: Testing Workflows | Developer Documentation
---

Workflows are easiest to test as event-driven systems. Run the workflow, collect the events it streamed, assert the final result, and inspect the context state when state matters.

The `workflows.testing` module includes a small runner for this:

```
from workflows.testing import WorkflowTestRunner
```

## End-to-end tests

`WorkflowTestRunner` starts the workflow, drains its event stream, awaits the final result, and returns everything in one object:

```
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
    )
```

The returned object has:

| Field         | Meaning                                                        |
| ------------- | -------------------------------------------------------------- |
| `result`      | The final value returned by awaiting the workflow handler.     |
| `collected`   | Every streamed event that was not excluded.                    |
| `event_types` | A count of collected events by event class.                    |
| `ctx`         | The final `Context`, useful for state assertions or snapshots. |

## Internal events

By default the runner exposes internal events, including `StepStateChanged`. That is useful when you want to assert execution shape:

```
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
```

If a test only cares about user events, turn internal events off:

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

Or keep internal events available but exclude the noisy ones from the collected list:

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

## State assertions

Use the returned context when a workflow writes to `ctx.store`:

```
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
```

For durable workflow code, prefer asserting behavior after a real snapshot and restore:

```
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)
```

That catches the mistakes unit tests often miss: state that cannot serialize, events that are not importable when restored, and side effects that are not safe to repeat.