如何在自动化测试中等待EventBridge触发的Step Function执行完成？

我们当前的流程是：文件上传至S3后，触发EventBridge S3事件，进而启动Step Function，待数据迁移到目标位置后Step Function执行完毕。

现在需要在Python测试环境（本地及CI/CD）中实现该流程的自动化端到端测试，核心问题是如何在检查输出位置前合理等待Step Function执行完成。

我已经考虑过几种方案，但都存在弊端：

• 设置定时器，等待n分钟直至任务完成——效率极低，完全不灵活
• 在Step Function中添加写入SNS的最终步骤，持续轮询SNS消息——只为测试新增基础设施，生产环境用不上，而且轮询还是低效
• 为Step Function添加标签或参数以便查找ARN，再轮询其执行状态——EventBridge触发Step Function时没法附加标签，轮询本身也有一定低效性
• 持续检查输出位置——同样效率低下，还可能出现误判（比如文件存在但未完全写入）
• 使用Step Functions Local测试Step Function——没法复现EventBridge触发逻辑，不是完整的端到端测试

想请教：有没有办法找到由我们主动触发的EventBridge事件所启动的Step Function的ARN？或者有其他更高效的解决方案？

1. 通过EventBridge事件历史追踪关联执行ID #

主动上传文件触发EventBridge事件后，可通过EventBridge的事件历史接口匹配对应Step Function的执行实例：

• 用Python的boto3调用eventbridge.list_events()，设置过滤条件：
  • 事件源为s3.amazonaws.com
  • 事件类型为ObjectCreated:*
  • 匹配你上传的S3对象键（detail.object.key）
• 找到该S3事件后，在同一时间窗口内查询事件源为states.amazonaws.com的StartExecution事件，通过detail.input中的S3对象键关联到对应的Step Function执行，从而获取执行ARN/ID。
• 拿到执行ID后，调用stepfunctions.describe_execution()轮询状态，直到进入终态（SUCCEEDED/FAILED/ABORTED），这种方式比盲等或轮询输出位置精准得多。

2. 给Step Function执行注入测试唯一标识 #

利用EventBridge规则的输入转换功能，给Step Function的执行输入添加测试专属的唯一ID（比如UUID），无需修改生产逻辑：

• 在测试环境的EventBridge规则中配置输入转换，示例配置：

  {
    "s3Detail": "$.detail",
    "testCorrelationId": "<你的测试会话UUID>"
  }
  

• 测试时生成唯一的testCorrelationId，上传文件后调用stepfunctions.list_executions()，过滤目标状态机的近期执行；遍历执行列表，通过describe_execution()获取执行输入，匹配testCorrelationId找到对应的执行ID。
• 之后用指数退避策略轮询该执行的状态，直到完成。

3. 通过CloudWatch日志关联执行ID #

如果Step Function已开启CloudWatch日志，可通过Log Insight查询快速定位执行ID：

• 编写Log Insight查询语句，匹配目标S3对象键：

  fields @timestamp, @message
  | filter @message like '<你的S3对象键>'
  | filter logStream like '<你的Step Function日志流前缀>'
  | sort @timestamp desc
  | limit 1
  

• 从返回的日志中提取执行ID，再轮询Step Function的执行状态。这种方式无需修改业务流程，仅依赖日志配置。

4. 优化轮询策略降低资源消耗 #

无论用哪种方式获取执行ID，都可以通过指数退避轮询减少无效请求：
使用Python的tenacity库实现指数退避逻辑，示例代码：

from tenacity import retry, stop_after_attempt, wait_exponential
import boto3

sf_client = boto3.client('stepfunctions')

@retry(wait=wait_exponential(multiplier=1, min=1, max=30), stop=stop_after_attempt(10))
def wait_for_execution_completion(execution_arn):
    execution_detail = sf_client.describe_execution(executionArn=execution_arn)
    status = execution_detail['status']
    if status in ['SUCCEEDED', 'FAILED', 'ABORTED']:
        return status
    # 未进入终态则抛出异常，触发重试
    raise RuntimeError(f"Execution {execution_arn} still in status: {status}")

# 调用示例
target_execution_arn = "获取到的执行ARN"
final_status = wait_for_execution_completion(target_execution_arn)

内容的提问来源于stack exchange，提问作者WarSame