Authoring¶

Install the testing SDK¶

TypeScriptPythonJavaGoC#

npm install --save-dev @aws/durable-execution-sdk-js-testing

pip install aws-durable-execution-sdk-python-testing

Add the testing dependency to your build file with test scope:

<dependency>
  <groupId>software.amazon.lambda.durable</groupId>
  <artifactId>aws-durable-execution-sdk-java-testing</artifactId>
  <version>2.2.0</version>
  <scope>test</scope>
</dependency>

durabletest ships in the same module as the runtime SDK. Add the SDK, then import github.com/aws/aws-durable-execution-sdk-go/durable/durabletest from your test files.

go get github.com/aws/aws-durable-execution-sdk-go

Add the testing package to your test project:

dotnet add package Amazon.Lambda.DurableExecution.Testing

Write a minimal test¶

Create a runner with your handler, call run(), and assert on the result.

TypeScriptPythonJavaGoC#

Call setupTestEnvironment() in beforeAll and teardownTestEnvironment() in afterAll. Create a new runner instance in beforeEach so each test starts with a clean state.

import { withDurableExecution, DurableContext } from "@aws/durable-execution-sdk-js";
import {
  LocalDurableTestRunner,
} from "@aws/durable-execution-sdk-js-testing";
import { ExecutionStatus } from "@aws-sdk/client-lambda";

const handler = withDurableExecution(async (event: unknown, context: DurableContext) => {
  const result = await context.step("greet", () => "hello");
  return result;
});

let runner: LocalDurableTestRunner;

beforeAll(async () => {
  await LocalDurableTestRunner.setupTestEnvironment();
});

afterAll(async () => {
  await LocalDurableTestRunner.teardownTestEnvironment();
});

beforeEach(() => {
  runner = new LocalDurableTestRunner({ handlerFunction: handler });
});

it("returns the expected result", async () => {
  const result = await runner.run();

  expect(result.getStatus()).toBe(ExecutionStatus.SUCCEEDED);
  expect(result.getResult()).toBe("hello");
});

Use DurableFunctionTestRunner as a context manager. The context manager starts the scheduler thread on entry and stops it on exit.

from aws_durable_execution_sdk_python import DurableContext, durable_execution, durable_step
from aws_durable_execution_sdk_python.types import StepContext
from aws_durable_execution_sdk_python.execution import InvocationStatus
from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner


@durable_step
def greet(ctx: StepContext) -> str:
    return "hello"


@durable_execution
def handler(event, context: DurableContext) -> str:
    return context.step(greet())


def test_returns_expected_result():
    runner = DurableFunctionTestRunner(handler=handler)
    with runner:
        result = runner.run(timeout=10)

    assert result.status is InvocationStatus.SUCCEEDED
    assert result.result == '"hello"'

Use LocalDurableTestRunner.create() with the input type and a handler function. Call runUntilComplete() to drive the full replay loop until the execution completes or fails.

import static org.junit.jupiter.api.Assertions.*;

import org.junit.jupiter.api.Test;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.model.ExecutionStatus;
import software.amazon.lambda.durable.testing.LocalDurableTestRunner;

class MinimalTest {

    @Test
    void returnsExpectedResult() {
        var runner = LocalDurableTestRunner.create(
            Void.class,
            (input, context) -> context.step("greet", String.class, ctx -> "hello")
        );

        var result = runner.runUntilComplete(null);

        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
        assertEquals("hello", result.getResult(String.class));
    }
}

Create a runner with NewLocalRunner, drive it with RunUntilComplete, and read the typed result with durabletest.ResultAs[T]. The runner needs no test handle, setup, or teardown.

// Save this file as minimal_test.go and run go test.
package main

import (
    "testing"

    "github.com/aws/aws-durable-execution-sdk-go/durable"
    "github.com/aws/aws-durable-execution-sdk-go/durable/durabletest"
)

func greetHandler(ctx durable.Context, _ any) (string, error) {
    return durable.Step[string](ctx, "greet", func(durable.StepContext) (string, error) {
        return "hello", nil
    })
}

