Authoring¶

Install the testing SDK¶

TypeScriptPythonJavaGoRustC#

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

The local runner is in the SDK crate, in the test_util module behind the test-util feature. Add the SDK as a dev-dependency with that feature, from the same Git source as the runtime dependency, so the feature is on only in tests. Add the Tokio features that #[tokio::test] needs:

[dev-dependencies]
aws-durable-execution-sdk = { git = "https://github.com/aws/aws-durable-execution-sdk-rust", features = ["test-util"] }
tokio = { version = "1", features = ["macros", "rt"] }

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.

TypeScriptPythonJavaGoRustC#

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

Create a runner with LocalRunner::new(), pass the handler and the event to run, and await the returned handle to get a TestResult. Mark each test with #[tokio::test]. Each run starts a fresh execution, so one runner can drive several runs.

use aws_durable_execution_sdk as durable;
use durable::test_util::{ExecutionStatus, LocalRunner};

async fn handler(
    _event: serde_json::Value,
    ctx: durable::DurableContext,
) -> Result<String, durable::BoxError> {
    let result = ctx
        .step(|_| async { Ok("hello".to_owned()) })
        .name("greet")
        .await?;
    Ok(result)
}

#[tokio::test]
async fn returns_the_expected_result() -> Result<(), durable::BoxError> {
    let runner = LocalRunner::new();
    let result = runner.run(handler, serde_json::json!({})).await?;

    assert_eq!(result.status(), ExecutionStatus::Succeeded);
    assert_eq!(result.output().map(String::as_str), Some("hello"));
    Ok(())
}

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.

TypeScriptPythonJavaGoRustC#

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")
    }
}

use aws_durable_execution_sdk as durable;
use durable::test_util::{ExecutionStatus, LocalRunner};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
struct Event {
    fail: bool,
}

async fn handler(event: Event, _ctx: durable::DurableContext) -> Result<String, durable::BoxError> {
    if event.fail {
        return Err("intentional failure".into());
    }
    Ok("ok".to_owned())
}

#[tokio::test]
async fn reports_a_failed_execution() -> Result<(), durable::BoxError> {
    let runner = LocalRunner::new();
    let result = runner.run(handler, Event { fail: true }).await?;

    assert_eq!(result.status(), ExecutionStatus::Failed);
    let message = result.error_message().ok_or("no error message")?;
    assert!(message.contains("intentional failure"));
    Ok(())
}

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.

TypeScriptPythonJavaGoRustC#

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

use std::sync::atomic::{AtomicU32, Ordering};

use aws_durable_execution_sdk as durable;
use durable::builders::RetryStrategyConfig;
use durable::test_util::{ExecutionStatus, LocalRunner};

static ATTEMPTS: AtomicU32 = AtomicU32::new(0);

async fn handler(
    _event: serde_json::Value,
    ctx: durable::DurableContext,
) -> Result<String, durable::BoxError> {
    Ok(ctx
        .step(|_| async {
            if ATTEMPTS.fetch_add(1, Ordering::SeqCst) + 1 < 3 {
                return Err("transient error".into());
            }
            Ok("done".to_owned())
        })
        .name("flaky")
        .retry_strategy_config(RetryStrategyConfig::builder().max_attempts(3).build())
        .await?)
}

#[tokio::test]
async fn retries_and_eventually_succeeds() -> Result<(), durable::BoxError> {
    ATTEMPTS.store(0, Ordering::SeqCst);
    let runner = LocalRunner::new();
    let result = runner.run(handler, serde_json::json!({})).await?;

    assert_eq!(result.status(), ExecutionStatus::Succeeded);
    assert_eq!(result.output().map(String::as_str), Some("done"));
    Ok(())
}

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.

TypeScriptPythonJavaGoRustC#

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

The local runner always uses a virtual clock in whole seconds. Between invocations it moves the clock to the end of the earliest pending wait or retry delay. Both ctx.wait() durations and step retry delays complete in zero wall-clock time.

let result = runner.run(handler, event).await?; // waits and retries fire on a virtual clock

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.

TypeScriptPythonJavaGoRustC#

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

use aws_durable_execution_sdk as durable;
use durable::test_util::{ExecutionStatus, LocalRunner};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
struct Event {
    premium: bool,
}

async fn handler(event: Event, ctx: durable::DurableContext) -> Result<String, durable::BoxError> {
    if event.premium {
        return Ok(ctx
            .step(|_| async { Ok("premium".to_owned()) })
            .name("premium-path")
            .await?);
    }
    Ok(ctx
        .step(|_| async { Ok("standard".to_owned()) })
        .name("standard-path")
        .await?)
}

#[tokio::test]
async fn takes_the_premium_path() -> Result<(), durable::BoxError> {
    let runner = LocalRunner::new();
    let result = runner.run(handler, Event { premium: true }).await?;

    assert_eq!(result.status(), ExecutionStatus::Succeeded);
    assert_eq!(result.output().map(String::as_str), Some("premium"));
    Ok(())
}

#[tokio::test]
async fn takes_the_standard_path() -> Result<(), durable::BoxError> {
    let runner = LocalRunner::new();
    let result = runner.run(handler, Event { premium: false }).await?;

    assert_eq!(result.status(), ExecutionStatus::Succeeded);
    assert_eq!(result.output().map(String::as_str), Some("standard"));
    Ok(())
}

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¶