模拟 boto3 S3 客户端方法 Python
Mocking boto3 S3 client method Python
我正在尝试从 boto3 s3 客户端对象模拟一个单一方法来抛出异常。但是我需要所有其他方法才能让这个 class 正常工作。
这样我就可以在执行 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 方法上抛出异常的任何想法？
我一在这里发帖，就想出了一个解决方案。希望对您有所帮助:)
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()
using the the botocore.stub.Stubberclass。虽然是一个更简洁的解决方案，但我无法模拟特定操作。
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()
这是一个放入正常响应的示例。此外，stubber 现在可以在上下文中使用。请务必注意，stubber 将尽可能验证您提供的响应是否与服务实际 return 匹配。这并不完美，但可以防止您插入完全无意义的回复。
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'
}
希望对您有所帮助。
如果只使用 moto 呢？
它带有一个非常方便的 decorator:
from moto import mock_s3
@mock_s3
def test_my_model_save():
pass
如果您不想使用 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)
我不得不模拟 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'。
我的诀窍是创建我自己的客户端，然后用 returns pre-created 客户端的函数修补 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 creds 的固定装置，这样 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 条目，以便它用于每个测试 运行.
我正在尝试从 boto3 s3 客户端对象模拟一个单一方法来抛出异常。但是我需要所有其他方法才能让这个 class 正常工作。
这样我就可以在执行 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 方法上抛出异常的任何想法？
我一在这里发帖，就想出了一个解决方案。希望对您有所帮助:)
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()
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()
这是一个放入正常响应的示例。此外，stubber 现在可以在上下文中使用。请务必注意，stubber 将尽可能验证您提供的响应是否与服务实际 return 匹配。这并不完美，但可以防止您插入完全无意义的回复。
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'
}
希望对您有所帮助。
如果只使用 moto 呢？
它带有一个非常方便的 decorator:
from moto import mock_s3
@mock_s3
def test_my_model_save():
pass
如果您不想使用 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)
我不得不模拟 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'。
我的诀窍是创建我自己的客户端，然后用 returns pre-created 客户端的函数修补 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 creds 的固定装置，这样 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 条目，以便它用于每个测试 运行.