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

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

Unit test a function which downloads json file from s3

python

nose

python-3.x