模仿 boto3 S3客户端方法 Python

我试图从 boto3s3客户机对象模拟一个单一方法来抛出异常。但是我需要所有其他方法使这个类能够正常工作。

这样，当执行 上传 _ 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方法上的异常有什么想法吗？

141405 次浏览
小开
最佳答案

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

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 using the the Botocore Stubber Stubber class. Whilst a cleaner solution I was un-able to mock specific operations.

小开

Botocore 有一个客户端存根，您可以仅用于此目的: 医生。

下面是一个输入错误的例子:

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

Here's an example of putting a normal response in. Additionally, the stubber can now be used in a context. It's important to note that the stubber will verify, so far as it is able, that your provided response matches what the service will actually return. This isn't perfect, but it will protect you from inserting total nonsense responses.

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
小开

下面是一个简单的 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.describe_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'
}

希望能有所帮助。

小开

What about simply using 莫托?

It comes with a very handy 装潢师:

from moto import mock_s3


@mock_s3
def test_my_model_save():
pass
小开

如果你不想使用 moto或 botocore 存根(这个存根可以防止 HTTP 请求被发送到 AWS API 端点) ，你可以使用更详细的 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)
小开

我不得不嘲笑 boto3客户端的一些集成测试，这是一个有点痛苦！我遇到的问题是 moto不能很好地支持 KMS，但是我不想为 S3存储桶重写我自己的 mock。所以我创建了这个所有答案的变体。它也可以在全球范围内工作，这是非常酷的！

我有它设置了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'

我知道这可能不是最漂亮的一段代码，但如果你正在寻找一些通用的东西，它应该工作得很好！

小开

下面是我的解决方案，用 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()

我还定义了另一个 fixture，它设置虚拟证书，这样 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条目，以便在每次运行测试时使用它。


提交答案