Testing AWS Lambda locally using LocalStack and pytest

Testing loosely coupled microservices requires the use of diverse testing tools. Unlike monoliths, it is challenging to have a complete copy of the production system (System Under Test) for integration testing. Often, developers are hesitant to make changes in the pipeline due to uncertainty about potential downstream impacts. In such cases, it is valuable for developers to test integration components (beyond unit tests) on their local machines. We have found LocalStack is an excellent tool for emulating a local version of your cloud stack (e.g., AWS).

In this post, Qxf2 will provide a comprehensive, step-by-step guide to help you kickstart your journey with LocalStack and Lambdas. we specifically focus on this essential step as emulating serverless applications is typically a crucial aspect of LocalStack implementations.

A. Steps to get set with LocalStack:

Let us first get setup with LocalStack. The setup requires Python and Docker.

1. Install LocalStack and check version:

Install LocalStack using python pip package installer.

 python3 -m pip install localstack

Check version:

 localstack --version

2. Start LocalStack using docker-compose:

Follow the steps below to start LocalStack:

version: "3.8"
 
services:
  localstack:
    container_name: "${LOCALSTACK_DOCKER_NAME-localstack_main}"
    image: localstack/localstack
    ports:
      - "127.0.0.1:4566:4566"            # LocalStack Gateway
      - "127.0.0.1:4510-4559:4510-4559"  # external services port range
    environment:
      - DEBUG=${DEBUG-}
      - DOCKER_HOST=unix:///var/run/docker.sock
    volumes:
      - "${LOCALSTACK_VOLUME_DIR:-./volume}:/var/lib/localstack"
      - "/var/run/docker.sock:/var/run/docker.sock"

2.2 Pull docker image:
To start LocalStack, we need the LocalStack Docker image. Use the following command to pull the image:

docker pull localstack/localstack:latest

2.3 Start LocalStack:
Use the following command to start LocalStack in detached mode:

 docker-compose up -d

3. Verify LocalStack up or not:

Use the following command to check the status of LocalStack:

 localstack status

4. Verify LocalStack available services:

Use the following command to check the available services:

 localstack status services

B. Our URL Filter Lambda:

To write a test for the Lambda, we need to know what the Lambda does and its code. Let’s take a look at the following Lambda code. This code is designed to filter out the URL and accepts input from our internal SQS, which contains a message along with the URL.

Note: We have trimmed our URL-filtered Lambda code for this blog post and made it free from external dependencies. Here, we included unused package/import (requests) under requirements and code to show how the Lambda gets deployed along with its dependencies.

#lambda_with_dependency.py
"""
Lambda to to pull URL from ETC channel messages
And filter out URLs from message
"""
import json
import os
import re
import validators
import requests
 
EXCLUDE_URL_STRINGS = ['skype.com', 'meet.google.com', 'trello.com/b']
 
def clean_message(message):
    "Clean up the message received"
    message = message.replace("'", '-')
    message = message.replace('"', '-')
 
    return message
 
def get_message_contents(event):
    "Retrieve the message contents from the SQS event"
    record = event.get('Records')[0]
    message = record.get('body')
    message = json.loads(message)['Message']
    message = json.loads(message)
 
    return message
 
def get_url(message):
    "Get the URL from the message"
    regex = r"(?i)\b((?:https?://|www\d{0,3}[.]|[a-z0-9.\-]+[.][a-z]{2,4}/)"
    regex += r"(?:[^\s()<>]+|\(([^\s()<>]+|(\([^\s()<>]+\)))*\))"
    regex += r"(?:\(([^\s()<>]+|(\([^\s()<>]+\)))*\)|[^\s`!()\[\]{};:'\".,<>?«»“”‘’]))"
    url_patterns = re.findall(regex,message)
    urls = []
    for url in url_patterns:
        if url[0][-1] != '-':
            present_flag = False
            for exclude_url in EXCLUDE_URL_STRINGS:
                if exclude_url in url[0]:
                    present_flag = True
                    break
            if not present_flag and validators.url(url[0]):
                urls.append(url[0])
 
    return urls
 
