测试
AWS 提供了专用的测试 SDK，用于在本地和云端运行和检查 durable functions 的执行情况。该测试 SDK 提供两种测试模式：
- Local testing - 在开发环境中快速进行单元测试
- Cloud testing - 针对已部署资源进行集成测试，以在生产前验证网络和 IAM 策略
为什么 Durable Function 测试不一样
durable functions 的测试与传统 Lambda 函数的测试不同
传统 Lambda 函数测试
如果我们之前在 Lambda 上构建过应用程序，我们已经习惯了这种测试 Lambda 函数的思维模型：
- 在单次调用中执行（秒或分钟）
- 具有简单的输入 → 处理 → 输出流程
- 调用之间没有持久状态（除了环境中的全局变量）
- 快速完成，使集成测试变得可行
Durable Function 测试的复杂性
Durable functions 引入了新的测试挑战：
Multi-Invocation Execution：单次 durable 执行跨越多个 Lambda 函数调用。每次调用在继续之前都会重放之前的步骤，这使得端到端测试完整工作流变得困难。
State Persistence：Durable functions 通过检查点在调用之间维护状态。测试需要验证状态在执行边界之间是否被正确保存和恢复。
Time-Based Operations：工作流通常包括等待、延迟和超时，可能跨越分钟、小时或天。传统测试需要等待这些延迟完成。
Checkpoint Replay：检查点重放模型意味着我们的函数代码会多次执行，从保存的检查点重放。测试必须验证跨重放的正确行为。
Durable Execution Testing SDK 解决了这些挑战Durable Execution SDK 测试库可在毫秒内本地运行工作流，即使是通常需要数小时或数天的工作流也不例外。无需部署到 AWS 即可进行测试，并跳过等待和延迟。
测试Durable Function
下载源码：
https://pingfan.s3.amazonaws.com/files/src-lambda-durable.zip
unzip src-lambda-durable.zip
cd src/python/02-testing
打开 order_processor.py 文件。
Durable steps 定义：
我们为 validate_order、process_payment 和 confirm_order 定义 durable steps。每个步骤返回不同的状态。
from aws_durable_execution_sdk_python import (
DurableContext,
durable_execution,
durable_step,
)
from aws_durable_execution_sdk_python.config import Duration
@durable_step
def validate_order(step_context, order_id):
step_context.logger.info(f"Validating order {order_id}")
return {"orderId": order_id, "status": "validated"}
@durable_step
def process_payment(step_context, order_id):
step_context.logger.info(f"Processing payment for order {order_id}")
return {"orderId": order_id, "status": "paid", "amount": 99.99}
@durable_step
def confirm_order(step_context, order_id):
step_context.logger.info(f"Confirming order {order_id}")
return {"orderId": order_id, "status": "confirmed"}
@durable_execution
def lambda_handler(event, context: DurableContext):
order_id = event['orderId']
# Step 1: Validate order
validation_result = context.step(validate_order(order_id))
# Step 2: Process payment
payment_result = context.step(process_payment(order_id))
# Wait for 10 seconds to simulate external confirmation
context.wait(Duration.from_seconds(10))
# Step 3: Confirm order
confirmation_result = context.step(confirm_order(order_id))
return {
"orderId": order_id,
"status": "completed",
"steps": [validation_result, payment_result, confirmation_result]
}
handler 定义了 durable execution 工作流并调用每个步骤。它返回整体状态和各个步骤的结果。请注意 context.wait() 调用，它会等待 10 秒以模拟外部确认。
创建测试函数
在 02-testing 文件夹中创建名为 test_order_processor.py 的文件，并将以下内容粘贴到其中：
from aws_durable_execution_sdk_python_testing import DurableFunctionTestRunner
from aws_durable_execution_sdk_python.execution import InvocationStatus
from order_processor import lambda_handler
def test_order_processor_success():
# Create a test runner for your function
runner = DurableFunctionTestRunner(handler=lambda_handler)
with runner:
# Run the function with test input
result = runner.run(
input={"orderId": "order-12345"},
timeout=20
)
# Verify the execution succeeded
assert result.status is InvocationStatus.SUCCEEDED
# Check the result
execution_result = result.result
print(execution_result)
# Convert to dict if it's a string
if isinstance(execution_result, str):
import json
execution_result = json.loads(execution_result)
# Check basic success criteria
assert execution_result["orderId"] == "order-12345"
assert execution_result["status"] == "completed"
assert len(execution_result["steps"]) == 3
# Check that all steps executed correctly
assert execution_result["steps"][0]["status"] == "validated"
assert execution_result["steps"][1]["status"] == "paid"
assert execution_result["steps"][1]["amount"] == 99.99
assert execution_result["steps"][2]["status"] == "confirmed"
我们为函数创建一个运行器，并在本地模拟 Lambda 函数执行。run() 使用测试数据执行我们的函数，并返回包含状态和输出的结果对象。即使有较长的等待时间，也能在毫秒内完成。
通过检查最终执行状态来验证结果：使用标准 Python 断言验证结果数据的 InvocationStatus.SUCCEEDED、FAILED 或 RUNNING，并验证步骤结果。
设置环境和依赖项
在 IDE 中打开新终端并输入以下内容
cd src/python/02-testing/
python3 -m venv .venv
source .venv/bin/activate
pip3 install -r requirements.txt
运行测试
执行测试：
python3 -m pytest test_order_processor.py -v -s
我们应该看到如下输出：
请注意打印出的执行结果。它包含整体状态以及每个步骤的结果。
另请注意运行测试所花费的时间。由于等待时间，运行时间略超过 10 秒。
使用TIME_SCALE运行测试
时间缩放允许我们：
- 立即完成
context.wait() - 使重试延迟立即发生
- 让测试在毫秒内运行
这使测试快速且确定，无需等待实际延迟，也有助于测试长时间运行的工作流。在 durable functions 测试 SDK 中，我们使用名为 DURABLE_EXECUTION_TIME_SCALE 的环境变量来实现这一点。让我们看看它是如何工作的。
执行以下命令：
export DURABLE_EXECUTION_TIME_SCALE=0.01
python3 -m pytest test_order_processor.py -v -s
我们将获得类似以下的输出。请注意它几乎是瞬间完成的。
其他测试场景（可选）
我们可以测试其他场景，如重试、回调、错误、并行等。使用 Python Testing SDK 中的示例 查看如何在 src 文件夹中使用不同模式，以及如何在 test 文件夹中测试每种模式。
CI/CD 流水线集成
在本地验证 durable functions 后，我们可以将这些测试集成到 CI/CD 流水线中，以便在部署前进行自动化测试。
GitHub Actions 示例
在我们的仓库中创建 .github/workflows/test.yml 文件：
name: Test Durable Functions
on: [push， pull_request]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
cd src/python/02-testing
python -m pip install --upgrade pip
pip install -r requirements.txt
- name: Run tests with time scaling
env:
DURABLE_EXECUTION_TIME_SCALE: 0.01
run: |
cd src/python/02-testing
python -m pytest test_order_processor.py -v
AWS CodeBuild 示例
在我们的仓库中创建 buildspec.yml 文件：
version: 0.2
phases:
install:
runtime-versions:
python: 3.11
pre_build:
commands:
- cd src/python/02-testing
- pip install -r requirements.txt
build:
commands:
- export DURABLE_EXECUTION_TIME_SCALE=0.01
- python -m pytest test_order_processor.py -v
reports:
pytest_reports:
files:
- 'test-results.xml'
file-format: 'JUNITXML'
CI/CD 集成最佳实践
- 始终使用时间缩放（
DURABLE_EXECUTION_TIME_SCALE=0.01）在 CI/CD 中快速构建 - 在部署前运行测试，作为构建流水线的一部分，以便尽早发现问题
- 使用测试覆盖率工具，如
pytest-cov，以确保全面的测试覆盖率 - 快速失败，在流水线早期、昂贵的构建或部署步骤之前进行测试
- 测试多种场景，包括成功案例、错误处理和边缘情况