I am trying to mock a lambda invocation using pytest and moto. Below is my working code.
import zipfile
import os
import boto3
from moto import mock_lambda,mock_iam
import pytest
import io
from botocore.exceptions import ClientError
import json
@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'
@pytest.fixture(scope='function')
def lambda_mock(aws_credentials):
with mock_lambda():
yield boto3.client('lambda', region_name='us-east-1')
def get_role_name():
with mock_iam():
iam = boto3.client("iam", region_name='us-east-1')
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"]
@pytest.fixture(scope='function')
def lambda_mock(aws_credentials):
with mock_lambda():
yield boto3.client('lambda', region_name='us-east-1')
def _process_lambda(func_str):
zip_output = io.BytesIO()
zip_file = zipfile.ZipFile(zip_output, "w", zipfile.ZIP_DEFLATED)
zip_file.writestr("lambda_function.py", func_str)
zip_file.close()
zip_output.seek(0)
return zip_output.read()
def get_test_zip_file1():
pfunc = """
def lambda_handler(event, context):
print("custom log event")
return "hello world"
"""
return _process_lambda(pfunc)
# create mocked lambda with zip file
@mock_lambda
def test_mock_some_lambda(lambda_mock):
lambda_mock.create_function(
FunctionName='abc',
Runtime='python3.7',
Role=get_role_name(),
Handler='lambda_function.lambda_handler',
Code={"ZipFile": get_test_zip_file1()},
Publish=True,
Timeout=30,
MemorySize=128
)
resp=lambda_mock.invoke(FunctionName='abc')
response=json.load(resp['Payload'])
assert response=='hello world'
However, I want to separate lambda mocking and my testing function. So I created a pytest fixture which creates a lambda function like below.
@pytest.fixture(scope='function')
def mock_some_lambda(lambda_mock):
yield lambda_mock.create_function(
FunctionName='abc',
Runtime='python3.7',
Role=get_role_name(),
Handler='lambda_function.lambda_handler',
Code={"ZipFile": get_test_zip_file1()},
Publish=True,
Timeout=30,
MemorySize=128
)
And then a new test function which uses this fixture
@mock_lambda
def test_update_allowed(lambda_mock,mock_some_lambda):
resp=lambda_mock.invoke(FunctionName='abc')
response=json.load(resp['Payload'])
assert response=='hello world'
This throws botocore.exceptions.ClientError: An error occurred (404) when calling the Invoke operation:
Looks like the lambda function creation is not sticking in the test function and when I run invoke(), it's not able to find the function.
How do I make this work ?