def lambda_handler(event, context):
    """
    Method run when Lambda is triggered
    calls the filtering logic
    """
    content = get_message_contents(event)
    message = content['msg']
    channel = content['chat_id']
    user = content['user_id']
    print(f'{message}, {user}, {channel}')
 
    response=""
    final_url=[]
    if channel == os.environ.get('ETC_CHANNEL') and user != os.environ.get('Qxf2Bot_USER'):
        print("Getting message posted on ETC ")
        cleaned_message = clean_message(message)
        final_url=get_url(cleaned_message)
        #Filtered URL is printed by lambda
        print("Final url is :",final_url)
 
    return {
        'body': json.dumps(final_url)
    }

#lambda_with_dependency.py """ Lambda to to pull URL from ETC channel messages And filter out URLs from message """ import json import os import re import validators import requests EXCLUDE_URL_STRINGS = ['skype.com', 'meet.google.com', 'trello.com/b'] def clean_message(message): "Clean up the message received" message = message.replace("'", '-') message = message.replace('"', '-') return message def get_message_contents(event): "Retrieve the message contents from the SQS event" record = event.get('Records')[0] message = record.get('body') message = json.loads(message)['Message'] message = json.loads(message) return message def get_url(message): "Get the URL from the message" regex = r"(?i)\b((?:https?://|www\d{0,3}[.]|[a-z0-9.\-]+[.][a-z]{2,4}/)" regex += r"(?:[^\s()<>]+|\(([^\s()<>]+|(\([^\s()<>]+\)))*\))" regex += r"(?:\(([^\s()<>]+|(\([^\s()<>]+\)))*\)|[^\s`!()\[\]{};:'\".,<>?«»“”‘’]))" url_patterns = re.findall(regex,message) urls = [] for url in url_patterns: if url[0][-1] != '-': present_flag = False for exclude_url in EXCLUDE_URL_STRINGS: if exclude_url in url[0]: present_flag = True break if not present_flag and validators.url(url[0]): urls.append(url[0]) return urls def lambda_handler(event, context): """ Method run when Lambda is triggered calls the filtering logic """ content = get_message_contents(event) message = content['msg'] channel = content['chat_id'] user = content['user_id'] print(f'{message}, {user}, {channel}') response="" final_url=[] if channel == os.environ.get('ETC_CHANNEL') and user != os.environ.get('Qxf2Bot_USER'): print("Getting message posted on ETC ") cleaned_message = clean_message(message) final_url=get_url(cleaned_message) #Filtered URL is printed by lambda print("Final url is :",final_url) return { 'body': json.dumps(final_url) }

Requirements file contains following:

boto3==1.27.0
botocore==1.30.0
Requests==2.31.0
validators==0.20.0

C. Write test for URL filter Lambda using LocalStack and pytest:

To write a test for Lambda using the LocalStack, follow the steps below::
– 1. Create/deploy a Lambda on the LocalStack
– 2. Test to trigger the Lambda and verify result
– 3. Delete the Lambda

Step 1: Create/deploy a Lambda on the LocalStack

Following methods helps to create a Lambda on LocalStack with above our Lambda code:

#test_utils.py
 
# Specify the paths and configuration
LAMBDA_ZIP='./lambda_with_dependency.zip'
LAMBDA_FOLDER_PATH = './'
REQUIREMENTS_FILE_PATH = './requirements.txt'
CONFIG = botocore.config.Config(retries={'max_attempts': 0})
LOCALSTACK_ENDPOINT = 'http://localhost.localstack.cloud:4566'
 
def get_lambda_client():
    "get lambda client"
    return boto3.client(
        'lambda',
        aws_access_key_id= 'test',
        aws_secret_access_key= 'test',
        region_name='us-east-1',
        endpoint_url= LOCALSTACK_ENDPOINT,
        config=CONFIG
    )
 
