将 Boto3(版本 1.8 或更高版本)与 Moto 一起使用时如何模拟 AWS 调用
How to mock AWS calls when using Boto3 (version 1.8 or higher) with Moto
我有一个用 python 编写的 API 可以调用 AWS 服务,特别是 sqs、s3 和 dynamodb。我正在尝试为 API 编写单元测试,并且我想模拟对 AWS 的所有调用。我对 moto 作为模拟这些服务的一种方式进行了大量研究,但是我尝试过的每个实现都不会模拟我的调用并向 AWS 发送真实请求。在调查这个问题时,我发现 people discussing 在使用 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 fixture,我在其中创建了一个 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/moto 我听说的与较新版本的 boto3 不兼容?不幸的是,不能选择降级我的 boto3 版本。有没有其他方法可以通过另一个库获得我想要的结果?我已经对 localstack 进行了一些研究,但我想确保这是我完全放弃 moto 之前唯一的选择。
我找到了一种模拟所有 AWS 调用的方法!我现在有信心 moto 和 boto3>=1.8 目前存在严重的不兼容问题。事实证明,问题出在 botocore >= 1.11.0,它不再使用请求,而是直接使用 urllib3:这意味着 moto 不能像以前那样使用响应,因此存在不兼容问题。不过,为了解决这个问题,我为每个我想模拟的 AWS 服务创建了 stand-alone moto 服务器,这非常有效!通过创建模拟服务器而不是模拟请求本身,moto 使用响应没有任何问题。
我使用单独的 start_local.py 脚本在后台设置这些模拟服务器 运行ning。接下来,我确保更改我的单元测试的 boto3 资源和客户端对象,以引用这些模拟端点。现在我可以 运行 我的 pytests 而无需对 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
我有一个用 python 编写的 API 可以调用 AWS 服务,特别是 sqs、s3 和 dynamodb。我正在尝试为 API 编写单元测试,并且我想模拟对 AWS 的所有调用。我对 moto 作为模拟这些服务的一种方式进行了大量研究,但是我尝试过的每个实现都不会模拟我的调用并向 AWS 发送真实请求。在调查这个问题时,我发现 people discussing 在使用 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 fixture,我在其中创建了一个 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/moto 我听说的与较新版本的 boto3 不兼容?不幸的是,不能选择降级我的 boto3 版本。有没有其他方法可以通过另一个库获得我想要的结果?我已经对 localstack 进行了一些研究,但我想确保这是我完全放弃 moto 之前唯一的选择。
我找到了一种模拟所有 AWS 调用的方法!我现在有信心 moto 和 boto3>=1.8 目前存在严重的不兼容问题。事实证明,问题出在 botocore >= 1.11.0,它不再使用请求,而是直接使用 urllib3:这意味着 moto 不能像以前那样使用响应,因此存在不兼容问题。不过,为了解决这个问题,我为每个我想模拟的 AWS 服务创建了 stand-alone moto 服务器,这非常有效!通过创建模拟服务器而不是模拟请求本身,moto 使用响应没有任何问题。
我使用单独的 start_local.py 脚本在后台设置这些模拟服务器 运行ning。接下来,我确保更改我的单元测试的 boto3 资源和客户端对象,以引用这些模拟端点。现在我可以 运行 我的 pytests 而无需对 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