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

Final codes

Again mentioning the GitHub repository:

https://github.com/melon-ruet/lambda-testing-localstack