def create_zip_file_with_lambda_files_and_packages(lambda_zip, lambda_folder_path, temp_directory):
    "Create a zip file with lambda files and installed packages"
    with zipfile.ZipFile(lambda_zip, 'w', zipfile.ZIP_DEFLATED) as zipf:
        # Add the Lambda function code to the ZIP
        for root, dirs, files in os.walk(lambda_folder_path):
            for file in files:
                file_path = os.path.join(root, file)
                if (not file_path.endswith('.zip') and not fnmatch.fnmatch(file_path, '*_pycache_*')
                    and not fnmatch.fnmatch(file_path, '*.pytest*')):
                    zipf.write(file_path, os.path.relpath(file_path, lambda_folder_path))
 
        # Add the installed packages to the ZIP at root
        for root, dirs, files in os.walk(temp_directory):
            for file in files:
                file_path = os.path.join(root, file)
                zipf.write(file_path, os.path.relpath(file_path, temp_directory))
 
def delete_temp_file_and_its_content(temp_directory):
    "Delete the temporary directory and its contents"
    for root, dirs, files in os.walk(temp_directory, topdown=False):
        for file in files:
            file_path = os.path.join(root, file)
            os.remove(file_path)
        for dir in dirs:
            dir_path = os.path.join(root, dir)
            os.rmdir(dir_path)
    os.rmdir(temp_directory)
 
def create_lambda_zip(lambda_zip, lambda_folder_path, requirements_file_path):
    "Create a zip file to deploy Lambda along with its dependencies"
    # Create a temporary directory to install packages
    temp_directory = '../package'
    os.makedirs(temp_directory, exist_ok=True)
 
 
    # Install packages to the temporary directory
    subprocess.check_call(['pip', 'install', '-r', requirements_file_path, '-t', temp_directory])
 
    # Create a new ZIP file with Lambda files and installed packages
    create_zip_file_with_lambda_files_and_packages(lambda_zip, lambda_folder_path, temp_directory)
 
    # Remove the temporary directory and its contents
    delete_temp_file_and_its_content(temp_directory)
 
def create_lambda(function_name):
    "Create Lambda"
    lambda_client = get_lambda_client()
    create_lambda_zip(LAMBDA_ZIP, LAMBDA_FOLDER_PATH, REQUIREMENTS_FILE_PATH)
    with open(LAMBDA_ZIP, 'rb') as zip_file:
        zipped_code = zip_file.read()
    lambda_client.create_function(
        FunctionName=function_name,
        Runtime='python3.8',
        Role='arn:aws:iam::123456789012:role/test-role',
        Handler=function_name + '.lambda_handler',
        Code={"ZipFile": zipped_code},
        Timeout=180,
        Environment={
            'Variables': {
                'ETC_CHANNEL': '[email protected]',
                'Qxf2Bot_USER': 'dummy_user.id.006',
            }
        }
    )

#test_utils.py # Specify the paths and configuration LAMBDA_ZIP='./lambda_with_dependency.zip' LAMBDA_FOLDER_PATH = './' REQUIREMENTS_FILE_PATH = './requirements.txt' CONFIG = botocore.config.Config(retries={'max_attempts': 0}) LOCALSTACK_ENDPOINT = 'http://localhost.localstack.cloud:4566' def get_lambda_client(): "get lambda client" return boto3.client( 'lambda', aws_access_key_id= 'test', aws_secret_access_key= 'test', region_name='us-east-1', endpoint_url= LOCALSTACK_ENDPOINT, config=CONFIG ) def create_zip_file_with_lambda_files_and_packages(lambda_zip, lambda_folder_path, temp_directory): "Create a zip file with lambda files and installed packages" with zipfile.ZipFile(lambda_zip, 'w', zipfile.ZIP_DEFLATED) as zipf: # Add the Lambda function code to the ZIP for root, dirs, files in os.walk(lambda_folder_path): for file in files: file_path = os.path.join(root, file) if (not file_path.endswith('.zip') and not fnmatch.fnmatch(file_path, '*_pycache_*') and not fnmatch.fnmatch(file_path, '*.pytest*')): zipf.write(file_path, os.path.relpath(file_path, lambda_folder_path)) # Add the installed packages to the ZIP at root for root, dirs, files in os.walk(temp_directory): for file in files: file_path = os.path.join(root, file) zipf.write(file_path, os.path.relpath(file_path, temp_directory)) def delete_temp_file_and_its_content(temp_directory): "Delete the temporary directory and its contents" for root, dirs, files in os.walk(temp_directory, topdown=False): for file in files: file_path = os.path.join(root, file) os.remove(file_path) for dir in dirs: dir_path = os.path.join(root, dir) os.rmdir(dir_path) os.rmdir(temp_directory) def create_lambda_zip(lambda_zip, lambda_folder_path, requirements_file_path): "Create a zip file to deploy Lambda along with its dependencies" # Create a temporary directory to install packages temp_directory = '../package' os.makedirs(temp_directory, exist_ok=True) # Install packages to the temporary directory subprocess.check_call(['pip', 'install', '-r', requirements_file_path, '-t', temp_directory]) # Create a new ZIP file with Lambda files and installed packages create_zip_file_with_lambda_files_and_packages(lambda_zip, lambda_folder_path, temp_directory) # Remove the temporary directory and its contents delete_temp_file_and_its_content(temp_directory) def create_lambda(function_name): "Create Lambda" lambda_client = get_lambda_client() create_lambda_zip(LAMBDA_ZIP, LAMBDA_FOLDER_PATH, REQUIREMENTS_FILE_PATH) with open(LAMBDA_ZIP, 'rb') as zip_file: zipped_code = zip_file.read() lambda_client.create_function( FunctionName=function_name, Runtime='python3.8', Role='arn:aws:iam::123456789012:role/test-role', Handler=function_name + '.lambda_handler', Code={"ZipFile": zipped_code}, Timeout=180, Environment={ 'Variables': { 'ETC_CHANNEL': '[email protected]', 'Qxf2Bot_USER': 'dummy_user.id.006', } } )

