Как смоделировать вызовы AWS при использовании Boto3 (версия 1.8 или выше) с Moto - PullRequest
Новая
       62

Как смоделировать вызовы AWS при использовании Boto3 (версия 1.8 или выше) с Moto

0 голосов
/ 11 июля 2019

У меня есть API, написанный на python, который выполняет вызовы сервисов AWS, в частности sqs, s3 и dynamicodb. Я пытаюсь написать модульные тесты для API и хочу высмеивать все вызовы AWS. Я провел много исследований в области мотоциклов, чтобы посмеяться над этими сервисами, однако каждая попытка реализации не имитирует мои звонки и отправляет реальные запросы в AWS. Заглядывая в эту проблему, я обнаружил, что люди обсуждают некоторые несовместимости между boto и moto при использовании boto3> = 1.8. Есть ли способ обойти это? Мой окончательный вопрос заключается в следующем: есть ли простой способ смоделировать вызовы boto3 для sqs, s3 и dynamicodb, используя moto или какую-то другую библиотеку при использовании boto3> = 1.8?

Вот мои текущие версии boto3 и moto, которые я использую: 

boto3 == 1.9.314
moto == 1.3.11

Ниже моя последняя попытка использовать moto для насмешки звонков на sqs. Я определил прибор Pytest, в котором я создаю сеанс mock_sqs и (надеюсь, фальшивую) очередь. Я использую этот прибор для модульного тестирования моей функции get_queue_item.

SQS Script 

# 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, к сожалению, не вариант. Есть ли другой способ получить желаемые результаты с другой библиотекой? Я немного заглянул в локальный стек, но хочу убедиться, что это мой единственный вариант, прежде чем я полностью разочаруюсь в мотоцикле.

1 Ответ

0 голосов
/ 16 июля 2019

Я нашел способ высмеять все мои звонки AWS! Теперь я уверен, что moto и boto3> = 1.8 в настоящее время имеют серьезные проблемы несовместимости. Оказывается, проблема в botocore> = 1.11.0, который больше не использует запросы и вместо этого напрямую использует urllib3: это означает, что moto не может использовать ответы так же, как раньше, и, следовательно, возникают проблемы несовместимости. Чтобы обойти это, я вместо этого создал автономные мото-серверы для каждой из сервисов AWS, над которыми я хотел поиграть, и это сработало! Создавая фиктивные серверы и не насмехаясь над самими запросами, не было никаких проблем с мотоциклом, использующим ответы.

Я установил эти фиктивные серверы в фоновом режиме, используя отдельный скрипт start_local.py. Затем я удостоверился, что изменил объекты ресурсов и клиентов моего модульного теста boto3, чтобы теперь ссылаться на эти фиктивные конечные точки. Теперь я могу запускать свои тесты без звонков в 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
...