使用moto mock AWS SQS时出现AttributeError与ClientError如何解决

第一个报错（AttributeError: 'dict' object has no attribute 'send_message'） #

原因：你使用boto3的client模式调用create_queue接口，返回值是包含队列属性的字典，不是boto3 resource模式下的Queue实例对象，本身不携带send_message方法。
修复：直接使用你注释的conn.send_message写法即可，这是client模式下发送消息的标准写法。 #

第二个报错（InvalidAddress 地址不合法） #

原因：你业务代码中的sqs客户端是全局初始化的，模块被导入时就已经创建完成，此时测试用例的@mock_sqs装饰器还没启动mock服务，该客户端没有被moto拦截，会默认请求真实的AWS SQS公开地址，和moto本地mock的endpoint不匹配。
修复方案二选一即可：

方案1：修改业务代码为惰性初始化客户端 #

不要全局创建sqs客户端，改成在函数调用时再初始化，同时显式指定region，避免默认地址不匹配：

# 修正后的业务代码
import boto3

def get_msg_from_queue(queue_url: str) -> dict:
    # 函数执行时才创建客户端，保证mock启动后生成的客户端会被拦截
    sqs = boto3.client('sqs', region_name='us-east-1')
    return sqs.receive_message(
        QueueUrl=queue_url, 
        AttributeNames=['All'],
        MaxNumberOfMessages=1, 
        VisibilityTimeout=3600, 
        WaitTimeSeconds=0
    )

方案2：调整测试代码的导入顺序 #

如果不想修改业务代码，就在测试用例的mock启动后再导入业务模块，把导入逻辑放到测试函数内部即可：

# 修正后的测试代码
from moto import mock_sqs
import boto3
import json
import os

@mock_sqs
def test_get_all_msg_from_queue():
    # mock启动后再导入业务方法，此时业务模块的全局sqs客户端会被moto拦截
    from myClass import get_msg_from_queue
    
    conn = boto3.client('sqs', region_name='us-east-1')
    queue = conn.create_queue(QueueName='Test')
    queue_url = queue["QueueUrl"]
    os.environ["SQS_URL"] = queue_url
    
    # 使用client模式发送消息
    conn.send_message(
        QueueUrl=queue_url, 
        MessageBody=json.dumps({'a': '1', 'b': '2', 'c': '3'})
    )

    resp = get_msg_from_queue(queue_url) 
    assert resp is not None

额外注意事项 #

所有调用SQS的客户端（测试侧、业务侧）必须显式指定相同的region，moto对region的匹配校验比较严格，否则也会出现endpoint不匹配的问题。

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