在Moto中使用Boto3(1.8版或更高版本)时如何模拟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.在研究这个问题时，我发现人们在讨论使用boto3时boto和moto之间存在一些不兼容之处> = 1.8.有没有办法解决?我的最终问题是:在使用boto3> = 1.8时，是否有一种简单的方法可以使用moto或其他库来模拟对sqs，s3和dynamodb的boto3调用?

I have an API written in python that makes calls to AWS services, specifically sqs, s3, and dynamodb. I am trying to write unit tests for the API and I want to mock all calls to AWS. I have done a lot of research into moto as a way to mock these services however every implementation I have tried does not mock my calls and sends real requests to AWS. Looking into this problem I found people discussing some incompatibilities between boto and moto when using boto3>=1.8. Is there any way around this? My ultimate question is this: Is there an easy way to mock boto3 calls to sqs, s3, and dynamodb using either moto or some other library when using boto3>=1.8?

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

Here are my current versions of boto3 and moto I am using:

boto3 == 1.9.314
moto == 1.3.11

以下是我最近使用moto模拟对sqs的调用的尝试.我定义了一个pytest固定装置，在其中创建了一个mock_sqs会话和一个(希望是假的)队列.我使用此夹具对我的get_queue_item函数进行单元测试.

Below is my latest attempt at using moto to mock calls to sqs. I defined a pytest fixture where I create a mock_sqs session and a (hopefully fake) queue. I use this fixture to unit test my get_queue_item function.

# 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之前，我想确保这是我唯一的选择.

When I go to check the console however, I see the queue has actually been created. I have also tried moving around the order of my imports but nothing seemed to work. I tried using mock decorators and I even briefly played around with moto's stand-alone server mode. Am I doing something wrong or is it really just the boto3/moto incompatibility I have been hearing about with newer versions of boto3? Downgrading my version of boto3 is not an option unfortunately. Is there another way to get the results I want with another library? I have looked a little bit into localstack but I want to make sure that is my only option before I give up on moto entirely.

推荐答案

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

I figured out a way to mock all my AWS calls! I am confident now that moto and boto3>=1.8 currently has serious incompatibility issues. Turns out the problem is with botocore >= 1.11.0 which no longer uses requests and instead directly uses urllib3: This means moto cannot use responses the same way it did before, hence the incompatibility issues. To get around this though, I instead created stand-alone moto servers for each of the AWS services I wanted to mock which worked like a charm! By creating the mock servers and not mocking the requests themselves, there wasn't any issues with moto using responses.

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

I set these mock servers running in the backgound by using a separate start_local.py script. Next I made sure to change my unit test's boto3 reource and client objects to now reference these mock endpoints. Now I can run my pytests without any calls being made to aws and no need to mock aws credentials!

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

Below is the new start_local.py script and my updated sqs unit test:

# 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

这篇关于在Moto中使用Boto3(1.8版或更高版本)时如何模拟AWS调用的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！