Workflow Testing

Ironflow provides an in-memory test client that allows you to unit test your functions without a running server, NATS, or a database. Tests run instantly and deterministically.

---

Installation

JavaScript

The test client is included in @ironflow/node:

npm install @ironflow/node

Import from the /test subpath:

import { createTestClient } from "@ironflow/node/test";

Go

The test client lives in the ironflowtest package:

import "github.com/sahina/ironflow-go/ironflow/ironflowtest"

Python

The test kit ships in the ironflow package as ironflow.testing:

from ironflow.testing import TestClient

---

How It Works

The test client replaces the durable step execution engine with a synchronous mock layer. When you call emit(), the client:

1. Resolves the function triggered by the event.
2. Executes the handler synchronously.
3. Replaces step.run() calls with your registered mocks.
4. Resolves step.sleep() and step.waitForEvent() immediately using pre-registered data.
5. Returns a TestRun object containing the final output and an audit log of all executed steps.

---

Quick Start

JavaScript

import { describe, it, expect } from "vitest";
import { createTestClient } from "@ironflow/node/test";
import { myWorkflow } from "./workflows";

describe("myWorkflow", () => {
  it("processes a successful checkout", async () => {
    const t = createTestClient({ functions: [myWorkflow] });

    // 1. Mock the internal steps
    t.mockStep("validate-order", () => ({ valid: true }));
    t.mockStep("charge-card", () => ({ txId: "tx_123" }));

    // 2. Execute the function
    const run = await t.emit("order.placed", { orderId: "ord_1" });

    // 3. Assert results
    expect(run.status).toBe("completed");
    expect(run.output).toMatchObject({ success: true });
    expect(run.stepOutput("charge-card")).toEqual({ txId: "tx_123" });
  });
});

Go

func TestMyWorkflow(t *testing.T) {
    tc := ironflowtest.NewClient(t, ironflowtest.Config{
        Functions: []ironflow.Function{MyWorkflow},
    })

    // 1. Mock internal steps
    tc.MockStep("validate-order", func() (any, error) {
        return map[string]bool{"valid": true}, nil
    })

    // 2. Execute
    run := tc.Emit(t, "order.placed", map[string]string{"orderId": "ord_1"})

    // 3. Assert
    if run.Status != "completed" {
        t.Errorf("expected completed, got %s", run.Status)
    }
}

Python

import asyncio
from ironflow.testing import TestClient
from workflows import my_workflow

def test_successful_checkout():
    tc = TestClient([my_workflow])

    # 1. Mock the internal steps
    tc.mock_step("validate-order", lambda: {"valid": True})
    tc.mock_step("charge-card", lambda: {"tx_id": "tx_123"})

    # 2. Execute the function (emit is async)
    run = asyncio.run(tc.emit("order.placed", {"order_id": "ord_1"}))

    # 3. Assert results
    assert run.status == "completed"
    assert run.step_output("charge-card") == {"tx_id": "tx_123"}

---

Testing Advanced Patterns

Testing step.waitForEvent

All three test kits resolve an optional wait immediately when no matching event is queued: use onTimeout: "return-null" in JavaScript, on_timeout="return-null" in Python, or WaitForEventOptional in Go (including branch scopes). Default waits still fail for a missing event fixture. Queueing a null payload creates a real event with null data; it does not simulate a timeout.

To test a workflow that pauses for an event, you must “prime” the test client with the expected event before emitting the trigger.

it("handles approval flow", async () => {
  const t = createTestClient({ functions: [approvalWorkflow] });

  // Pre-register the event that waitForEvent will find
  t.sendEvent("order.approved", { orderId: "ord_1", approvedBy: "alice" });

  const run = await t.emit("order.placed", { orderId: "ord_1" });
  expect(run.status).toBe("completed");
});

In Python, queue the event with send_event:

tc.send_event("order.approved", {"order_id": "ord_1", "approved_by": "alice"})
run = asyncio.run(tc.emit("order.placed", {"order_id": "ord_1"}))

In Go, queue the event with SendEvent:

tc.SendEvent("order.approved", map[string]string{"orderId": "ord_1", "approvedBy": "alice"})
run := tc.Emit(t, "order.placed", map[string]string{"orderId": "ord_1"})

Testing step.invoke

step.invoke() calls are mocked separately from step.run(), keyed by the target function ID. An unmocked invoke throws.

JavaScript

const t = createTestClient({ functions: [parentWorkflow] });

t.mockInvoke("child-workflow", (data) => ({ processed: true }));

const run = await t.emit("order.placed", { orderId: "ord_1" });
expect(run.status).toBe("completed");

Go

tc.MockInvoke("child-workflow", func(data any) (any, error) {
    return map[string]bool{"processed": true}, nil
})

Python

tc.mock_invoke("child-workflow", lambda data: {"processed": True})

Testing Sagas (Compensations)

The test client automatically tracks which compensations were registered and executes them in reverse order if a terminal failure occurs.

it("runs compensations on failure", async () => {
  const t = createTestClient({ functions: [orderSaga] });

  t.mockStep("charge", () => ({ id: "ch_1" }));
  t.mockStep("ship", () => { throw new NonRetryableError("Out of stock"); }); // NonRetryableError from "@ironflow/node"

  const run = await t.emit("order.placed", {});

  expect(run.status).toBe("failed");
  // Verify the 'charge' was compensated
  expect(run.compensationsRan).toContain("charge");
});

In Python, run.compensations_ran lists the compensations in the order they ran (newest first):

run = asyncio.run(tc.emit("order.placed", {}))
assert run.status == "failed"
assert "charge" in run.compensations_ran

In Go, run.CompensationsRan lists the step names in the order the compensations ran (newest first):

tc.MockStep("charge", func() (any, error) { return map[string]string{"id": "ch_1"}, nil })
tc.MockStep("ship", func() (any, error) { return nil, ironflow.NewNonRetryableError("Out of stock") })

run := tc.Emit(t, "order.placed", map[string]string{})
if run.Status != "failed" || run.CompensationsRan[0] != "charge" {
    t.Errorf("expected a failed run that compensated charge, got %s %v", run.Status, run.CompensationsRan)
}

---

Best Practices

1. Isolation: Use a fresh createTestClient for every test case to ensure no mock or event leakage.
2. Determinism: In JavaScript, parallel branches (step.parallel) run sequentially in test mode to ensure your assertions are stable. In Python they run concurrently under asyncio.gather, so assert on the results and not on the order of side effects.
3. Mock Depth: You only need to mock step.run and step.invoke. step.invokeAsync draws from the same mockInvoke registry keyed by function ID, so one mock covers both. Other primitives — step.sleep, step.sleepUntil, step.publish — are handled automatically.