Подтвердить что ты не робот

Mocking boto3 S3-клиентский метод Python

Я пытаюсь смоделировать метод singluar из объекта клиента boto3 s3, чтобы вызвать исключение. Но мне нужны все другие методы, чтобы этот класс работал как обычно.

Это так, чтобы я мог проверить исключительный тест исключения, когда и происходит ошибка при выполнении upload_part_copy

1-я попытка

import boto3
from mock import patch

with patch('botocore.client.S3.upload_part_copy', side_effect=Exception('Error Uploading')) as mock:
    client = boto3.client('s3')
    # Should return actual result
    o = client.get_object(Bucket='my-bucket', Key='my-key')
    # Should return mocked exception
    e = client.upload_part_copy()

Однако это дает следующую ошибку:

ImportError: No module named S3

2-я попытка

Посмотрев исходный код botocore.client.py, я обнаружил, что он делает что-то умное, а метод upload_part_copy не существует. Я обнаружил, что вместо этого он вызывает BaseClient._make_api_call поэтому я попытался

import boto3
from mock import patch

with patch('botocore.client.BaseClient._make_api_call', side_effect=Exception('Error Uploading')) as mock:
    client = boto3.client('s3')
    # Should return actual result
    o = client.get_object(Bucket='my-bucket', Key='my-key')
    # Should return mocked exception
    e = client.upload_part_copy()

Это вызывает исключение... но для get_object которого я хочу избежать.

Любые идеи о том, как я могу только выбросить исключение на метод upload_part_copy?

4b9b3361

ОТВЕТЫ

Ответ 1

Как только я опубликовал здесь, мне удалось найти решение. Здесь он надеется, что это поможет:)

import botocore
from botocore.exceptions import ClientError
from mock import patch
import boto3

orig = botocore.client.BaseClient._make_api_call

def mock_make_api_call(self, operation_name, kwarg):
    if operation_name == 'UploadPartCopy':
        parsed_response = {'Error': {'Code': '500', 'Message': 'Error Uploading'}}
        raise ClientError(parsed_response, operation_name)
    return orig(self, operation_name, kwarg)

with patch('botocore.client.BaseClient._make_api_call', new=mock_make_api_call):
    client = boto3.client('s3')
    # Should return actual result
    o = client.get_object(Bucket='my-bucket', Key='my-key')
    # Should return mocked exception
    e = client.upload_part_copy()

Иордания Philips также опубликовала отличное решение, используя botocore.stub.Stubber класс. В то время как более чистое решение я не мог издеваться над конкретными операциями.

Ответ 2

У Botocore есть клиентский стербер, который вы можете использовать только для этой цели: docs.

Здесь приведен пример ошибки:

import boto3
from botocore.stub import Stubber

client = boto3.client('s3')
stubber = Stubber(client)
stubber.add_client_error('upload_part_copy')
stubber.activate()

# Will raise a ClientError
client.upload_part_copy()

Здесь приведен пример ввода нормального ответа. Кроме того, теперь можно использовать стербер, который можно использовать в контексте. Важно отметить, что стербер будет проверять, насколько это возможно, ваш предоставленный ответ соответствует тому, что служба действительно вернется. Это не идеально, но это защитит вас от вставки полных бессмысленных ответов.

import boto3
from botocore.stub import Stubber

client = boto3.client('s3')
stubber = Stubber(client)
list_buckets_response = {
    "Owner": {
        "DisplayName": "name",
        "ID": "EXAMPLE123"
    },
    "Buckets": [{
        "CreationDate": "2016-05-25T16:55:48.000Z",
        "Name": "foo"
    }]
}
expected_params = {}
stubber.add_response('list_buckets', list_buckets_response, expected_params)

with stubber:
    response = client.list_buckets()

assert response == list_buckets_response

Ответ 3

Здесь приведен пример простого утилит python, который можно использовать для подделки client = boto3.client('ec2') api call...

import boto3 

class MyAWSModule():
    def __init__(self):
        client = boto3.client('ec2')
        tags = client.describe_tags(DryRun=False)


class TestMyAWSModule(unittest.TestCase):
    @mock.patch("boto3.client.get_tags")
    @mock.patch("boto3.client")
    def test_open_file_with_existing_file(self, mock_boto_client, mock_describe_tags):
        mock_boto_client.return_value = mock_get_tags_response
        my_aws_module = MyAWSModule()

        mock_boto_client.assert_call_once('ec2')
        mock_describe_tags.assert_call_once_with(DryRun=False)

mock_get_tags_response = {
    'Tags': [
        {
            'ResourceId': 'string',
            'ResourceType': 'customer-gateway',
            'Key': 'string',
            'Value': 'string'
        },
    ],
'NextToken': 'string'
}

надеюсь, что это поможет.

Ответ 4

Как насчет простого использования мото?

Он поставляется с очень удобным декоратором:

from moto import mock_s3

@mock_s3
def test_my_model_save():
    pass

Ответ 5

