A journey to AWS Lambda integration testing with Python, Localstack, and Terraform
To start with unit testing for AWS Lambda, we can just choose a testing framework and start writing our beautiful unit tests. For testing AWS services we also can use some mock tools like moto. But what about integration testing? A solution may be to deploy with the Continuous Deployment tool and run some test code against real AWS services. But there are some problems:
- It takes more time to deploy every time
- Running test code against AWS takes more time
- Increases the AWS bills
So one good solution could be using a local infrastructure that resembles the same as AWS real infrastructure. So here comes our friend Localstack to solve the problem. And today we will take a look at how we can use Localstack for integration testing of AWS Lambda.
Prerequisite
- Python for lambda code
- Python for test code
- Pipenv to manage python virtual environment
- Terraform, an IaC tool to deploy lambda at the Localstack
- Localstack to use AWS infrastructure locally
Background
In one of my medium posts, the way of deploying python codes at Lambda with python packages at the lambda layer was described. So here we will not describe that part here. Also as Lambda Layer is a part of the pro version of Localstack, we will avoid the lambda layer. The only change will be rather deploying to real infrastructure, we will deploy to Localstack. So let’s start with sharing the end result GitHub repository first 😄 🤫
The skeleton
Let’s start with installing pipenv from here. Then we will install terraform from here.
# Create project directory
mkdir lambda-testing-localstack
cd lambda-testing-localstack# Create directory for lambda codes
mkdir lambda# Add handler.py file in lambda directory
touch lambda/handler.py# Create directory for terraform
mkdir terraform# Add Pipfile in project root directory
touch Pipfile# Create directory for test code
mkdir tests# Create test code python file
touch tests/test_lambda.py# So the skeleton will look like this
tree
.
├── lambda
│ └── handler.py
├── Pipfile
├── terraform
└── tests
└── test_lambda.py
Add requirements
Let’s add some python requirements and python version 3.8 in Pipfile
[[source]]
name = "pypi"
url = "https://pypi.org/simple"
verify_ssl = true
[dev-packages]
localstack = "==0.11.3.3"
pipenv = "==2020.8.13"
# Below packages are required to start localstack from python code
h11 = "==0.10.0"
amazon_kclpy = "==2.0.1"
quart = "==0.13.0"
flask_cors = "==3.0.8"
moto = "==1.3.14"
moto-ext = ">=1.3.15.12"
crontab = "==0.22.9"
pyOpenSSL = "==19.1.0"
[requires]
python_version = "3.8"As Localstack will not be in with lambda codes, so it is added in dev-packages section. Run
pipenv install --devto initiate a virtual environment and install packages with pipenv
Add Lambda Code
We will add a very simple code. Our lambda will just get the webpage https://example.com and return the page source as text.
# lambda/lambda_handler.pyimport logging
import urllib.request
LOGGER = logging.getLogger()
LOGGER.setLevel(logging.INFO)
def lambda_handler(event, context):
with urllib.request.urlopen('https://example.com') as response:
html = response.read().decode('utf-8')
LOGGER.info(html)
return html
Add Files to deploy Lambda
Let’s add a terraform variable for the lambda function name. So from our test code, we will provide the lambda function name and then test it to make the testing more dynamic.
# terraform/vars.tfvariable "lambda_function_name" {
type = string
default = "test-lambda-function"
}
Now add terraform code for lambda at terraform/lambda.tf
# terraform/lambda.tf// Zip lambda function codes
data "archive_file" "lambda_zip_file" {
output_path = "${path.module}/lambda_zip/lambda.zip"
source_dir = "${path.module}/../lambda"
excludes = ["__pycache__", "*.pyc"]
type = "zip"
}
// IAM Policy document for lambda assume role
data "aws_iam_policy_document" "lambda_assume_role" {
version = "2012-10-17"
statement {
sid = "LambdaAssumeRole"
effect = "Allow"
actions = [
"sts:AssumeRole"
]
principals {
identifiers = [
"lambda.amazonaws.com"
]
type = "Service"
}
}
}
// Lambda IAM role
resource "aws_iam_role" "lambda_role" {
name = "test-lambda-role"
assume_role_policy = data.aws_iam_policy_document.lambda_assume_role.json
lifecycle {
create_before_destroy = true
}
}
// Lambda function terraform code
resource "aws_lambda_function" "lambda_function" {
function_name = var.lambda_function_name
filename = data.archive_file.lambda_zip_file.output_path
source_code_hash = data.archive_file.lambda_zip_file.output_base64sha256
handler = "handler.lambda_handler"
role = aws_iam_role.lambda_role.arn
runtime = "python3.8"
lifecycle {
create_before_destroy = true
}
}
Now we are going to add terraform/localstack.tf to which will tell terraform to use Localstack for deployment
# terraform/localstack.tfprovider "aws" {
region = "eu-west-1"
access_key = "fakekey"
secret_key = "fakekey"
skip_credentials_validation = true
skip_metadata_api_check = true
skip_requesting_account_id = true
s3_force_path_style = true
endpoints {
lambda = "http://localhost:4566"
iam = "http://localhost:4566"
}
}
Deploy to Localstack with Terraform
Now time to test deployment at LocalStack
# Start localstack service only for Lambda and IAM
SERVICES=lambda,iam localstack start# In another terminal
# Terraform initialization
terraform init# Plan to build
terraform plan# Deploy all
terraform apply -auto-approve# Destroy all because we will make it automated as we
# are going to implement automated test
terraform destroy -auto-approve# Kill the localstack service using Ctrl + C in 1st terminal
Automate Test Code
Now time to automate the lambda testing. Our strategy:
- Deploy terraform to LocalStack from test code e.g python code
- Execute lambda using boto3
- Check if we get the webpage as text
- Destroy infrastructure with terraform to LocalStack using test code
Now add terraform helper function at tests/terraform_helper.py which will responsible for creating infrastructure at test start and destroy after the test
# tests/terraform_helper.pyimport subprocess
import os
TERRAFORM_DIR_PATH = os.path.dirname(os.path.realpath(__file__)) + "/../terraform/"
def terraform_init():
"""Terraform init command"""
tf_init = ["terraform", "init", TERRAFORM_DIR_PATH]
subprocess.check_call(tf_init)
def create_resources():
"""Create a tf resource."""
proc = subprocess.Popen("terraform apply -auto-approve " + TERRAFORM_DIR_PATH, shell=True)
proc.wait()
def destroy_resources():
"""Destroy all tf resources.
This method will destroy any resources it can find in the state file,
and delete all resources from the state file.
"""
tf_destroy = [
"terraform",
"destroy",
"-auto-approve",
TERRAFORM_DIR_PATH
]
subprocess.call(tf_destroy)
tf_refresh = [
"terraform",
"refresh",
TERRAFORM_DIR_PATH
]
subprocess.call(tf_refresh)
def terraform_start():
""" teardown and create resources at the beginning of feature test """
terraform_init()
destroy_resources()
return create_resources()
Let’s add test code and we are going to use python’s unit test module to test our code
# tests/test_lambda.pyimport os
import unittest
import boto3
from localstack.services import infra
from tests import terraform_helper
class AWSLambdaTest(unittest.TestCase):
localstack_endpoint = 'http://localhost:4566'
lambda_function_name = 'dynamic-test-lambda-function'
def set_tf_var(self):
os.environ["TF_VAR_lambda_function_name"] = self.lambda_function_name
def setUp(self):
# Start localstack
infra.start_infra(apis=['lambda', 'iam', 'cloudwatch'], asynchronous=True)
self.set_tf_var()
terraform_helper.terraform_start()
def test_lambda_response(self):
client = boto3.client(
service_name='lambda',
endpoint_url=self.localstack_endpoint
)
response = client.invoke(
FunctionName=self.lambda_function_name,
InvocationType='RequestResponse'
)
assert response['StatusCode'] == 200
assert response['Payload']
html = response['Payload'].read().decode('utf-8')
# Check if "Example Domain" text exists in example.com
assert 'Example Domain' in html
def tearDown(self):
terraform_helper.destroy_resources()
# Stop localstack
infra.stop_infra()
if __name__ == '__main__':
unittest.main()
Finally, the project will look like this
tree
.
├── lambda
│ └── handler.py
├── Pipfile
├── Pipfile.lock
├── terraform
│ ├── lambda.tf
│ ├── localstack.tf
│ └── vars.tf
└── tests
├── __init__.py
├── terraform_helper.py
└── test_lambda.pyTime to test the Test
Now come to the moment of playing 🎻
As we implemented the tests with the unit test module, we can run the test using
pipenv shell
python -m unittestAfter a lot of logs we can see final output like this
.
-------------------------------------------------------
Ran 1 test in 36.943sOK
⚠️ WARNING
Localstack is growing very faster. So if it works today may break tomorrow if the required packages are not fixed. So please try to fix all the packages version in Pipfile. Sometimes it may require some python packages to start various services like DynamoDB or StepFuntions. Add those at Pipfile accordingly.
Final codes
Again mentioning the GitHub repository:
