AWS Lambda Durable Functions

import {
  LocalDurableTestRunner,
  OperationType,
  OperationStatus
} from '@aws/durable-execution-sdk-js-testing';

describe('My Durable Function', () => {
  beforeAll(() => 
    LocalDurableTestRunner.setupTestEnvironment({ skipTime: true })
  );
  
  afterAll(() => 
    LocalDurableTestRunner.teardownTestEnvironment()
  );

  it('should execute workflow', async () => {
    const runner = new LocalDurableTestRunner({ 
      handlerFunction: handler 
    });
    
    const execution = await runner.run({ 
      payload: { userId: '123' } 
    });

    expect(execution.getStatus()).toBe('SUCCEEDED');
    expect(execution.getResult()).toEqual({ success: true });
  });
});

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

from aws_durable_execution_sdk_python_testing import DurableFunctionTestRunner
from aws_durable_execution_sdk_python.execution import InvocationStatus
from src.my_function import handler

def test_workflow():
    """Test durable function locally."""
    runner = DurableFunctionTestRunner(handler=handler)
    
    with runner:
        result = runner.run(input={'user_id': '123'}, timeout=10)

    assert result.status is InvocationStatus.SUCCEEDED

it('should execute steps in order', async () => {
  const runner = new LocalDurableTestRunner({ handlerFunction: handler });
  await runner.run({ payload: { test: true } });

  // ✅ CORRECT: Get by name
  const fetchStep = runner.getOperation('fetch-user');
  expect(fetchStep.getType()).toBe(OperationType.STEP);
  expect(fetchStep.getStatus()).toBe(OperationStatus.SUCCEEDED);

  const processStep = runner.getOperation('process-data');
  expect(processStep.getStatus()).toBe(OperationStatus.SUCCEEDED);

  // ❌ WRONG: Get by index (brittle, breaks easily)
  // const step1 = runner.getOperationByIndex(0);
});

from aws_durable_execution_sdk_python.lambda_service import OperationType
def test_steps_execute():
    """Test step execution."""
    runner = DurableFunctionTestRunner(handler=handler)
    
    with runner:
        result = runner.run(input={'test': True}, timeout=10)

    # ✅ CORRECT: Get step by name
    fetch_step = result.get_step('fetch-user')
    assert fetch_step is not None

    # ✅ Also valid: filter result.operations by type
    step_names = {op.name for op in result.operations if op.operation_type == OperationType.STEP}
    assert step_names >= {'fetch-user', 'process-data'}
    assert 'process-data' in step_names

it('should handle replay correctly', async () => {
  const runner = new LocalDurableTestRunner({ handlerFunction: handler });
  
  // First execution
  const execution1 = await runner.run({ payload: { value: 42 } });
  expect(execution1.getStatus()).toBe('SUCCEEDED');

  // Simulate replay
  const execution2 = await runner.run({ payload: { value: 42 } });
  expect(execution2.getStatus()).toBe('SUCCEEDED');
  
  // Results should be identical
  expect(execution1.getResult()).toEqual(execution2.getResult());
});

it('should wait for specified duration', async () => {
  const runner = new LocalDurableTestRunner({ 
    handlerFunction: handler 
  });

  const executionPromise = runner.run({ payload: {} });

  // Advance time by 60 seconds
  await runner.skipTime({ seconds: 60 });

  const execution = await executionPromise;
  expect(execution.getStatus()).toBe('SUCCEEDED');

  const waitOp = runner.getOperation('delay');
  expect(waitOp.getType()).toBe(OperationType.WAIT);
  expect(waitOp.getWaitDetails()?.waitSeconds).toBe(60);
});

it('should use correct test runner API', async () => {
  const runner = new LocalDurableTestRunner({
    handlerFunction: handler,
  });

  // ✅ CORRECT: Wrap event in payload
  const execution = await runner.run({
    payload: { name: 'Alice', userId: '123' }
  });

  // ✅ CORRECT: Type cast result
  const result = execution.getResult() as {
    greeting: string;
    message: string;
  };

  expect(result.greeting).toBe('Hello, Alice!');

  // ✅ CORRECT: Get operations by name
  const greetingStep = runner.getOperation('generate-greeting');
  expect(greetingStep.getStepDetails()?.result).toBe('Hello, Alice!');
});