Мне пришлось издеваться boto3 клиентом boto3 для тестирования интеграции, и это было немного больно! Проблема, с которой я столкнулся, состоит в том, что moto не очень хорошо поддерживает KMS, но я не хотел переписывать свой собственный макет для S3. Таким образом, я создал этот морф из всех ответов. Также это работает глобально, что довольно круто!

У меня есть настройки с 2 файлами.

Первый - это aws_mock.py. Для насмешки над KMS я получил несколько предопределенных ответов от живого клиента boto3.

from unittest.mock import MagicMock

import boto3
from moto import mock_s3

# 'create_key' response
create_resp = { ... }

# 'generate_data_key' response
generate_resp = { ... }

# 'decrypt' response
decrypt_resp = { ... }

def client(*args, **kwargs):
    if args[0] == 's3':
        s3_mock = mock_s3()
        s3_mock.start()
        mock_client = boto3.client(*args, **kwargs)

    else:
        mock_client = boto3.client(*args, **kwargs)

        if args[0] == 'kms':
            mock_client.create_key = MagicMock(return_value=create_resp)
            mock_client.generate_data_key = MagicMock(return_value=generate_resp)
            mock_client.decrypt = MagicMock(return_value=decrypt_resp)

    return mock_client

Второй - это настоящий тестовый модуль. Давайте назовем это test_my_module.py. Я опустил код my_module. А также функции, которые находятся под тестом. Позвольте вызвать эти функции foo, bar.

from unittest.mock import patch

import aws_mock
import my_module

@patch('my_module.boto3')
def test_my_module(boto3):
    # Some prep work for the mock mode
    boto3.client = aws_mock.client

    conn = boto3.client('s3')
    conn.create_bucket(Bucket='my-bucket')

    # Actual testing
    resp = my_module.foo()
    assert(resp == 'Valid')

    resp = my_module.bar()
    assert(resp != 'Not Valid')

    # Etc, etc, etc...

Еще одна вещь, не уверен, что это исправлено, но я обнаружил, что moto не был счастлив, если вы не установите некоторые переменные среды, такие как учетные данные и регион. Они не должны быть действительными учетными данными, но они должны быть установлены. Существует вероятность того, что это может быть исправлено к тому времени, когда вы прочитаете это! Но вот код на случай, если он вам понадобится, код оболочки на этот раз!

export AWS_ACCESS_KEY_ID='foo'
export AWS_SECRET_ACCESS_KEY='bar'
export AWS_DEFAULT_REGION='us-east-1'

Я знаю, что это, вероятно, не самый красивый кусок кода, но если вы ищете что-то универсальное, оно должно работать довольно хорошо!

Ответ 6

Вот мое решение для исправления клиента boto, используемого в недрах моего проекта, с помощью приспособлений pytest. Я использую только "mturk" в своем проекте.

Уловка для меня заключалась в том, чтобы создать свой собственный клиент, а затем установить исправление boto3.client с функцией, которая возвращает этого предварительно созданного клиента.

@pytest.fixture(scope='session')
def patched_boto_client():
    my_client = boto3.client('mturk')

    def my_client_func(*args, **kwargs):
        return my_client

    with patch('bowels.of.project.other_module.boto3.client', my_client_func):
        yield my_client_func


def test_create_hit(patched_boto_client):    
    client = patched_boto_client()
    stubber = Stubber(client)
    stubber.add_response('create_hit_type', {'my_response':'is_great'})
    stubber.add_response('create_hit_with_hit_type', {'my_other_response':'is_greater'})
    stubber.activate()

    import bowels.of.project # this module imports 'other_module'
    bowels.of.project.create_hit_function_that_calls_a_function_in_other_module_which_invokes_boto3_dot_client_at_some_point()

Я также определяю другое устройство, которое устанавливает фиктивные кредиты aws, чтобы бот не случайно получил какой-то другой набор учетных данных в системе. Я буквально установил "foo" и "bar" в качестве кредитов для тестирования - это не редактирование.

Важно, чтобы AWS_PROFILE env был не установлен, потому что в противном случае бот будет искать этот профиль.

@pytest.fixture(scope='session')
def setup_env():
    os.environ['AWS_ACCESS_KEY_ID'] = 'foo'
    os.environ['AWS_SECRET_ACCESS_KEY'] = 'bar'
    os.environ.pop('AWS_PROFILE', None)

А затем я указываю setup_env как запись pytest usefixtures, чтобы она использовалась для каждого запуска теста.

Ответ 7

Если вы не хотите использовать moto или заглушку botocore (кажется, что заглушка не препятствует выполнению HTTP-запросов к конечным точкам API AWS), вы можете использовать более подробный метод unittest.mock:

foo/bar.py

import boto3

def my_bar_function():
    client = boto3.client('s3')
    buckets = client.list_buckets()
    ...

bar_test.py

import unittest
from unittest import mock


class MyTest(unittest.TestCase):

     @mock.patch('foo.bar.boto3.client')
     def test_that_bar_works(self, mock_s3_client):
         self.assertTrue(mock_s3_client.return_value.list_buckets.call_count == 1)