The above code is capable of creating a Lambda along with its requirements. To deploy the Lambda at the beginning of the test, call the above method under the ‘setup_class’ method. The ‘setup_class’ method runs before our main test starts executing, and I have added a method to check and wait for the Lambda to be active.
Note: Here, we are creating a zip file that includes the Lambda code and its requirements, and deploying the Lambda function. This is because the Lambda layers feature is only available for LocalStack Pro users.

#test_lambda.py
 
    @classmethod
    def setup_class(cls):
        "Create the lambda_with_dependency"
        print('\nCreating a lambda_with_dependency')
        testutils.create_lambda('lambda_with_dependency')
        cls.wait_for_function_active('lambda_with_dependency')
 
    @classmethod
    def wait_for_function_active(cls, function_name):
        "Wait till Lambda is up and active"
        lambda_client = testutils.get_lambda_client()
        while True:
            response = lambda_client.get_function(FunctionName=function_name)
            function_state = response['Configuration']['State']
 
            if function_state == 'Active':
                break
 
            time.sleep(1)  # Wait for 1 second before checking again

Step 2: Test to trigger Lambda and verify Lambda output

The following code triggers the Lambda and returns the Lambda response.

#testutils.py
 
   def test_that_lambda_returns_filtered_url(self):
        "Test Lambda's received message"
        print('\nInvoking the Lambda and verifying return message')
        message = {
                    "Records": [
                        {
                        "body": "{\"Message\":\"{\\\"msg\\\": \\\"Checkout how we can test Lambda "
                        "locally using LocalStack "
                        "https://qxf2.com/blog/testing-aws-lambda-locally-using-localstack-and-pytest\\\","
                        "\\\"chat_id\\\": \\\"[email protected]\\\", "
                        "\\\"user_id\\\":\\\"dummy_user.id.007\\\"}\"}"
                        }
                    ]
                    }
        payload = testutils.invoke_function_and_get_url('lambda_with_dependency', message)
        self.assertEqual(payload['body'], '["https://qxf2.com/blog/testing-aws-lambda-locally-using-localstack-and-pytest"]')

Make a call to the above method from the test file, along with the Lambda name and input message.

#test_lambda.py
 
   def test_that_lambda_returns_filtered_url(self):
        "Test Lambda's received message"
        print('\nInvoking the Lambda and verifying return message')
        message = {
                    "Records": [
                        {
                        "body": "{\"Message\":\"{\\\"msg\\\": \\\"Checkout how we can test Lambda "
                        "locally using LocalStack "
                        "https://qxf2.com/blog/testing-aws-lambda-locally-using-localstack-and-pytest\\\","
                        "\\\"chat_id\\\": \\\"[email protected]\\\", "
                        "\\\"user_id\\\":\\\"dummy_user.id.007\\\"}\"}"
                        }
                    ]
                    }
        payload = testutils.invoke_function_and_get_url('lambda_with_dependency', message)
        self.assertEqual(payload['body'], '["https://qxf2.com/blog/testing-aws-lambda-locally-using-localstack-and-pytest"]')

