使用moto模拟lambda响应

Question

在我的代码中，调用了一个lambda来返回true/false响应。我试图在我的单元测试中模拟这个lambda，但没有成功。

这是我的代码：

def _test_update_allowed():
    old = ...
    new = ...
    assert(is_update_allowed(old, new) == True)

在内部，is_update_allowed调用了lambda，这就是我想要模拟的。

我尝试在我的测试之上添加以下代码：

import zipfile
import io
import boto3
import os

@pytest.fixture(scope='function')
def aws_credentials():
    """Mocked AWS Credentials for moto."""
    os.environ['AWS_ACCESS_KEY_ID'] = 'testing'
    os.environ['AWS_SECRET_ACCESS_KEY'] = 'testing'
    os.environ['AWS_SECURITY_TOKEN'] = 'testing'
    os.environ['AWS_SESSION_TOKEN'] = 'testing'


CLIENT = boto3.client('lambda', region_name='us-east-1')

# Expected response setup and zip file for lambda mock creation
def lambda_event():
    code = '''
        def lambda_handler(event, context):
            return event
        '''
    zip_output = io.BytesIO()
    zip_file = zipfile.ZipFile(zip_output, 'w', zipfile.ZIP_DEFLATED)
    zip_file.writestr('lambda_function.py', code)
    zip_file.close()
    zip_output.seek(0)
    return zip_output.read()

# create mocked lambda with zip file
def mock_some_lambda(lambda_name, return_event):
    return CLIENT.create_function(
        FunctionName=lambda_name,
        Runtime='python2.7',
        Role='arn:aws:iam::123456789:role/does-not-exist',
        Handler='lambda_function.lambda_handler',
        Code={
            'ZipFile': return_event,
        },
        Publish=True,
        Timeout=30,
        MemorySize=128
    )

然后将我的测试更新为：

@mock_lambda
def _test_update_allowed():
    mock_some_lambda('hello-world-lambda', lambda_event())
    old = ...
    new = ...
    assert(is_update_allowed(old, new) == True)

但我收到了以下错误，这让我认为它实际上是在尝试与AWS对话

botocore.exceptions.ClientError: An error occurred (UnrecognizedClientException) when calling the CreateFunction operation: The security token included in the request is invalid.

Answer 1

从错误消息中，我可以确认这绝对不是AWS问题。它清楚地声明它正在尝试使用一些无效的凭据。因此，这可以归结为代码。

我假设您已经有了必要的库的import语句，因为这些语句在共享代码中也是不可见的

import pytest
import moto

from mock import mock, patch
from moto import mock_lambda

因此您需要使用

def aws_credentials():
 .....

在创建客户端时，因为从代码中我看不到您正在使用相同的。

@pytest.fixture(scope='function')
def lambda_mock(aws_credentials):
with mock_lambda():
    yield boto3.client('lambda', region_name='us-east-1')

最终你的嘲笑

@pytest.fixture(scope='function')
def mock_some_lambda(lambda_mock):
    lambda_mock.create_function(
        FunctionName=lambda_name,
        Runtime='python2.7',
        Role='arn:aws:iam::123456789:role/does-not-exist',
        Handler='lambda_function.lambda_handler',
        Code={
            'ZipFile': return_event,
        },
        Publish=True,
        Timeout=30,
        MemorySize=128
    )
 yield

然后测试函数

    def _test_update_allowed(lambda_mock,mock_some_lambda):
        lambda_mock.invoke(...)
          .....

不能给出一个有效的例子，因为不确定完整的逻辑是什么。在take a look this帖子之间。