如何解决单元测试从 s3 下载 json 文件的函数
def read_json(bucket,key):
"""
:param bucket:
:param key:
:return:
"""
s3 = boto3.resource('s3')
content_object = S3.Object(bucket,key)
file_content = content_object.get()['Body'].read().decode('utf-8')
json_content = json.loads(file_content)
return json_content
我正在测试此功能,如下所示:
@mock.patch('boto3.resource')
def test_read_json(mock_resource):
mock_resource.Object.return_value.return_value = '{ "key":"value",' \
'"key_1":value_1,' \
'"key_2":"value_2"}'
json = helpers.read_json('bucket','key')
mock_resource.assert_called_once_with('s3')
tools.assert_is_instance(json,'json')
TypeError: JSON 对象必须是 str、bytes 或 bytearray,而不是 Magicmock
知道我在这里可能做错了什么吗?
解决方法
确实不好看,但是创建一些模拟类可以解决问题:
class MockBody():
def read(self):
return '{ "key": "value",' \
'"key_1": "value_1",' \
'"key_2": "value_2" }'.encode('utf-8')
class MockBoto3Resource():
class Object():
def __init__(self,bucket,key):
pass
def get(self):
return {'Body': MockBody()}
然后像这样在你的测试中使用它:
@mock.patch('boto3.resource')
def test_read_json(mock_resource):
mock_resource.return_value = MockBoto3Resource
[...]
您可能想看看 Stubber (https://botocore.amazonaws.com/v1/documentation/api/latest/reference/stubber.html),我个人从未使用过它,但您可以通过使用 botocore.stub.Stubber
使其更加优雅。
我稍微改变了我的函数以获得相同的结果:
def read_json(bucket,key):
"""
:param bucket:
:param key:
:return:
"""
try:
response = s3.get_object(bucket,key)
file_content = response.get('Body').read().decode('utf-8')
json_content = json.loads(file_content)
except ClientError as ex:
LOGGER.error("Received error: %s",ex)
raise
return json_content
对应的单元测试:
@mock.patch('delivery_qa.utils.helpers.s3.get_object')
def test_read_json(s3_get_mock):
body_mock = Mock()
body_mock.read.return_value.decode.return_value = json.dumps('response')
# When first time function would be called,it would return body_mock
s3_get_mock.side_effect = [{'Body': body_mock}]
first_call = helpers.read_json('bucket','key')
tools.assert_equal(first_call,'response')
@raises(ClientError)
@mock.patch('delivery_qa.utils.helpers.s3.get_object')
def test_read_json_exception(s3_get_mock):
body_mock = Mock()
body_mock.read.return_value.decode.return_value = json.dumps('response')
# When first time function would be called,it would return ClientError
s3_get_mock.side_effect = [ClientError(MagicMock(),MagicMock())]
helpers.read_json('bucket','key')
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。