如何用Terratest测试SQS与Lambda的完整消息消费流程？

问题 #

我用Terratest测试Terraform部署的SQS+Lambda联动架构（Lambda作为SQS的消息消费者），想验证完整流程：向SQS发测试消息，然后读取消息。但Lambda会抢先消费消息，导致Terratest的WaitForQueueMessage超时拿不到消息，CloudWatch日志已经确认Lambda确实消费了消息。

测试代码 #

ack_queue_url := terraform.Output(t, terraformOptions, "acknowledgment_queue_url")
time_out_sec := 120
test_message := fmt.Sprintf("terratest-test-message-%s", uniqueId)
aws.SendMessageToQueue(t, awsRegion, ack_queue_url, test_message)
response := aws.WaitForQueueMessage(t, awsRegion, ack_queue_url, time_out_sec)
assert.NoError(t, response.Error)
fmt.Println("###Message Body####:", response.MessageBody)
aws.DeleteMessageFromQueue(t, awsRegion, ack_queue_url,  response.ReceiptHandle)
delete_response := aws.WaitForQueueMessage(t, awsRegion, ack_queue_url, time_out_sec)
assert.Error(t, delete_response.Error, aws.ReceiveMessageTimeout{QueueUrl: ack_queue_url, TimeoutSec: time_out_sec})

测试输出 #

logger.go:66: "https://sqs.us-east-1.amazonaws.com/1234567/tst-queue"
sqs.go:150: Sending message terratest-test-message-DkKAvt to queue https://sqs.us-east-1.amazonaws.com/1234567/tst-queue
sqs.go:170: Message id b9b0a000-1d71-4821-8659-21aebe33cdc0 sent to queue https://sqs.us-east-1.amazonaws.com/1234567/tst-queue
sqs.go:234: Waiting for message on https://sqs.us-east-1.amazonaws.com/1234567/tst-queue (0s)
sqs.go:234: Waiting for message on https://sqs.us-east-1.amazonaws.com/1234567/tst-queue(20s)
sqs.go:234: Waiting for message on https://sqs.us-east-1.amazonaws.com/1234567/tst-queue (40s)
sqs.go:234: Waiting for message on https://sqs.us-east-1.amazonaws.com/1234567/tst-queue (60s)
sqs.go:234: Waiting for message on https://sqs.us-east-1.amazonaws.com/1234567/tst-queue (80s)
sqs.go:234: Waiting for message on https://sqs.us-east-1.amazonaws.com/1234567/tst-queue (100s)
                Error Trace:    /Users/xxxx/Projects/dummy_test/terratest/complete_test.go:83
                Error:          Received unexpected error:
                                Failed to receive messages on https://sqs.us-east-1.amazonaws.com/1234567/tst-queue within 120 seconds
                Test:           complete_test
###Message Body####: 
sqs.go:125: Deleting message from queue https://sqs.us-east-1.amazonaws.com/1234567/tst-queue()
    sqs.go:119: MissingParameter: The request must contain the parameter ReceiptHandle.
                status code: 400, request id: 4421cdfc-4326-5755-9623-91ca3414ed6f

解决方案 #

方法1：测试阶段临时禁用Lambda触发器 #

在Terraform中通过变量控制SQS触发器的启用状态，分阶段验证：

1. 定义Terraform变量控制触发器开关：

variable "enable_lambda_sqs_trigger" {
  type    = bool
  default = true
}

resource "aws_lambda_event_source_mapping" "sqs_lambda" {
  event_source_arn = aws_sqs_queue.ack.arn
  function_name    = aws_lambda_function.consumer.arn
  enabled          = var.enable_lambda_sqs_trigger
}

2. Terratest测试流程调整：
   • 部署时禁用触发器，先验证SQS消息发送、读取功能正常
   • 启用触发器，等待Lambda消费消息，再验证SQS消息被清空，同时可检查Lambda执行记录确认消费成功

Terratest中设置变量示例：

terraformOptions := &terraform.Options{
  // 其他配置
  Vars: map[string]interface{}{
    "enable_lambda_sqs_trigger": false,
  },
}

方法2：验证Lambda的消费结果 #

既然Lambda会消费消息，直接验证Lambda的处理结果：

1. 给测试消息添加唯一标识（比如代码中的uniqueId）
2. Lambda消费后，将处理日志输出到CloudWatch，或把结果写入DynamoDB/验证用SQS
3. Terratest发送消息后，等待读取CloudWatch日志中包含该唯一标识的记录，或读取验证存储中的数据，确认Lambda完成处理

示例：用Terratest查询CloudWatch日志

logGroupName := fmt.Sprintf("/aws/lambda/%s", lambdaFunctionName)
logFilter := fmt.Sprintf('"terratest-test-message-%s"', uniqueId)
// 等待日志中出现目标消息标识
aws.WaitForLogMessages(t, awsRegion, logGroupName, logFilter, 1, time_out_sec)

方法3：给测试消息设置延迟 #

发送消息时设置延迟，让Terratest有足够时间先读取消息（延迟期间消息不可被任何消费者读取）：

// 发送带10秒延迟的消息
aws.SendMessageToQueueWithAttributes(t, awsRegion, ack_queue_url, test_message, map[string]string{
  "DelaySeconds": "10",
})
// 立即执行消息读取操作
response := aws.WaitForQueueMessage(t, awsRegion, ack_queue_url, time_out_sec)

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