模拟boto3 S3客户端方法Python

Question

我正在尝试从 boto3 s3 客户端对象模拟一个单一方法以引发异常。但我需要所有其他方法才能让这个类正常工作。

这样我就可以在执行upload_part_copy 时发生错误时测试单个异常测试

第一次尝试

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

第二次尝试

查看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 方法上抛出异常有什么想法吗？

Answer 1

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

Answer 2

我一在这里发帖，就想出了一个解决方案。希望对您有所帮助:)

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()

Jordan Philips also posted a great solution 使用 botocore.stub.Stubber 类。虽然更清洁的解决方案我无法模拟特定操作。

Answer 3

这是一个简单的python单元测试示例，可用于伪造client = boto3.client('ec2') api调用...

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_describe_tags.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'
}

希望对您有所帮助。

Answer 4

如果您不想使用 moto 或 botocore 存根（存根确实不阻止向 AWS API 端点发出 HTTP 请求），您可以使用更详细的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)

Answer 5

简单地使用moto怎么样？

它带有一个非常方便的decorator：

from moto import mock_s3

@mock_s3
def test_my_model_save():
    pass

Answer 6

我不得不模拟 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 并不满意，除非您设置一些环境变量，例如凭据和区域。它们不必是实际凭据，但确实需要设置。当您阅读本文时，它可能会修复！但是这里有一些代码以备不时之需，这次是 shell 代码！

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

我知道它可能不是最漂亮的代码，但如果您正在寻找通用的东西，它应该会很好用！

Answer 7

这是我使用pytest 固定装置修补项目内部使用的boto 客户端的解决方案。我只在我的项目中使用“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 凭据的装置，这样 boto 就不会意外获取系统上的其他一组凭据。我确实将 'foo' 和 'bar' 设置为我的测试凭据——这不是修订。

取消设置 AWS_PROFILE env 很重要，否则 boto 会寻找该配置文件。

@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 条目，以便每次测试运行都使用它。