升级Moto至4.2.x后Pytest测试报SQS去重相关ClientError如何解决?
Moto 4.2.x升级后SQS SendMessage测试错误原因及修复方案
错误原因
Moto 4.2.x版本开始严格对齐AWS SQS的官方行为规范:针对FIFO类型队列,AWS要求必须满足以下两个条件之一才能发送消息:
- 队列启用
ContentBasedDeduplication(基于内容自动去重) - 发送消息时显式提供
MessageDeduplicationId参数
而Moto 4.1.x及更早版本没有强制校验这一规则,导致升级后原本通过的测试代码触发了参数合法性校验错误。
修复方案
方案一:给FIFO队列启用自动内容去重
如果测试场景无需手动控制去重逻辑,可在创建队列时开启自动去重配置:
import boto3 from moto import mock_sqs @mock_sqs def test_sqs_send_message(): sqs_client = boto3.client('sqs', region_name='us-east-1') # 创建带自动去重的FIFO队列 queue_url = sqs_client.create_queue( QueueName='test-fifo-queue.fifo', Attributes={ 'FifoQueue': 'true', 'ContentBasedDeduplication': 'true' } )['QueueUrl'] # 无需额外参数即可发送消息 sqs_client.send_message(QueueUrl=queue_url, MessageBody='test content')
方案二:发送消息时显式指定去重ID
如果需要自定义去重规则,在调用SendMessage时添加MessageDeduplicationId参数(建议使用唯一标识,比如UUID):
import boto3 from moto import mock_sqs import uuid @mock_sqs def test_sqs_send_message_with_dedupe_id(): sqs_client = boto3.client('sqs', region_name='us-east-1') queue_url = sqs_client.create_queue( QueueName='test-fifo-queue.fifo', Attributes={'FifoQueue': 'true'} )['QueueUrl'] # 发送消息时指定唯一去重ID sqs_client.send_message( QueueUrl=queue_url, MessageBody='test content', MessageDeduplicationId=str(uuid.uuid4()) )
内容的提问来源于stack exchange,提问作者wonton
相关产品推荐
相关产品推荐

