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

无法给出一个工作示例，因为不确定完整的逻辑是什么。请看一下这篇文章。

Answer 2

这些问题似乎是由于不存在的arn角色造成的。试着像在moto库测试中一样嘲笑它

def get_role_name():
with mock_iam():
iam = boto3.client("iam", region_name=_lambda_region)
try:
return iam.get_role(RoleName="my-role")["Role"]["Arn"]
except ClientError:
return iam.create_role(
RoleName="my-role",
AssumeRolePolicyDocument="some policy",
Path="/my-path/",
)["Role"]["Arn"]