func TestMinimal(t *testing.T) {
    runner := durabletest.NewLocalRunner(greetHandler)

    result, err := runner.RunUntilComplete(nil)
    if err != nil {
        t.Fatal(err)
    }
    if result.Status != durabletest.Succeeded {
        t.Fatalf("status = %s, want SUCCEEDED\n%s", result.Status, result.FormatTree())
    }

    got, err := durabletest.ResultAs[string](result)
    if err != nil {
        t.Fatal(err)
    }
    if got != "hello" {
        t.Errorf("result = %q, want hello", got)
    }
}

Construct DurableTestRunner<TIn, TOut> with your workflow delegate and new TestRunnerOptions { SkipTime = true }, call RunAsync(), then EnsureSucceeded() and assert on Result.

using Amazon.Lambda.DurableExecution;
using Amazon.Lambda.DurableExecution.Testing;
using Xunit;

public class MinimalTest
{
    private static Task<string> Workflow(object? input, IDurableContext ctx)
        => ctx.StepAsync(async (_, _) => "hello", name: "greet");

    [Fact]
    public async Task ReturnsExpectedResult()
    {
        await using var runner = new DurableTestRunner<object?, string>(
            Workflow,
            new TestRunnerOptions { SkipTime = true });

        TestResult<string> result = await runner.RunAsync(null);

        result.EnsureSucceeded();
        Assert.Equal("hello", result.Result);
    }
}

Test a failed execution¶

When your handler throws outside a step, or a step exhausts all retries, the execution fails. Assert on the status and inspect the error.

TypeScriptPythonJavaGoC#

import { withDurableExecution, DurableContext } from "@aws/durable-execution-sdk-js";
import { LocalDurableTestRunner } from "@aws/durable-execution-sdk-js-testing";
import { ExecutionStatus } from "@aws-sdk/client-lambda";

const handler = withDurableExecution(async (event: { fail: boolean }, context: DurableContext) => {
  if (event.fail) {
    throw new Error("intentional failure");
  }
  return "ok";
});

let runner: LocalDurableTestRunner;

beforeAll(async () => {
  await LocalDurableTestRunner.setupTestEnvironment();
});

afterAll(async () => {
  await LocalDurableTestRunner.teardownTestEnvironment();
});

beforeEach(() => {
  runner = new LocalDurableTestRunner({ handlerFunction: handler });
});

it("reports a failed execution", async () => {
  const result = await runner.run({ payload: { fail: true } });

  expect(result.getStatus()).toBe(ExecutionStatus.FAILED);
  expect(result.getError().errorMessage).toContain("intentional failure");
});

from aws_durable_execution_sdk_python import DurableContext, durable_execution
from aws_durable_execution_sdk_python.execution import InvocationStatus
from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner


@durable_execution
def handler(event: dict, context: DurableContext):
    if event.get("fail"):
        raise ValueError("intentional failure")
    return "ok"


def test_reports_failed_execution():
    runner = DurableFunctionTestRunner(handler=handler)
    with runner:
        result = runner.run(input='{"fail": true}', timeout=10)

    assert result.status is InvocationStatus.FAILED
    assert result.error is not None

import static org.junit.jupiter.api.Assertions.*;

import org.junit.jupiter.api.Test;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.model.ExecutionStatus;
import software.amazon.lambda.durable.testing.LocalDurableTestRunner;

class TestFailureTest {

    record Input(boolean fail) {}

    @Test
    void reportsFailedExecution() {
        var runner = LocalDurableTestRunner.create(
            Input.class,
            (input, context) -> {
                if (input.fail()) {
                    throw new RuntimeException("intentional failure");
                }
                return "ok";
            }
        );

        var result = runner.runUntilComplete(new Input(true));

        assertEquals(ExecutionStatus.FAILED, result.getStatus());
        assertTrue(result.getError().isPresent());
    }
}