// ❌ WRONG: Missing payload wrapper and type casting
it('incorrect api usage', async () => {
  const runner = new LocalDurableTestRunner({ handlerFunction: handler });

  // ❌ Missing payload wrapper
  const execution = await runner.run({ name: 'Alice' });

  // ❌ No type casting - result is 'unknown'
  const result = execution.getResult();
  // expect(result.greeting).toBe('...'); // Type error!
});

import { WaitingOperationStatus } from '@aws/durable-execution-sdk-js-testing';

it('should handle callback success', async () => {
  const runner = new LocalDurableTestRunner({ handlerFunction: handler });

  // Start execution (will pause at callback)
  const executionPromise = runner.run({
    payload: { approver: '[email protected]' }
  });

  // ✅ CRITICAL: Get operation by NAME
  const callbackOp = runner.getOperation('wait-for-approval');

  // ✅ CRITICAL: Wait for operation to reach STARTED status
  await callbackOp.waitForData(WaitingOperationStatus.STARTED);

  // ✅ CRITICAL: Must JSON.stringify callback data!
  await callbackOp.sendCallbackSuccess(
    JSON.stringify({ approved: true, comments: 'Looks good' })
  );

  const execution = await executionPromise;
  expect(execution.getStatus()).toBe('SUCCEEDED');

  // ✅ CRITICAL: Parse JSON string result
  const result: any = execution.getResult();
  const approval = typeof result.approval === 'string'
    ? JSON.parse(result.approval)
    : result.approval;

  expect(approval.approved).toBe(true);
  expect(approval.comments).toBe('Looks good');
});

it('should handle callback failure', async () => {
  const runner = new LocalDurableTestRunner({ handlerFunction: handler });

  const executionPromise = runner.run({ payload: {} });

  await new Promise(resolve => setTimeout(resolve, 100));

  const callbackOp = runner.getOperation('wait-for-approval');
  
  // Send callback failure
  await callbackOp.sendCallbackFailure(
    'ApprovalDenied',
    'Request was rejected'
  );

  const execution = await executionPromise;
  expect(execution.getStatus()).toBe('FAILED');
});

def test_callback_creation():
    """Test that callback is created correctly."""
    runner = DurableFunctionTestRunner(handler=handler)
    
    with runner:
        result = runner.run(input={'approver': '[email]'}, timeout=10)

    # Find callback operations in the result
    callback_ops = [
        op for op in result.operations
        if op.operation_type == OperationType.CALLBACK
    ]
    assert len(callback_ops) == 1
    assert callback_ops[0].name == 'wait-for-approval'
    assert callback_ops[0].callback_id is not None

it('should handle callback heartbeats', async () => {
  const runner = new LocalDurableTestRunner({ handlerFunction: handler });

  const executionPromise = runner.run({ payload: {} });

  await new Promise(resolve => setTimeout(resolve, 100));

  const callbackOp = runner.getOperation('long-running-process');
  
  // Send heartbeats
  await callbackOp.sendCallbackHeartbeat();
  await runner.skipTime({ minutes: 2 });
  await callbackOp.sendCallbackHeartbeat();
  await runner.skipTime({ minutes: 2 });
  
  // Complete callback
  await callbackOp.sendCallbackSuccess(JSON.stringify({ status: 'completed' }));

  const execution = await executionPromise;
  expect(execution.getStatus()).toBe('SUCCEEDED');
});

it('should retry on failure', async () => {
  let attemptCount = 0;
  
  const testHandler = withDurableExecution(async (event, context: DurableContext) => {
    return await context.step('flaky-operation', async () => {
      attemptCount++;
      if (attemptCount < 3) {
        throw new Error('Temporary failure');
      }
      return { success: true };
    });
  });

  const runner = new LocalDurableTestRunner({ handlerFunction: testHandler });
  const execution = await runner.run({ payload: {} });

  expect(execution.getStatus()).toBe('SUCCEEDED');
  expect(attemptCount).toBe(3);

  const step = runner.getOperation('flaky-operation');
  expect(step.getStatus()).toBe(OperationStatus.SUCCEEDED);
});

