单元测试从s3下载json文件的函数

Question

我有一个函数如下:

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

你知道我在这里可能做错了什么吗?

Answer 1

这确实不美观，但创建一些模拟类将解决问题:

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使它更优雅。

Answer 2

我稍微改变了我的函数，以达到相同的结果:

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