模拟 boto3 S3 客户端方法 Python
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/37143597/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me):
StackOverFlow
Mocking boto3 S3 client method Python
提问by ptimson
I'm trying to mock a singluar method from the boto3 s3 client object to throw an exception. But I need all other methods for this class to work as normal.
我正在尝试从 boto3 s3 客户端对象模拟一个单一的方法来抛出异常。但是我需要这个类的所有其他方法才能正常工作。
This is so I can test a singular Exception test when and error occurs performing a upload_part_copy
这样我就可以在执行upload_part_copy并且发生错误时测试单个异常测试
1st Attempt
第一次尝试
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()
However this gives the following error:
但是，这会导致以下错误：
ImportError: No module named S3
2nd Attempt
第二次尝试
After looking at the botocore.client.py source code I found that it is doing something clever and the method upload_part_copydoes not exist. I found that it seems to call BaseClient._make_api_callinstead so I tried to mock that
在查看 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()
This throws an exception... but on the get_objectwhich I want to avoid.
这引发了一个异常......但get_object我想避免。
Any ideas about how I can only throw the exception on the upload_part_copymethod?
关于我如何只能在upload_part_copy方法上抛出异常的任何想法？
采纳答案by ptimson
As soon as I posted on here I managed to come up with a solution. Here it is hope it helps :)
我一在这里发帖，我就设法想出了一个解决方案。在这里希望它有帮助:)
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 solutionusing the the botocore.stub.Stubberclass. Whilst a cleaner solution I was un-able to mock specific operations.
Jordan Philips 还使用botocore.stub.Stubber类发布了一个很好的解决方案。虽然是一个更简洁的解决方案，但我无法模拟特定的操作。
回答by Jordon Phillips
Botocore has a client stubber you can use for just this purpose: docs.
Botocore 有一个客户端 stubber，您可以将其用于此目的：docs。
Here's an example of putting an error in:
下面是一个输入错误的例子：
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.
这是一个放入正常响应的示例。此外，现在可以在上下文中使用存根。需要注意的是，stubber 将尽可能验证您提供的响应是否与服务实际返回的内容相匹配。这并不完美，但它可以保护您免于插入完全无意义的响应。
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
回答by Aidan Melen
Here's an example of a simple python unittest that can be used to fake client = boto3.client('ec2')api call...
这是一个简单的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'
}
hopefully that helps.
希望这有帮助。
回答by wikier
回答by c4urself
If you don't want to use either motoor the botocore stubber (the stubber does notprevent HTTP requests being made to AWS API endpoints it seems), you can use the more verbose unittest.mock way:
如果您不想使用moto或 botocore stubber（stubber 似乎不会阻止向 AWS API 端点发出 HTTP 请求），您可以使用更详细的 unittest.mock 方式：
foo/bar.py
foo/bar.py
import boto3
def my_bar_function():
client = boto3.client('s3')
buckets = client.list_buckets()
...
bar_test.py
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)
回答by Barmaley
I had to mock boto3client for some integration testing and it was a bit painful! The problem that I had is that motodoes not support KMSvery well, yet I did not want to rewrite my own mock for the S3buckets. So I created this morph of all of the answers. Also it works globally which is pretty cool!
我不得不模拟boto3客户端进行一些集成测试，这有点痛苦！我遇到的问题是moto支持KMS不太好，但我不想为S3桶重写我自己的模拟。所以我创建了所有答案的这种变形。它也可以在全球范围内使用，这非常酷！
I have it setup with 2 files.
我用 2 个文件设置了它。
First one is aws_mock.py. For the KMSmocking I got some predefined responses that came from live boto3client.
第一个是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
Second one is the actual test module. Let's call it test_my_module.py. I've omitted the code of my_module. As well as functions that are under the test. Let's call those foo, barfunctions.
第二个是实际的测试模块。让我们称之为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...
One more thing, not sure if that is fixed but I found out that motowas not happy unless you set some environmental variables like credentials and region. They don't have to be actual credentials but they do need to be set. There is a chance it might be fixed by the time you read this! But here is some code in case you do need it, shell code this time!
还有一件事，不确定这是否已修复，但我发现moto除非您设置一些环境变量（如凭据和区域），否则这并不令人满意。它们不必是实际凭据，但确实需要设置。当您阅读本文时，它有可能已修复！但这里有一些代码，以防你需要它，这次是 shell 代码！
export AWS_ACCESS_KEY_ID='foo'
export AWS_SECRET_ACCESS_KEY='bar'
export AWS_DEFAULT_REGION='us-east-1'
I know it is probably not the prettiest piece of code but if you are looking for something universal it should work pretty well!
我知道这可能不是最漂亮的一段代码，但是如果您正在寻找通用的东西，它应该可以很好地工作！
回答by deargle
Here is my solution for patching a boto client used in the bowels of my project, with pytestfixtures. I'm only using 'mturk' in my project.
这是我的解决方案，用于使用pytest固定装置修补我项目内部使用的 boto 客户端。我只在我的项目中使用“mturk”。
The trick for me was to create my own client, and then patch boto3.clientwith a function that returns that pre-created client.
对我来说，诀窍是创建我自己的客户端，然后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()
I also define another fixture that sets up dummy aws creds so that boto doesn't accidentally pick up some other set of credentials on the system. I literally set 'foo' and 'bar' as my creds for testing -- that's not a redaction.
我还定义了另一个设置虚拟 aws 凭据的装置，以便 boto 不会意外地在系统上获取其他一些凭据。我确实将 'foo' 和 'bar' 设置为我的测试凭证——这不是修订。
It's important that AWS_PROFILEenv be unset because otherwise boto will go looking for that profile.
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)
And then I specify setup_envas a pytest usefixturesentry so that it gets used for every test run.
然后我指定setup_env为 pytestusefixtures条目，以便它用于每次测试运行。
