python - 用 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 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(...)
          .....

不能给出一个有效的例子，因为不确定完整的逻辑是什么。看看 this 帖子。