Testing kit¶

FlowTestKit — helpers for writing compact unit tests over flows, including single-run execution, node-sequence assertions, and error simulation.

testkit ¶

Utilities for writing concise PenguiFlow tests.

The helpers in this module provide a minimal harness around PenguiFlow so unit tests can focus on the behaviour of their nodes instead of the runtime plumbing. Each helper intentionally works with the public runtime surface to avoid relying on private attributes, keeping the harness forward compatible with the v1 API.

run_one `async` ¶

run_one(flow: PenguiFlow, message: Message, *, registry: Any | None = None, timeout_s: float | None = 1.0) -> Any

Run message through flow and return the first Rookery payload.

The flow is started and stopped for the caller. The original message's trace_id is tracked so :func:assert_node_sequence can introspect the execution order afterwards.

assert_preserves_message_envelope `async` ¶

assert_preserves_message_envelope(node: Callable[[Message, Any], Awaitable[Any]] | Any, *, message: Message | None = None, ctx: Any | None = None) -> Message

Execute node and assert it preserves the Message envelope.

Parameters:

Name	Type	Description	Default
`node`	`Callable[[Message, Any], Awaitable[Any]] \| Any`	Either a bare async callable or a :class:`penguiflow.node.Node` whose first parameter is a :class:`~penguiflow.types.Message` instance.	required
`message`	`Message \| None`	Optional sample message. When omitted, a minimal envelope is synthesised.	`None`
`ctx`	`Any \| None`	Optional context object passed to the node. By default a stub context is used that simply no-ops `emit`/`emit_nowait`.	`None`

Returns:

Name	Type	Description
`Message`	`Message`	The resulting message from the node, allowing additional assertions.

Raises:

Type	Description
`AssertionError`	If the node does not return a `Message` or mutates core envelope fields (headers or trace_id).
`TypeError`	If `node` is not awaitable.

assert_node_sequence ¶

assert_node_sequence(trace_id: str, expected: Sequence[str]) -> None

Assert that expected matches the recorded node start order.

get_recorded_events ¶

get_recorded_events(trace_id: str) -> tuple[FlowEvent, ...]

Return the recorded :class:FlowEvent history for trace_id.

The FlowTestKit recorder maintains a bounded cache of trace histories. This helper exposes the immutable snapshot so tests can assert on diagnostics such as node_failed payloads or retry attempts without touching the private cache directly.

simulate_error ¶

simulate_error(node_name: str, code: FlowErrorCode | str, *, fail_times: int = 1, result: Any | None = None, result_factory: Callable[[Any], Awaitable[Any] | Any] | None = None, exception_type: type[Exception] = RuntimeError) -> Callable[[Any, Any], Awaitable[Any]]

Return an async callable that fails fail_times before succeeding.

The returned coroutine is suitable for wrapping in :class:~penguiflow.node.Node and is especially useful for retry-centric tests. By default the callable echoes the incoming message once the simulated failures are exhausted, but result/result_factory can override the successful return value.