Step 3: Delete the Lambda

The following Python code deletes the Lambda. We need to call this method along with the function name.

#testutils.py
 
def delete_lambda(function_name):
    "Delete Lambda"
    lambda_client = get_lambda_client()
    lambda_client.delete_function(
        FunctionName=function_name
    )
    os.remove(LAMBDA_ZIP)

Call the above method under the ‘teardown_class’ method, along with the Lambda function name mentioned in step 1. This method will be executed at the end of the test and will handle the deletion of the Lambda.

#test_lambda.py
 
    @classmethod
    def teardown_class(cls):
        "Delete the lambda and teardown the session"
        print('\nDeleting the lambda_with_dependency')
        testutils.delete_lambda('lambda_with_dependency')

D. Consolidated test code:

The test code is divided into two files: testutils.py, which contains reusable code, and test_lambda.py, which contains specific test code.

Complete test_utils.py code:-

#test_utils.py
 
"test utils for LocalStack tests"
import json
import os
import zipfile
import subprocess
import fnmatch
import boto3
import botocore
 
# Specify the paths and configuration
LAMBDA_ZIP='./lambda_with_dependency.zip'
LAMBDA_FOLDER_PATH = './'
REQUIREMENTS_FILE_PATH = './requirements.txt'
CONFIG = botocore.config.Config(retries={'max_attempts': 0})
LOCALSTACK_ENDPOINT = 'http://localhost.localstack.cloud:4566'
 
def get_lambda_client():
    "get lambda client"
    return boto3.client(
        'lambda',
        aws_access_key_id= 'test',
        aws_secret_access_key= 'test',
        region_name='us-east-1',
        endpoint_url= LOCALSTACK_ENDPOINT,
        config=CONFIG
    )
 
def create_zip_file_with_lambda_files_and_packages(lambda_zip, lambda_folder_path, temp_directory):
    "Create a zip file with lambda files and installed packages"
    with zipfile.ZipFile(lambda_zip, 'w', zipfile.ZIP_DEFLATED) as zipf:
        # Add the Lambda function code to the ZIP
        for root, dirs, files in os.walk(lambda_folder_path):
            for file in files:
                file_path = os.path.join(root, file)
                if (not file_path.endswith('.zip') and not fnmatch.fnmatch(file_path, '*_pycache_*')
                    and not fnmatch.fnmatch(file_path, '*.pytest*')):
                    zipf.write(file_path, os.path.relpath(file_path, lambda_folder_path))
 
        # Add the installed packages to the ZIP at root
        for root, dirs, files in os.walk(temp_directory):
            for file in files:
                file_path = os.path.join(root, file)
                zipf.write(file_path, os.path.relpath(file_path, temp_directory))
 
def delete_temp_file_and_its_content(temp_directory):
    "Delete the temporary directory and its contents"
    for root, dirs, files in os.walk(temp_directory, topdown=False):
        for file in files:
            file_path = os.path.join(root, file)
            os.remove(file_path)
        for dir in dirs:
            dir_path = os.path.join(root, dir)
            os.rmdir(dir_path)
    os.rmdir(temp_directory)
 
def create_lambda_zip(lambda_zip, lambda_folder_path, requirements_file_path):
    "Create a zip file to deploy Lambda along with its dependencies"
    # Create a temporary directory to install packages
    temp_directory = '../package'
    os.makedirs(temp_directory, exist_ok=True)
 
 
    # Install packages to the temporary directory
    subprocess.check_call(['pip', 'install', '-r', requirements_file_path, '-t', temp_directory])
 
    # Create a new ZIP file with Lambda files and installed packages
    create_zip_file_with_lambda_files_and_packages(lambda_zip, lambda_folder_path, temp_directory)
 
    # Remove the temporary directory and its contents
    delete_temp_file_and_its_content(temp_directory)
 