it('should fail after max retries', async () => {
  const testHandler = withDurableExecution(async (event, context: DurableContext) => {
    return await context.step(
      'always-fails',
      async () => {
        throw new Error('Permanent failure');
      },
      {
        retryStrategy: createRetryStrategy({ maxAttempts: 3 })
      }
    );
  });

  const runner = new LocalDurableTestRunner({ handlerFunction: testHandler });
  const execution = await runner.run({ payload: {} });

  expect(execution.getStatus()).toBe('FAILED');
  expect(execution.getError()?.errorMessage).toContain('Permanent failure');
});

it('should process items concurrently', async () => {
  const runner = new LocalDurableTestRunner({ handlerFunction: handler });
  
  const execution = await runner.run({ 
    payload: { items: [1, 2, 3, 4, 5] } 
  });

  expect(execution.getStatus()).toBe('SUCCEEDED');

  const mapOp = runner.getOperation('process-items');
  expect(mapOp.getType()).toBe(OperationType.MAP);

  // Check individual item operations
  const item0 = runner.getOperation('process-0');
  expect(item0.getStatus()).toBe(OperationStatus.SUCCEEDED);
});

import { CloudDurableTestRunner } from '@aws/durable-execution-sdk-js-testing';

describe('Integration Tests', () => {
  it('should execute in real Lambda', async () => {
    const runner = new CloudDurableTestRunner({
      functionName: 'my-durable-function:1',  // Qualified ARN required
      client: new LambdaClient({ region: 'us-east-1' })
    });

    const execution = await runner.run({
      payload: { userId: '123' },
      config: { pollInterval: 1000 }
    });

    expect(execution.getStatus()).toBe('SUCCEEDED');
    
    const step = runner.getOperation('fetch-user');
    expect(step.getStatus()).toBe(OperationStatus.SUCCEEDED);
  });
});

# Set environment variables for cloud mode
export AWS_REGION=us-west-2
export QUALIFIED_FUNCTION_NAME="my-durable-function:$LATEST"
export LAMBDA_FUNCTION_TEST_NAME="my_function"

# Run in cloud mode
pytest --runner-mode=cloud -k test_workflow

def test_workflow_cloud():
    """Test against deployed Lambda function."""
    runner = DurableFunctionCloudTestRunner(
        function_name='my-function:$LATEST',
        region='us-west-2'
    )
    
    with runner:
        result = runner.run(input={'user_id': '123'}, timeout=60)

    assert result.status is InvocationStatus.SUCCEEDED

it('should validate operation details', async () => {
  const runner = new LocalDurableTestRunner({ handlerFunction: handler });
  await runner.run({ payload: {} });

  const step = runner.getOperation('process-data');
  
  // Check operation type
  expect(step.getType()).toBe(OperationType.STEP);
  
  // Check status
  expect(step.getStatus()).toBe(OperationStatus.SUCCEEDED);
  
  // Check timing
  expect(step.getStartTimestamp()).toBeDefined();
  expect(step.getEndTimestamp()).toBeDefined();
  
  // Check result
  const stepDetails = step.getStepDetails();
  expect(stepDetails?.result).toEqual({ processed: true });
});

// Brittle - breaks when operations change
const step = runner.getOperationByIndex(0);

// Robust - works even if operation order changes
const step = runner.getOperation('fetch-user');

// Race condition - callback might not exist yet
const callbackOp = runner.getOperation('wait-approval');
await callbackOp.sendCallbackSuccess('{}');

// Use waitForData with proper status
import { WaitingOperationStatus } from '@aws/durable-execution-sdk-js-testing';

const callbackOp = runner.getOperation('wait-approval');
await callbackOp.waitForData(WaitingOperationStatus.STARTED);
await callbackOp.sendCallbackSuccess(JSON.stringify({}));

module.exports = {
  preset: 'ts-jest',
  testEnvironment: 'node',
  roots: ['<rootDir>/src'],
  testMatch: ['**/*.test.ts'],
  transform: {
    '^.+\\.ts$': 'ts-jest',
  },
  collectCoverageFrom: [
    'src/**/*.ts',
    '!src/**/*.d.ts',
  ],
};