在 TypeScript 中测试 AWS Lambda Durable Functions

Source: Dev.to

测试库

SDK 为不同场景提供了测试工具：

• LocalDurableTestRunner 在进程内运行函数，并使用模拟的检查点服务器。即使是通常需要数小时的工作流，测试也能在毫秒级完成。这是大多数测试会使用的方式。
• CloudDurableTestRunner 对部署在 AWS 上的 Lambda 函数进行测试。用于集成测试或需要在真实 AWS 环境中验证行为的情况。
• run-durable CLI 提供无需编写测试代码的快速命令行测试。非常适合在开发期间进行快速迭代和调试。

我们将重点关注本地测试，因为这将是你花费最多时间的地方。

第一个测试

先从一个简单的 durable 函数开始：

// order-processor.ts
import { DurableContext, withDurableExecution } from '@aws/durable-execution-sdk-js';

export const handler = withDurableExecution(
  async (event: any, context: DurableContext) => {
    const order = await context.step('create-order', async () => {
      return { orderId: '123', total: 50 };
    });

    await context.wait({ seconds: 300 }); // 等待 5 分钟

    const notification = await context.step('send-notification', async () => {
      return { sent: true };
    });

    return { order, notification };
  }
);

下面演示如何对它进行测试：

// order-processor.test.ts
import { LocalDurableTestRunner } from '@aws/durable-execution-sdk-js-testing';
import { handler } from './order-processor';

describe('Order Processor', () => {
  beforeAll(async () => {
    await LocalDurableTestRunner.setupTestEnvironment({ skipTime: true });
  });

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

  it('should process order successfully', async () => {
    const runner = new LocalDurableTestRunner({
      handlerFunction: handler,
    });

    const execution = await runner.run();

    expect(execution.getStatus()).toBe('SUCCEEDED');
    expect(execution.getResult()).toEqual({
      order: { orderId: '123', total: 50 },
      notification: { sent: true }
    });
  });
});

就是这么简单。即使函数中有 5 分钟的等待，测试也会在毫秒级完成。skipTime: true 选项告诉测试运行器即时跳过基于时间的操作。

环境的设置与清理

每个测试套件都需要进行设置和清理：

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

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

• setupTestEnvironment() 启动本地检查点服务器，并可选地安装用于时间跳过的假定时器。
• teardownTestEnvironment() 清理所有资源。

在每个测试文件的 beforeAll 与 afterAll 中各调用一次。skipTime 选项对快速测试至关重要。启用后，context.wait()、setTimeout 以及重试延迟等操作会立即完成。若不启用，测试将真的等待指定的时间。

检查操作

测试库的真正威力在于可以检查函数到底执行了哪些操作：

it('should execute operations in correct order', async () => {
  const runner = new LocalDurableTestRunner({
    handlerFunction: handler,
  });

  const execution = await runner.run();

  // 获取所有操作
  const operations = execution.getOperations();
  expect(operations).toHaveLength(3); // create-order、wait、send-notification

  // 通过索引获取特定操作
  const createOrder = runner.getOperationByIndex(0);
  expect(createOrder.getType()).toBe('STEP');
  expect(createOrder.getStatus()).toBe('SUCCEEDED');
  expect(createOrder.getStepDetails()?.result).toEqual({
    orderId: '123',
    total: 50
  });

  // 获取等待操作
  const waitOp = runner.getOperationByIndex(1);
  expect(waitOp.getType()).toBe('WAIT');
  expect(waitOp.getWaitDetails()?.waitSeconds).toBe(300);

  // 获取通知操作
  const notification = runner.getOperationByIndex(2);
  expect(notification.getStepDetails()?.result).toEqual({ sent: true });
});

你可以通过索引、名称或 ID 来访问操作。每个操作都会公开其类型、状态以及特定类型的细节（如步骤结果或等待时长）。

测试失败与重试

真实的工作流会出现错误。下面演示如何测试重试行为：

// payment-processor.ts
import { DurableContext, withDurableExecution, createRetryStrategy } from '@aws/durable-execution-sdk-js';

export const handler = withDurableExecution(
  async (event: any, context: DurableContext) => {
    const payment = await context.step('process-payment', async () => {
      const response = await fetch('https://api.payments.com/charge', {
        method: 'POST',
        body: JSON.stringify({ amount: event.amount })
      });
      if (!response.ok) throw new Error('Payment failed');
      return response.json();
    }, {
      retryStrategy: createRetryStrategy({
        maxAttempts: 3,
        backoffRate: 2,
        initialInterval: 1000
      })
    });

    return payment;
  }
);

使用 mock 来测试：

// payment-processor.test.ts
import { LocalDurableTestRunner } from '@aws/durable-execution-sdk-js-testing';
import { handler } from './payment-processor';

// 保存原始 fetch
const originalFetch = global.fetch;

describe('Payment Processor', () => {
  beforeAll(async () => {
    await LocalDurableTestRunner.setupTestEnvironment({ skipTime: true });
  });

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

  beforeEach(() => {
    // 仅对外部 API 进行 fetch mock
    global.fetch = jest.fn((url: string | URL | Request, ...args) => {
      const urlString = url.toString();
      // 让检查点服务器的调用直接通过
      if (urlString.includes('127.0.0.1') || urlString.includes('localhost')) {
        return originalFetch(url as any, ...args);
      }
      // Mock 外部 API 调用
      return Promise.reject(new Error('Unmocked fetch call'));
    }) as any;
  });

  afterEach(() => {
    global.fetch = originalFetch;
  });

  it('should succeed on first attempt', async () => {
    (global.fetch as jest.Mock).mockImplementation((url: string | URL | Request, ...args) => {
      const urlString = url.toString();
      if (urlString.includes('127.0.0.1') || urlString.includes('localhost')) {
        return originalFetch(url as any, ...args);
      }
      if (urlString.includes('api.payments.com')) {
        return Promise.resolve({
          ok: true,
          json: async () => ({ transactionId: 'txn-123', status: 'success' })
        });
      }
      return Promise.reject(new Error('Unmocked fetch'));
    });

    const runner = new LocalDurableTestRunner({
      handlerFunction: handler,
    });

    const execution = await runner.run({ payload: { amount: 100 } });

    expect(execution.getStatus()).toBe('SUCCEEDED');
    expect(execution.getResult()).toEqual({
      transactionId: 'txn-123',
      status: 'success'
    });
  });

  it('should retry on failure and eventually succeed', async () => {
    let callCount = 0;
    (global.fetch as jest.Mock).mockImplementation