def create_lambda(function_name):
    "Create Lambda"
    lambda_client = get_lambda_client()
    create_lambda_zip(LAMBDA_ZIP, LAMBDA_FOLDER_PATH, REQUIREMENTS_FILE_PATH)
    with open(LAMBDA_ZIP, 'rb') as zip_file:
        zipped_code = zip_file.read()
    lambda_client.create_function(
        FunctionName=function_name,
        Runtime='python3.8',
        Role='arn:aws:iam::123456789012:role/test-role',
        Handler=function_name + '.lambda_handler',
        Code={"ZipFile": zipped_code},
        Timeout=180,
        Environment={
            'Variables': {
                'ETC_CHANNEL': '[email protected]',
                'Qxf2Bot_USER': 'dummy_user.id.006',
            }
        }
    )
 
def delete_lambda(function_name):
    "Delete Lambda"
    lambda_client = get_lambda_client()
    lambda_client.delete_function(
        FunctionName=function_name
    )
    os.remove(LAMBDA_ZIP)
 
def invoke_function_and_get_url(function_name, event):
    "trigger Lambda and return received message"
    lambda_client = get_lambda_client()
 
    # Convert the event message to JSON
    event_payload = json.dumps(event)
 
    # Invoke the Lambda function
    response = lambda_client.invoke(
        FunctionName=function_name,
        InvocationType='RequestResponse',
        Payload=event_payload
    )
 
    # Parse the response from the Lambda function
    response_payload = response['Payload'].read().decode('utf-8')
    response_data = json.loads(response_payload)
    print ("response data:", response_data)
 
    return response_data

#test_utils.py "test utils for LocalStack tests" import json import os import zipfile import subprocess import fnmatch import boto3 import botocore # Specify the paths and configuration LAMBDA_ZIP='./lambda_with_dependency.zip' LAMBDA_FOLDER_PATH = './' REQUIREMENTS_FILE_PATH = './requirements.txt' CONFIG = botocore.config.Config(retries={'max_attempts': 0}) LOCALSTACK_ENDPOINT = 'http://localhost.localstack.cloud:4566' def get_lambda_client(): "get lambda client" return boto3.client( 'lambda', aws_access_key_id= 'test', aws_secret_access_key= 'test', region_name='us-east-1', endpoint_url= LOCALSTACK_ENDPOINT, config=CONFIG ) def create_zip_file_with_lambda_files_and_packages(lambda_zip, lambda_folder_path, temp_directory): "Create a zip file with lambda files and installed packages" with zipfile.ZipFile(lambda_zip, 'w', zipfile.ZIP_DEFLATED) as zipf: # Add the Lambda function code to the ZIP for root, dirs, files in os.walk(lambda_folder_path): for file in files: file_path = os.path.join(root, file) if (not file_path.endswith('.zip') and not fnmatch.fnmatch(file_path, '*_pycache_*') and not fnmatch.fnmatch(file_path, '*.pytest*')): zipf.write(file_path, os.path.relpath(file_path, lambda_folder_path)) # Add the installed packages to the ZIP at root for root, dirs, files in os.walk(temp_directory): for file in files: file_path = os.path.join(root, file) zipf.write(file_path, os.path.relpath(file_path, temp_directory)) def delete_temp_file_and_its_content(temp_directory): "Delete the temporary directory and its contents" for root, dirs, files in os.walk(temp_directory, topdown=False): for file in files: file_path = os.path.join(root, file) os.remove(file_path) for dir in dirs: dir_path = os.path.join(root, dir) os.rmdir(dir_path) os.rmdir(temp_directory) def create_lambda_zip(lambda_zip, lambda_folder_path, requirements_file_path): "Create a zip file to deploy Lambda along with its dependencies" # Create a temporary directory to install packages temp_directory = '../package' os.makedirs(temp_directory, exist_ok=True) # Install packages to the temporary directory subprocess.check_call(['pip', 'install', '-r', requirements_file_path, '-t', temp_directory]) # Create a new ZIP file with Lambda files and installed packages create_zip_file_with_lambda_files_and_packages(lambda_zip, lambda_folder_path, temp_directory) # Remove the temporary directory and its contents delete_temp_file_and_its_content(temp_directory) def create_lambda(function_name): "Create Lambda" lambda_client = get_lambda_client() create_lambda_zip(LAMBDA_ZIP, LAMBDA_FOLDER_PATH, REQUIREMENTS_FILE_PATH) with open(LAMBDA_ZIP, 'rb') as zip_file: zipped_code = zip_file.read() lambda_client.create_function( FunctionName=function_name, Runtime='python3.8', Role='arn:aws:iam::123456789012:role/test-role', Handler=function_name + '.lambda_handler', Code={"ZipFile": zipped_code}, Timeout=180, Environment={ 'Variables': { 'ETC_CHANNEL': '[email protected]', 'Qxf2Bot_USER': 'dummy_user.id.006', } } ) def delete_lambda(function_name): "Delete Lambda" lambda_client = get_lambda_client() lambda_client.delete_function( FunctionName=function_name ) os.remove(LAMBDA_ZIP) def invoke_function_and_get_url(function_name, event): "trigger Lambda and return received message" lambda_client = get_lambda_client() # Convert the event message to JSON event_payload = json.dumps(event) # Invoke the Lambda function response = lambda_client.invoke( FunctionName=function_name, InvocationType='RequestResponse', Payload=event_payload ) # Parse the response from the Lambda function response_payload = response['Payload'].read().decode('utf-8') response_data = json.loads(response_payload) print ("response data:", response_data) return response_data