// Save this file as test_failure_test.go and run go test.
package main

import (
    "errors"
    "strings"
    "testing"

    "github.com/aws/aws-durable-execution-sdk-go/durable"
    "github.com/aws/aws-durable-execution-sdk-go/durable/durabletest"
)

type failInput struct {
    Fail bool `json:"fail"`
}

func failHandler(ctx durable.Context, event failInput) (string, error) {
    if event.Fail {
        return "", errors.New("intentional failure")
    }
    return "ok", nil
}

func TestFailure(t *testing.T) {
    runner := durabletest.NewLocalRunner(failHandler)

    result, err := runner.RunUntilComplete(failInput{Fail: true})
    if err != nil {
        t.Fatal(err)
    }
    if result.Status != durabletest.Failed {
        t.Fatalf("status = %s, want FAILED\n%s", result.Status, result.FormatTree())
    }
    if result.Error == nil {
        t.Fatal("result.Error is nil, want an error")
    }
    if !strings.Contains(result.Error.Message, "intentional failure") {
        t.Errorf("error message = %q, want it to contain %q", result.Error.Message, "intentional failure")
    }
}

using Amazon.Lambda.DurableExecution;
using Amazon.Lambda.DurableExecution.Testing;
using Xunit;

public class TestFailureTest
{
    private static async Task<string> Workflow(FailureInput input, IDurableContext ctx)
    {
        if (input.Fail)
        {
            throw new InvalidOperationException("intentional failure");
        }
        return await Task.FromResult("ok");
    }

    [Fact]
    public async Task ReportsFailedExecution()
    {
        await using var runner = new DurableTestRunner<FailureInput, string>(
            Workflow,
            new TestRunnerOptions { SkipTime = true });

        TestResult<string> result = await runner.RunAsync(new FailureInput(Fail: true));

        Assert.True(result.IsFailed);
        Assert.NotNull(result.Error);
    }
}

public record FailureInput(bool Fail);

Test retries¶

The test runner drives the full retry loop via replay. Configure a retry strategy on the step, and the runner re-invokes the handler as many times as needed.

TypeScriptPythonJavaGoC#

import { withDurableExecution, DurableContext, createRetryStrategy } from "@aws/durable-execution-sdk-js";
import { LocalDurableTestRunner } from "@aws/durable-execution-sdk-js-testing";
import { ExecutionStatus } from "@aws-sdk/client-lambda";

let attempts = 0;

const handler = withDurableExecution(async (event: unknown, context: DurableContext) => {
  return await context.step("flaky", () => {
    attempts++;
    if (attempts < 3) {
      throw new Error("transient error");
    }
    return "done";
  }, {
    retryStrategy: createRetryStrategy({ maxAttempts: 3 }),
  });
});

let runner: LocalDurableTestRunner;

beforeAll(async () => {
  await LocalDurableTestRunner.setupTestEnvironment({ skipTime: true });
});

afterAll(async () => {
  await LocalDurableTestRunner.teardownTestEnvironment();
});

beforeEach(() => {
  attempts = 0;
  runner = new LocalDurableTestRunner({ handlerFunction: handler });
});

it("retries and eventually succeeds", async () => {
  const result = await runner.run();

  expect(result.getStatus()).toBe(ExecutionStatus.SUCCEEDED);
  expect(result.getResult()).toBe("done");
});

from aws_durable_execution_sdk_python import DurableContext, durable_execution, durable_step
from aws_durable_execution_sdk_python.types import StepContext
from aws_durable_execution_sdk_python.config import StepConfig
from aws_durable_execution_sdk_python.retries import create_retry_strategy, RetryStrategyConfig
from aws_durable_execution_sdk_python.execution import InvocationStatus
from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner

attempts = 0


@durable_step
def flaky(ctx: StepContext) -> str:
    global attempts
    attempts += 1
    if attempts < 3:
        raise RuntimeError("transient error")
    return "done"


