AWS Lambda integration testing with Python, Localstack, and Terraform

To start with unit testing for AWS Lambda integration testing with Python, Localstack, and Terraform 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 --dev

to 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.tfto 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.dirnam(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.py

Time 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 unittest

After 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.