1

我有一个用 python 编写的 API，可以调用 AWS 服务，特别是 sqs、s3 和 dynamodb。我正在尝试为 API 编写单元测试，并且我想模拟对 AWS 的所有调用。我已经对 moto 进行了大量研究，以此来模拟这些服务，但是我尝试过的每个实现都不会模拟我的调用，而是向 AWS 发送真正的请求。在研究这个问题时，我发现人们在使用 boto3>=1.8 时讨论了 boto 和 moto 之间的一些不兼容问题。有没有办法解决？我的最终问题是：在使用 boto3>=1.8 时，是否有一种简单的方法可以使用 moto 或其他库来模拟 boto3 对 sqs、s3 和 dynamodb 的调用？

这是我当前使用的 boto3 和 moto 版本：

boto3 == 1.9.314
moto == 1.3.11

下面是我最近尝试使用 moto 模拟对 sqs 的调用。我定义了一个 pytest 夹具，在其中创建了一个 mock_sqs 会话和一个（希望是假的）队列。我使用这个夹具对我的 get_queue_item 函数进行单元测试。

SQS 脚本

# ptr_api.aws.sqs
import boto3

REGION = 'us-east-1'

sqs_r = boto3.resource('sqs', REGION)
sqs_c = boto3.client('sqs', REGION)

def get_queue_item(queue_name):
    queue = sqs_r.get_queue_by_name(QueueName=queue_name)
    queue_url = queue.url

    response = sqs_c.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=1,    
        VisibilityTimeout=10,
        WaitTimeSeconds=3
    )

    try:
        message = response['Messages'][0]
        receipt_handle = message['ReceiptHandle']
        delete_response = sqs_c.delete_message(QueueUrl=queue_url,
        ReceiptHandle=receipt_handle)
        return message['Body']
    except Exception as e:
        print("error in get_queue_item: ")
        print(e)
        return False

测试 SQS 脚本

# test_sqs.py
import pytest
from moto import mock_sqs
import boto3
from ptr_api.aws.sqs import get_queue_item

@pytest.fixture
def sqs_mocker(scope='session', autouse=True):
   mock = mock_sqs()
   mock.start()
   
   sqs_r = boto3.resource('sqs', 'us-east-1')
   sqs_c = boto3.client('sqs', 'us-east-1')

   queue_name = 'test_queue_please_dont_actually_exist'

   queue_url = sqs_c.create_queue(
       QueueName=queue_name
   )['QueueUrl']

   yield (sqs_c, queue_url, queue_name)
   mock.stop()

def test_get_queue_item(sqs_mocker):
   sqs_c, queue_url, queue_name = sqs_mocker

   message_body = 'why hello there' # Create dummy message
   sqs_c.send_message(              # Send message to fake queue
       QueueUrl=queue_url,
       MessageBody=message_body,
   )

   res = get_queue_item(queue_name) # Test get_queue_item function

   assert res == message_body

然而，当我去检查控制台时，我看到队列实际上已经创建了。我也尝试过改变我的进口顺序，但似乎没有任何效果。我尝试使用模拟装饰器，甚至短暂地玩过 moto 的独立服务器模式。我做错了什么还是真的只是我听说过的较新版本的 boto3 与 boto3/moto 不兼容？不幸的是，降级我的 boto3 版本不是一种选择。有没有另一种方法可以通过另一个库获得我想要的结果？我对 localstack 进行了一些研究，但我想确保这是我完全放弃 moto 之前的唯一选择。

4

1 回答 1

4

我想出了一种方法来模拟我所有的 AWS 调用！我现在确信 moto 和 boto3>=1.8 目前存在严重的不兼容问题。原来问题在于 botocore >= 1.11.0，它不再使用请求，而是直接使用 urllib3：这意味着 moto 不能像以前那样使用响应，因此存在不兼容问题。不过，为了解决这个问题，我改为为我想模拟的每个 AWS 服务创建了独立的 moto 服务器，这就像一个魅力！通过创建模拟服务器而不模拟请​​求本身，moto 使用响应没有任何问题。

我使用单独的 start_local.py 脚本设置这些模拟服务器在后台运行。接下来，我确保将单元测试的 boto3 资源和客户端对象更改为现在引用这些模拟端点。现在我可以运行我的 pytest，而无需对 aws 进行任何调用，也无需模拟 aws 凭据！

下面是新的 start_local.py 脚本和我更新的 sqs 单元测试：

启动本地 AWS 服务

# start_local.py
import boto3
import threading, subprocess

def start_sqs(port=5002):
    subprocess.call(["moto_server", "sqs", f"-p{port}"])

sqs = threading.Thread(target=start_sqs)

sqs.start()

新的测试 SQS 脚本

import pytest
import boto3
import os
from ptr_api.aws import sqs

@pytest.fixture
def sqs_mocker(scope='session', autouse=True):

    sqs_r_mock = boto3.resource('sqs', region_name='us-east-1', endpoint_url=f'http://localhost:5002')
    sqs_c_mock = boto3.client('sqs', region_name='us-east-1', endpoint_url=f'http://localhost:5002')

    queue_name = 'test_queue'

    queue_url = sqs_c_mock.create_queue(
        QueueName=queue_name
    )['QueueUrl']

    yield (sqs_r_mock, sqs_c_mock, queue_url, queue_name)

def test_get_queue_item(sqs_mocker):

    sqs_r_mock, sqs_c_mock, queue_url, queue_name = sqs_mocker

    message_body = 'why hello there' # Create dummy message
    sqs_c_mock.send_message(         # Send message to fake queue
        QueueUrl=queue_url,
        MessageBody=message_body,
    )

    sqs.sqs_r = sqs_r_mock # VERY IMPORTANT - Override boto3 resource global variable within imported module with mock resource
    sqs.sqs_c = sqs_c_mock # VERY IMPORTANT - Override boto3 client global variable within imported module with mock client
    res = sqs.get_queue_item(queue_name) # Test get_queue_item function

    assert res == message_body
于 2019-07-16T17:24:11.693 回答