如何从Karate向SQS发送事件并进行测试？

一、Java原生SDK集成方案 #

Karate本身基于Java生态，直接使用AWS SDK for Java是最稳定、可扩展的方案，适合复杂逻辑场景。

步骤与代码示例 #

1. 引入依赖（Maven为例）：

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>sqs</artifactId>
    <version>2.25.0</version> <!-- 使用最新稳定版 -->
</dependency>

2. 编写Java工具类封装SQS操作：

import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.*;
import java.util.List;

public class SqsTestUtils {
    // 初始化SQS客户端（可根据环境配置区域）
    private static final SqsClient SQS_CLIENT = SqsClient.builder()
            .region(Region.US_EAST_1)
            .build();

    // 发送消息到指定队列
    public static String sendEvent(String queueUrl, String eventBody) {
        SendMessageRequest request = SendMessageRequest.builder()
                .queueUrl(queueUrl)
                .messageBody(eventBody)
                .build();
        return SQS_CLIENT.sendMessage(request).messageId();
    }

    // 接收队列消息（长轮询）
    public static List<Message> receiveEvents(String queueUrl, int maxMessages) {
        ReceiveMessageRequest request = ReceiveMessageRequest.builder()
                .queueUrl(queueUrl)
                .maxNumberOfMessages(maxMessages)
                .waitTimeSeconds(15) // 长轮询减少空响应
                .build();
        return SQS_CLIENT.receiveMessage(request).messages();
    }

    // 测试后删除消息，避免堆积
    public static void deleteMessage(String queueUrl, String receiptHandle) {
        DeleteMessageRequest request = DeleteMessageRequest.builder()
                .queueUrl(queueUrl)
                .receiptHandle(receiptHandle)
                .build();
        SQS_CLIENT.deleteMessage(request);
    }
}

3. 在Karate Feature中调用：

Feature: 用户创建事件SQS测试
  Background:
    * def SqsTestUtils = Java.type('com.yourpackage.SqsTestUtils')
    * def targetQueue = 'https://sqs.us-east-1.amazonaws.com/123456789012/user-events-queue'

  Scenario: 发送用户创建事件并验证接收
    # 构造事件体
    * def userEvent = { eventType: 'USER_CREATED', userId: 'U_001', timestamp: '#now' }
    # 发送事件
    * def messageId = SqsTestUtils.sendEvent(targetQueue, userEvent)
    * print '已发送消息ID:', messageId

    # 接收并验证事件
    * def receivedMessages = SqsTestUtils.receiveEvents(targetQueue, 1)
    * match receivedMessages.size() == 1
    * def receivedEvent = receivedMessages[0].body()
    * match receivedEvent == userEvent

    # 清理测试消息
    * SqsTestUtils.deleteMessage(targetQueue, receivedMessages[0].receiptHandle())

二、JavaScript脚本方案 #

适合快速原型开发，无需编写Java代码，直接在Karate中嵌入JavaScript逻辑。

方案1：使用AWS SDK for JavaScript #

Feature: 订单事件SQS测试
  Background:
    * def AWS = karate.require('aws-sdk')
    * def sqsClient = new AWS.SQS({ region: 'us-east-1' })
    * def orderQueue = 'https://sqs.us-east-1.amazonaws.com/123456789012/order-events-queue'

  Scenario: 发送订单支付事件
    # 构造请求参数
    * def sendParams = {
        QueueUrl: orderQueue,
        MessageBody: '{ "eventType": "ORDER_PAID", "orderId": "ORD_007", "amount": 99.99 }'
      }
    # 调用SDK发送消息
    * def sendResult = karate.call(sqsClient.sendMessage, sendParams)
    * match sendResult.MessageId != '#null'

    # 接收消息验证
    * def receiveParams = {
        QueueUrl: orderQueue,
        MaxNumberOfMessages: 1,
        WaitTimeSeconds: 10
      }
    * def receiveResult = karate.call(sqsClient.receiveMessage, receiveParams)
    * match receiveResult.Messages.size() == 1
    * def receivedBody = karate.jsonPath(receiveResult, '$..Body')[0]
    * match receivedBody.orderId == 'ORD_007'

方案2：直接调用SQS REST API #

无需依赖SDK，手动构造带AWS签名的POST请求：

Feature: 物流事件SQS测试
  Background:
    * def logisticsQueue = 'https://sqs.us-east-1.amazonaws.com/123456789012/logistics-events-queue'
    * def accessKey = karate.properties['aws.accessKey']
    * def secretKey = karate.properties['aws.secretKey']

  Scenario: 发送物流发货事件
    # 构造表单格式的请求体（符合SQS REST API要求）
    * def requestBody = 'Action=SendMessage&MessageBody={"eventType":"DELIVERY_SHIPPED","trackingId":"TRK_123"}'
    # 生成AWS V4签名
    * def signedHeaders = karate.signAwsV4('POST', logisticsQueue, requestBody, 'us-east-1', 'sqs')
    # 发送请求
    * def response = karate.post(logisticsQueue)
        .headers(signedHeaders)
        .body(requestBody)
    * match response.status == 200
    * match response.xml('/SendMessageResponse/SendMessageResult/MessageId') != '#null'

三、复用性优化：Karate工具Feature封装 #

将SQS操作封装为可复用的Feature，避免重复代码：

# sqs-common.feature
Feature: SQS通用操作工具
  Background:
    * def SqsTestUtils = Java.type('com.yourpackage.SqsTestUtils')

  Scenario: 发送事件
    * param queueUrl = '#notnull'
    * param eventBody = '#notnull'
    * def messageId = SqsTestUtils.sendEvent(queueUrl, eventBody)
    * return messageId

  Scenario: 接收并清理事件
    * param queueUrl = '#notnull'
    * def messages = SqsTestUtils.receiveEvents(queueUrl, 1)
    * if (messages.size() > 0) SqsTestUtils.deleteMessage(queueUrl, messages[0].receiptHandle())
    * return messages

在业务测试Feature中调用：

Feature: 会员升级流程测试
  Background:
    * def sqsCommon = call read('classpath:sqs-common.feature')
    * def memberQueue = 'https://sqs.us-east-1.amazonaws.com/123456789012/member-events-queue'

  Scenario: 会员升级触发SQS事件
    * def upgradeEvent = { eventType: 'MEMBER_UPGRADED', memberId: 'M_100', level: 'VIP' }
    * def messageId = sqsCommon.send({ queueUrl: memberQueue, eventBody: upgradeEvent })
    * def receivedEvents = sqsCommon.receive({ queueUrl: memberQueue })
    * match receivedEvents[0].body() == upgradeEvent

关键注意事项 #

• 凭证管理：通过环境变量、Karate配置文件或IAM角色传递AWS凭证，禁止硬编码
• 消息清理：测试后务必删除消息，避免队列堆积影响后续用例
• 长轮询配置：设置WaitTimeSeconds为10-20秒，减少空响应，提升测试稳定性
• 错误重试：对SQS临时错误（如限流）使用Karate的retry关键字添加重试逻辑

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