Complete test_lambda.py file:-

#test_lambda.py
"""
Test Lambda on the LocalStack:
- Deploy Lambda along with its dependencies
- Run test
- Delete Lambda
"""
import time
from unittest import TestCase
import testutils
 
class Test(TestCase):
    "Test to verify Lambda response"
 
    @classmethod
    def setup_class(cls):
        "Create the lambda_with_dependency"
        print('\nCreating a lambda_with_dependency')
        testutils.create_lambda('lambda_with_dependency')
        cls.wait_for_function_active('lambda_with_dependency')
 
    @classmethod
    def teardown_class(cls):
        "Delete the lambda and teardown the session"
        print('\nDeleting the lambda_with_dependency')
        testutils.delete_lambda('lambda_with_dependency')
 
    @classmethod
    def wait_for_function_active(cls, function_name):
        "Wait till Lambda is up and active"
        lambda_client = testutils.get_lambda_client()
        while True:
            response = lambda_client.get_function(FunctionName=function_name)
            function_state = response['Configuration']['State']
 
            if function_state == 'Active':
                break
 
            time.sleep(1)  # Wait for 1 second before checking again
 
    def test_that_lambda_returns_filtered_url(self):
        "Trigger and test Lambda's received message"
        print('\nInvoking the Lambda and verifying return message')
        message = {
                    "Records": [
                        {
                        "body": "{\"Message\":\"{\\\"msg\\\": \\\"Checkout how we can test Lambda "
                        "locally using LocalStack "
                        "https://qxf2.com/blog/testing-aws-lambda-locally-using-localstack-and-pytest\\\","
                        "\\\"chat_id\\\": \\\"[email protected]\\\", "
                        "\\\"user_id\\\":\\\"dummy_user.id.007\\\"}\"}"
                        }
                    ]
                    }
        payload = testutils.invoke_function_and_get_url('lambda_with_dependency', message)
        self.assertEqual(payload['body'], '["https://qxf2.com/blog/testing-aws-lambda-locally-using-localstack-and-pytest"]')

