单元测试从 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')