@durable_execution
def handler(event, context: DurableContext) -> str:
    config = StepConfig(
        retry_strategy=create_retry_strategy(RetryStrategyConfig(max_attempts=3))
    )
    return context.step(flaky(), config=config)


def test_retries_and_eventually_succeeds():
    global attempts
    attempts = 0

    runner = DurableFunctionTestRunner(handler=handler)
    with runner:
        result = runner.run(timeout=30)

    assert result.status is InvocationStatus.SUCCEEDED
    assert result.result == '"done"'

import static org.junit.jupiter.api.Assertions.*;

import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.services.lambda.model.OperationStatus;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.config.StepConfig;
import software.amazon.lambda.durable.model.ExecutionStatus;
import software.amazon.lambda.durable.retry.RetryStrategies;
import software.amazon.lambda.durable.testing.LocalDurableTestRunner;

class TestRetriesTest {

    @Test
    void retriesAndEventuallySucceeds() {
        var attempts = new AtomicInteger(0);

        var config = StepConfig.builder()
            .retryStrategy(RetryStrategies.exponentialBackoff(3))
            .build();

        var runner = LocalDurableTestRunner.create(
            Void.class,
            (input, context) -> context.step("flaky", String.class, ctx -> {
                if (attempts.incrementAndGet() < 3) {
                    throw new RuntimeException("transient error");
                }
                return "done";
            }, config)
        );

        var result = runner.runUntilComplete(null);

        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
        assertEquals("done", result.getResult(String.class));
    }
}

// Save this file as test_retries_test.go and run go test.
package main

import (
    "errors"
    "testing"

    "github.com/aws/aws-durable-execution-sdk-go/durable"
    "github.com/aws/aws-durable-execution-sdk-go/durable/durabletest"
)

func retryHandler(ctx durable.Context, _ any) (string, error) {
    return durable.Step[string](ctx, "flaky", func(sc durable.StepContext) (string, error) {
        if sc.Attempt() < 3 {
            return "", errors.New("transient error")
        }
        return "done", nil
    }, durable.WithRetry(durable.MustNewRetryStrategy(durable.RetryConfig{MaxAttempts: 3})))
}

func TestRetries(t *testing.T) {
    runner := durabletest.NewLocalRunner(retryHandler)

    result, err := runner.RunUntilComplete(nil)
    if err != nil {
        t.Fatal(err)
    }
    if result.Status != durabletest.Succeeded {
        t.Fatalf("status = %s, want SUCCEEDED\n%s", result.Status, result.FormatTree())
    }

    got, err := durabletest.ResultAs[string](result)
    if err != nil {
        t.Fatal(err)
    }
    if got != "done" {
        t.Errorf("result = %q, want done", got)
    }
}

using Amazon.Lambda.DurableExecution;
using Amazon.Lambda.DurableExecution.Testing;
using Xunit;

public class TestRetriesTest
{
    [Fact]
    public async Task RetriesAndEventuallySucceeds()
    {
        int attempts = 0;

        Task<string> Workflow(object? input, IDurableContext ctx)
        {
            var config = new StepConfig
            {
                RetryStrategy = RetryStrategy.Exponential(
                    maxAttempts: 3,
                    initialDelay: TimeSpan.FromMilliseconds(1)),
            };

            return ctx.StepAsync(
                async (_, _) =>
                {
                    if (++attempts < 3)
                    {
                        throw new InvalidOperationException("transient error");
                    }
                    return "done";
                },
                name: "flaky",
                config: config);
        }

        await using var runner = new DurableTestRunner<object?, string>(
            Workflow,
            new TestRunnerOptions { SkipTime = true });

        TestResult<string> result = await runner.RunAsync(null);

        result.EnsureSucceeded();
        Assert.Equal("done", result.Result);
        Assert.Equal(3, result.GetStep("flaky").Attempt);
    }
}

Skip time in tests¶

Retry backoffs and context.wait() durations use real time in production. The test runner collapses these delays so tests finish in milliseconds.

TypeScriptPythonJavaGoC#

