模拟 AWS lambda 和 boto3

Question

我有一个 lambda (python)，它返回客户资料的 json 表示。它通过从顶级帐户 json 开始，然后读取链接的 json 文件直到它用完链接来做到这一点。从 s3 读取的函数是递归的，但递归永远只有一层深。

这是从给定键实际获取json内容的方法。（桶是已知的）

def get_index_from_s3(key):
    try:
        response = s3.get_object(
            Bucket=bucket,
            Key=key
        )
        body = response.get('Body')
        content = body.read().decode('utf-8')
    except ClientError as ex:
        # print 'EXCEPTION MESSAGE: {}'.format(ex.response['Error']['Code'])
        content = '{}'

    message = json.loads(content)
    return message

该代码返回在指定键处找到的 json，或者在 get_object 由于 ClientError （这是 NoSuchKey 的结果）而失败的情况下返回一个空字典。

我已经测试过了，它有效。对该函数的第一次调用会获取一大块 json。解析 json，找到链接，进行第二次调用，然后构建配置文件。如果我删除链接键上的对象，我只会得到一个默认的空表示，正如预期的那样。

我的问题来自测试这个。我写了几个测试类，每个类都有一个排列方法，它们共享一个行为方法。

对于我的幸福之路，我使用以下安排：

def arrange(self):
    super(WhenCognitoAndNerfFoundTestCase, self).arrange()
    # self.s3_response = self.s3.get_object.return_value
    self.s3_body = self.s3.get_object.return_value.get.return_value
    self.s3_body.read.return_value.decode.side_effect = [
        self.cognito_content,
        self.nerf_content]
    signed_url = "https://this.is/a/signed/url/index.html"
    self.s3.generate_presigned_url.return_value = signed_url

这正是我想要的。s3_response是get_object的return_value，它有get返回的Body属性，后续读取的值返回一个json字符串。我使用 side_effect 设置为 json 字符串列表，以便我可以在每次调用时返回不同的字符串（只有两个） content = body.read().decode('utf-8')

但是当我想测试第二个存储桶中缺少内容的情况时，我遇到了困难。我目前对这种安排的尝试如下：

def arrange(self):
    super(WhenCognitoOnlyFoundTestCase, self).arrange()
    # self.s3_response = MagicMock()
    # botocore.response.StreamingBody
    self.s3.get_object.side_effect = [{},
                                      ClientError]
    # self.s3_response = self.s3.get_object.return_value
    self.s3_body = self.s3.get_object.return_value.get.return_value
    self.s3_body.read.return_value.decode.return_value = \
        self.cognito_content

运行测试结果如下：

    def get_index_from_s3(key):
        try:
            response  = s3.get_object(
                Bucket=bucket,
                Key=key
            )
            body = response.get('Body')
>           content = body.read().decode('utf-8')
E            AttributeError: 'NoneType' object has no attribute 'read'

master_profile.py:66: AttributeError

这是有道理的，因为 read 方法位于 s3.get_object 响应的 Body 属性上，在这种情况下为 None 。

所以我的问题是，我如何模拟这个东西以便我可以测试它？模拟 get_object 的响应的困难在于，虽然它只是一个字典，但 Body 属性是一个botocore.response.StreamingBody我不知道如何模拟的。

Answer 1

根据经验，您的目标应该是让您的问题自成一体。为了说明您做错的一些事情，我稍微修改了您的初始函数以使其独立。

假设s3_module我们要测试的定义如下：

import boto3
from botocore.exceptions import ClientError
import json

s3 = boto3.client('s3')

def get_index_from_s3(key):
    try:
        response = s3.get_object(
            Bucket='bucket',
            Key=key
        )
        body = response.get('Body')
        content = body.read().decode('utf-8')
    except ClientError as ex:
        import ipdb; ipdb.set_trace()
        # print 'EXCEPTION MESSAGE: {}'.format(ex.response['Error']['Code'])
        content = '{}'

    message = json.loads(content)
    return message

为了测试它，我们可以编写另一个s3_test具有类似测试的模块：

import pytest
from unittest.mock import patch, Mock, MagicMock
from botocore.exceptions import ClientError
import json

from s3_module import get_index_from_s3


@patch('s3_module.s3.get_object')
def test_get_index_from_s3(s3_get_mock):

    body_mock = Mock()
    body_mock.read.return_value.decode.return_value = json.dumps('first_response')
    s3_get_mock.side_effect = [{'Body': body_mock}, ClientError(MagicMock(), MagicMock())]

    first_response = get_index_from_s3('key1')
    assert  first_response == 'first_response'
    second_response = get_index_from_s3('key2')
    assert  second_response == {}

与您的解决方案相比，您缺少一些要点：

• self.s3.get_object.side_effect应该为第一个响应返回一个对象，该对象与您的其余代码一起使用，即包含Body内容可以是的键的字典read()，decoded()并由json.load()

• self.s3.get_object.side_effect应该返回ClientError为第二个响应正确构造的异常

您可以ClientError在 botocore 文档中查看有关如何构建异常的更多信息：http ://botocore.readthedocs.io/en/latest/client_upgrades.html#error-handling

您可以在文档中找到有关修补和模拟的更多信息：https : //docs.python.org/3/library/unittest.mock.html。

通常关于在哪里打补丁的部分非常有用：https : //docs.python.org/3/library/unittest.mock.html#where-to-patch