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:
-
2.1 Create docker-compose file:
Use the following content to create a docker-compose.yml file.
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) } |
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', } } ) |
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 |
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"]') |
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.
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.