Pass { skipTime: true } to setupTestEnvironment(). The runner swaps its real timer for a queue that fires immediately. Both context.wait() delays and step retry delays complete in zero wall-clock time.

await LocalDurableTestRunner.setupTestEnvironment({ skipTime: true });

Pass skip_time=True to DurableFunctionTestRunner (the default). The runner uses a virtual clock that advances instantly through modeled delays. Both context.wait() durations and step retry delays complete in zero wall-clock time.

runner = DurableFunctionTestRunner(handler=handler, skip_time=True)

Set skip_time=False to assert on real wait durations.

runUntilComplete() calls advanceTime() after each invocation. advanceTime() immediately marks PENDING step retries as READY and completes STARTED waits without real-time sleeps. If you call run() directly, call advanceTime() yourself between invocations.

runner.runUntilComplete(input); // advanceTime() runs after each invocation

The local runner uses a virtual clock that is always on. RunUntilComplete fires wait and retry timers on that clock between invocations. If you call Run directly, call CompletePendingTimers yourself between invocations.

result, err := runner.RunUntilComplete(input) // fires timers between invocations

TestRunnerOptions.SkipTime defaults to true, so RunAsync() completes both WaitAsync timers and step retry backoffs immediately without real-time sleeps. Set it to false to assert on real wait durations.

await using var runner = new DurableTestRunner<TIn, TOut>(
    Workflow, new TestRunnerOptions { SkipTime = true });
await runner.RunAsync(input); // waits and retry backoffs complete instantly

See Workflow patterns: Long waits for a worked example.

Test branching logic¶

Run the same handler with different inputs to cover each branch. Each test case gets its own runner instance.

TypeScriptPythonJavaGoC#

import { withDurableExecution, DurableContext } from "@aws/durable-execution-sdk-js";
import { LocalDurableTestRunner } from "@aws/durable-execution-sdk-js-testing";
import { ExecutionStatus } from "@aws-sdk/client-lambda";

const handler = withDurableExecution(async (event: { premium: boolean }, context: DurableContext) => {
  if (event.premium) {
    return await context.step("premium-path", () => "premium");
  }
  return await context.step("standard-path", () => "standard");
});

let runner: LocalDurableTestRunner;

beforeAll(async () => {
  await LocalDurableTestRunner.setupTestEnvironment();
});

afterAll(async () => {
  await LocalDurableTestRunner.teardownTestEnvironment();
});

beforeEach(() => {
  runner = new LocalDurableTestRunner({ handlerFunction: handler });
});

it("takes the premium path", async () => {
  const result = await runner.run({ payload: { premium: true } });

  expect(result.getStatus()).toBe(ExecutionStatus.SUCCEEDED);
  expect(result.getResult()).toBe("premium");
});

it("takes the standard path", async () => {
  const result = await runner.run({ payload: { premium: false } });

  expect(result.getStatus()).toBe(ExecutionStatus.SUCCEEDED);
  expect(result.getResult()).toBe("standard");
});

from aws_durable_execution_sdk_python import DurableContext, durable_execution, durable_step
from aws_durable_execution_sdk_python.types import StepContext
from aws_durable_execution_sdk_python.execution import InvocationStatus
from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner


@durable_step
def premium_path(ctx: StepContext) -> str:
    return "premium"


@durable_step
def standard_path(ctx: StepContext) -> str:
    return "standard"


@durable_execution
def handler(event: dict, context: DurableContext) -> str:
    if event.get("premium"):
        return context.step(premium_path())
    return context.step(standard_path())


def test_takes_premium_path():
    runner = DurableFunctionTestRunner(handler=handler)
    with runner:
        result = runner.run(input='{"premium": true}', timeout=10)

    assert result.status is InvocationStatus.SUCCEEDED
    assert result.result == '"premium"'


def test_takes_standard_path():
    runner = DurableFunctionTestRunner(handler=handler)
    with runner:
        result = runner.run(input='{"premium": false}', timeout=10)

    assert result.status is InvocationStatus.SUCCEEDED
    assert result.result == '"standard"'