#test_lambda.py """ Test Lambda on the LocalStack: - Deploy Lambda along with its dependencies - Run test - Delete Lambda """ import time from unittest import TestCase import testutils class Test(TestCase): "Test to verify Lambda response" @classmethod def setup_class(cls): "Create the lambda_with_dependency" print('\nCreating a lambda_with_dependency') testutils.create_lambda('lambda_with_dependency') cls.wait_for_function_active('lambda_with_dependency') @classmethod def teardown_class(cls): "Delete the lambda and teardown the session" print('\nDeleting the lambda_with_dependency') testutils.delete_lambda('lambda_with_dependency') @classmethod def wait_for_function_active(cls, function_name): "Wait till Lambda is up and active" lambda_client = testutils.get_lambda_client() while True: response = lambda_client.get_function(FunctionName=function_name) function_state = response['Configuration']['State'] if function_state == 'Active': break time.sleep(1) # Wait for 1 second before checking again def test_that_lambda_returns_filtered_url(self): "Trigger and test Lambda's received message" print('\nInvoking the Lambda and verifying return message') message = { "Records": [ { "body": "{\"Message\":\"{\\\"msg\\\": \\\"Checkout how we can test Lambda " "locally using LocalStack " "https://qxf2.com/blog/testing-aws-lambda-locally-using-localstack-and-pytest\\\"," "\\\"chat_id\\\": \\\"[email protected]\\\", " "\\\"user_id\\\":\\\"dummy_user.id.007\\\"}\"}" } ] } payload = testutils.invoke_function_and_get_url('lambda_with_dependency', message) self.assertEqual(payload['body'], '["https://qxf2.com/blog/testing-aws-lambda-locally-using-localstack-and-pytest"]')

E. Run the test:

Before you execute the test, it is crucial to confirm that LocalStack is up and running and that all the necessary requirements have been installed. You can use the following command to check the status:

 localstack status

This command will display the status, IP, and container name of LocalStack.

To run the test, you need to install the test requirements using pip. You have two options: either install all requirements using the requirements.txt file (pip install -r requirements.txt), or directly install boto3 and botocore using the pip command (pip install boto3 botocore). For the test to run successfully, you only need boto3 and botocore, as they are the ones utilized in our test file. The other requirements are related to Lambda, which our test code will take care of.

Now, proceed to run the test using pytest.

 pytest -s -v test_lambda.py

Please refer to the following screenshot for the output.

Successfully integrating LocalStack with pytest and using boto to deploy onto LocalStack does require some specific knowledge and expertise. Hope our blog helps you in testing your Lambda locally using LocalStack.

Hire Qxf2!

Testers from Qxf2 go well beyond standard test automation practices. We solve the most pressing testing challenges that prevent teams from moving quickly and confidently. With our expertise, we address critical testing roadblocks that hinder progress. Contact us today to experience the Qxf2 advantage and propel your testing efforts forward. Visit here.

Rohan Dudam

I love technology and learning new things. I explore both hardware and software. I am passionate about robotics and embedded systems which motivate me to develop my software and hardware skills. I have good knowledge of Python, Selenium, Arduino, C and hardware design. I have developed several robots and participated in robotics competitions. I am constantly exploring new test ideas and test tools for software and hardware. At Qxf2, I am working on developing hardware tools for automated tests ala Tapster. Incidentally, I created Qxf2’s first robot. Besides testing, I like playing cricket, badminton and developing embedded gadget for fun.

Testing AWS Lambda locally using LocalStack and pytest

Testing AWS Lambda locally using LocalStack and pytest

A. Steps to get set with LocalStack:

1. Install LocalStack and check version:

2. Start LocalStack using docker-compose:

3. Verify LocalStack up or not:

4. Verify LocalStack available services:

B. Our URL Filter Lambda:

C. Write test for URL filter Lambda using LocalStack and pytest:

Step 1: Create/deploy a Lambda on the LocalStack

Step 2: Test to trigger Lambda and verify Lambda output

Step 3: Delete the Lambda

D. Consolidated test code:

E. Run the test:

Hire Qxf2!

Leave a Reply Cancel reply

Subscribe to our weekly Newsletter

A. Steps to get set with LocalStack:

1. Install LocalStack and check version:

2. Start LocalStack using docker-compose:

3. Verify LocalStack up or not:

4. Verify LocalStack available services:

B. Our URL Filter Lambda:

C. Write test for URL filter Lambda using LocalStack and pytest:

Step 1: Create/deploy a Lambda on the LocalStack

Step 2: Test to trigger Lambda and verify Lambda output

Step 3: Delete the Lambda

D. Consolidated test code:

E. Run the test:

Hire Qxf2!

Related posts:

Leave a Reply Cancel reply

You may like this....