我正在尝试从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方法上抛出异常的任何想法？
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
我一发布在这里，我设法找到了解决方案。这是希望它有帮助:)
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类。虽然是一个更清洁的解决方案，但我无法模拟特定的操作。
这是一个简单的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_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'
}
希望这会有所帮助。
我不得不嘲笑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'
我知道它可能不是最漂亮的代码，但如果你正在寻找通用的东西，它应该可以很好地工作！
这是我用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条目，以便它可以用于每次测试运行。
如果您不想使用moto或botocore stubber（stubber不会阻止调用似乎是对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)