import static org.junit.jupiter.api.Assertions.*;

import org.junit.jupiter.api.Test;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.model.ExecutionStatus;
import software.amazon.lambda.durable.testing.LocalDurableTestRunner;

class TestBranchingTest {

    record Input(boolean premium) {}

    private LocalDurableTestRunner<Input, String> createRunner() {
        return LocalDurableTestRunner.create(
            Input.class,
            (input, context) -> {
                if (input.premium()) {
                    return context.step("premium-path", String.class, ctx -> "premium");
                }
                return context.step("standard-path", String.class, ctx -> "standard");
            }
        );
    }

    @Test
    void takesPremiumPath() {
        var result = createRunner().runUntilComplete(new Input(true));

        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
        assertEquals("premium", result.getResult(String.class));
    }

    @Test
    void takesStandardPath() {
        var result = createRunner().runUntilComplete(new Input(false));

        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
        assertEquals("standard", result.getResult(String.class));
    }
}

// Save this file as test_branching_test.go and run go test.
package main

import (
    "testing"

    "github.com/aws/aws-durable-execution-sdk-go/durable"
    "github.com/aws/aws-durable-execution-sdk-go/durable/durabletest"
)

type branchInput struct {
    Premium bool `json:"premium"`
}

func branchHandler(ctx durable.Context, event branchInput) (string, error) {
    if event.Premium {
        return durable.Step[string](ctx, "premium-path", func(durable.StepContext) (string, error) {
            return "premium", nil
        })
    }
    return durable.Step[string](ctx, "standard-path", func(durable.StepContext) (string, error) {
        return "standard", nil
    })
}

func TestBranching(t *testing.T) {
    cases := []struct {
        name  string
        input branchInput
        want  string
    }{
        {"premium path", branchInput{Premium: true}, "premium"},
        {"standard path", branchInput{Premium: false}, "standard"},
    }

    for _, tc := range cases {
        t.Run(tc.name, func(t *testing.T) {
            runner := durabletest.NewLocalRunner(branchHandler)

            result, err := runner.RunUntilComplete(tc.input)
            if err != nil {
                t.Fatal(err)
            }
            if result.Status != durabletest.Succeeded {
                t.Fatalf("status = %s, want SUCCEEDED\n%s", result.Status, result.FormatTree())
            }

            got, err := durabletest.ResultAs[string](result)
            if err != nil {
                t.Fatal(err)
            }
            if got != tc.want {
                t.Errorf("result = %q, want %q", got, tc.want)
            }
        })
    }
}

using Amazon.Lambda.DurableExecution;
using Amazon.Lambda.DurableExecution.Testing;
using Xunit;

public class TestBranchingTest
{
    private static Task<string> Workflow(BranchInput input, IDurableContext ctx)
    {
        if (input.Premium)
        {
            return ctx.StepAsync(async (_, _) => "premium", name: "premium-path");
        }
        return ctx.StepAsync(async (_, _) => "standard", name: "standard-path");
    }

    private static DurableTestRunner<BranchInput, string> CreateRunner()
        => new(Workflow, new TestRunnerOptions { SkipTime = true });

    [Fact]
    public async Task TakesPremiumPath()
    {
        await using var runner = CreateRunner();

        TestResult<string> result = await runner.RunAsync(new BranchInput(Premium: true));

        result.EnsureSucceeded();
        Assert.Equal("premium", result.Result);
    }

    [Fact]
    public async Task TakesStandardPath()
    {
        await using var runner = CreateRunner();

        TestResult<string> result = await runner.RunAsync(new BranchInput(Premium: false));

        result.EnsureSucceeded();
        Assert.Equal("standard", result.Result);
    }
}

public record BranchInput(bool Premium);

Authoring¶

Install the testing SDK¶

Write a minimal test¶

Test a failed execution¶

Test retries¶

Skip time in tests¶

Test branching logic¶

See also¶