moto

A python library that allows you to easily mock out tests based on AWS infrastructure.

Install AWS CLI

   1 # Install aws cli
   2 cd ~/Downloads
   3 curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"
   4 unzip awscliv2.zip
   5 sudo ./aws/install
   6 aws --version

Create S3 bucket

   1 export AWS_ACCESS_KEY_ID='testing'
   2 export AWS_SECRET_ACCESS_KEY='testing'
   3 export AWS_SECURITY_TOKEN='testing'
   4 export AWS_SESSION_TOKEN='testing'
   5 export AWS_DEFAULT_REGION='us-east-1'
   6 export AWS_ENDPOINT_URL="http://localhost:4000" 
   7 
   8 docker run --rm -p 4000:4000 --name moto motoserver/moto:latest

compose.yaml

   1 services:
   2   motoserver:
   3     image: motoserver/moto:latest
   4     ports:
   5       - "4000:4000"
   6     environment:
   7       - MOTO_PORT=4000 # set moto listener port with env var
   8       - MOTO_DOCKER_NETWORK_NAME=motoserver_default # docker network ls 
   9     volumes:
  10       - "/var/run/docker.sock:/var/run/docker.sock"

   1 docker compose up
   2 docker compose exec motoserver sh
   3 # in the container 
   4 cat /etc/os-release
   5 # NAME="Alpine Linux"
   6 exit
   7 
   8 aws s3 ls 
   9 aws s3api create-bucket --bucket my-bucket 
  10 # https://docs.aws.amazon.com/cli/latest/reference/s3api/
  11 echo "test" > test.txt
  12 aws s3api put-object --bucket my-bucket --key dir-1/test.txt --body test.txt 
  13 aws s3api get-object --bucket my-bucket --key dir-1/test.txt test2.txt 
  14 cat test2.txt 
  15 aws s3 ls

Create lambda function

start-moto.sh

   1 #!/bin/sh
   2 docker compose up

build.sh

   1 #!/bin/bash
   2 # sudo apt install python3-pip
   3 PACKAGE_NAME="py-my-function.zip"
   4 BUILD_DIR="dist"
   5 HANDLER_FILE="lambda_function.py"
   6 TEST_FILE="test_lambda.py"
   7 
   8 echo "Cleaning up old builds"
   9 rm -rf $BUILD_DIR
  10 rm -f $PACKAGE_NAME
  11 mkdir $BUILD_DIR
  12 echo "Installing dependencies in $BUILD_DIR ..."
  13 pip3 install -r requirements.txt -t $BUILD_DIR/
  14 echo "Running tests..."
  15 export PYTHONPATH=$PYTHONPATH:$(pwd)/$BUILD_DIR
  16 python3 -m unittest $TEST_FILE
  17 
  18 if [ $? -eq 0 ]; then
  19     echo "Tests passed! Proceeding to build..."
  20 else
  21     echo "Tests failed. Build aborted."
  22     exit 1
  23 fi
  24 
  25 echo "Clean up pycache"
  26 find $BUILD_DIR -type d -name "__pycache__" -exec rm -rf {} +
  27 echo "Copy source code"
  28 cp $HANDLER_FILE $BUILD_DIR/
  29 echo "Creating the ZIP $PACKAGE_NAME"
  30 cd $BUILD_DIR
  31 zip -r ../$PACKAGE_NAME .
  32 cd ..
  33 echo "Deployment package ready: $PACKAGE_NAME"

test.sh

   1 #!/bin/bash
   2 # sudo apt install python3-pip
   3 PACKAGE_NAME="py-my-function.zip"
   4 BUILD_DIR="dist"
   5 HANDLER_FILE="lambda_function.py"
   6 TEST_FILE="test_lambda.py"
   7 
   8 echo "Cleaning up old builds"
   9 rm -rf $BUILD_DIR
  10 rm -f $PACKAGE_NAME
  11 mkdir $BUILD_DIR
  12 echo "Installing dependencies in $BUILD_DIR ..."
  13 pip3 install -r requirements.txt -t $BUILD_DIR/
  14 echo "Running tests..."
  15 export PYTHONPATH=$PYTHONPATH:$(pwd)/$BUILD_DIR
  16 python3 -m unittest $TEST_FILE
  17 
  18 if [ $? -eq 0 ]; then
  19     echo "Tests passed! Proceeding to build..."
  20 else
  21     echo "Tests failed. Build aborted."
  22     exit 1
  23 fi

compose.yaml

   1 services:
   2   motoserver:
   3     image: motoserver/moto:latest
   4     ports:
   5       - "4000:4000"
   6     environment:
   7       - MOTO_PORT=4000 # set moto listener port with env var
   8       - MOTO_DOCKER_NETWORK_NAME=motoserver_default # docker network ls 
   9     volumes:
  10       - "/var/run/docker.sock:/var/run/docker.sock"

test_lambda.py

   1 import os
   2 import unittest
   3 from unittest.mock import patch, MagicMock
   4 from lambda_function import lambda_handler
   5 
   6 class TestLambda(unittest.TestCase):
   7     @patch.dict(os.environ, {'AWS_ACCESS_KEY_ID':'xxx1'})
   8     @patch.dict(os.environ, {'AWS_SECRET_ACCESS_KEY':'xxx2'})
   9     @patch.dict(os.environ, {'AWS_DEFAULT_REGION':'xxx3'})
  10     @patch.dict(os.environ, {'AWS_ENDPOINT_URL':'xxx4'})
  11     @patch('lambda_function.get_s3_client')
  12     @patch('lambda_function.get_buckets')
  13     def test_handler_success(self, mock_get_buckets, mock_get_s3client):
  14         # mock function get_buckets return value 
  15         mock_get_buckets.return_value=['aaaa','bbb']
  16 
  17         # the return value for get_s3_client is a MagicMock
  18         mock_get_s3client.return_value=MagicMock() 
  19         # when s3_client.create_bucket is called it returns dictionary cow chicken 
  20         mock_get_s3client.return_value.create_bucket.return_value = {'cow':'chicken'}
  21 
  22         # Execute lambda handler
  23         event = {'first_name':'aaa','last_name':'bbb'}
  24         context = None
  25 
  26         response = lambda_handler(event, context)
  27         self.assertEqual(response['message'],"Hello aaa bbb! <class 'dict'> {\"cow\": \"chicken\"}")
  28 
  29 if __name__ == '__main__':
  30     unittest.main()

requirements.txt

boto3
moto

lambda_function.py

   1 import boto3
   2 import os
   3 import json 
   4 import datetime 
   5 
   6 def get_s3_client():
   7     session = boto3.session.Session()
   8     s3_client = session.client( service_name='s3', endpoint_url='http://motoserver:4000' )
   9     return s3_client
  10 
  11 def get_buckets(s3_client):
  12     buckets=[]
  13     for bucket in s3_client.list_buckets()['Buckets']:
  14         buckets.append(bucket['Name'])
  15     return buckets 
  16 
  17 def lambda_handler(event, context):
  18     message = 'Hello {} {}!'.format(event['first_name'], event['last_name'])
  19     s3_client = get_s3_client()
  20 
  21     buckets=get_buckets(s3_client)
  22 
  23     response = s3_client.create_bucket(Bucket='examplebucket')
  24     assert isinstance(response,dict)
  25 
  26     body = {
  27         'message' : "%s %s %s"%(message , str(type(response)) , json.dumps(response) ),
  28         'buckets' : buckets,
  29         'currdatetime': datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
  30         'AWS_ACCESS_KEY_ID' : os.environ["AWS_ACCESS_KEY_ID"],
  31         'AWS_SECRET_ACCESS_KEY' : os.environ["AWS_SECRET_ACCESS_KEY"],
  32         'AWS_DEFAULT_REGION' : os.environ["AWS_DEFAULT_REGION"],
  33         'AWS_ENDPOINT_URL': os.environ['AWS_ENDPOINT_URL']
  34     }
  35 
  36     s3_client.put_object(Body=str(body), Bucket='examplebucket', Key='examplebucket/response.txt')
  37     return body
  38 
  39 """
  40 deploy.sh 
  41 aws s3 ls 
  42 aws s3api get-object --bucket examplebucket --key examplebucket/response.txt r1.txt 
  43 cat r1.txt 
  44 """

deploy.sh

   1 #!/bin/sh
   2 export AWS_ACCESS_KEY_ID='testing'
   3 export AWS_SECRET_ACCESS_KEY='testing'
   4 export AWS_SECURITY_TOKEN='testing'
   5 export AWS_SESSION_TOKEN='testing'
   6 export AWS_DEFAULT_REGION='us-east-1'
   7 export AWS_ENDPOINT_URL="http://localhost:4000" 
   8 # export DOCKER_HOST=unix:///var/run/docker.sock
   9 
  10 ROLE_NAME=lambda-ex
  11 LAMBDA_FUNCTION_NAME=py-my-function
  12 ZIP_NAME=py-my-function.zip 
  13 MODULE_NAME=lambda_function
  14 
  15 zip $ZIP_NAME $MODULE_NAME.py
  16 aws lambda delete-function --function-name $LAMBDA_FUNCTION_NAME
  17 aws iam delete-role --role-name $ROLE_NAME
  18 
  19 aws iam create-role --role-name $ROLE_NAME --assume-role-policy-document '{"Version": "2012-10-17","Statement": [{ "Effect": "Allow", "Principal": {"Service": "lambda.amazonaws.com"}, "Action": "sts:AssumeRole"}]}'
  20 
  21 ROLE_ARN=$(aws iam list-roles | jq ".Roles[0].Arn" | sed 's/\"//g')
  22 
  23 echo "Role Arn $ROLE_ARN"
  24 aws lambda create-function --function-name $LAMBDA_FUNCTION_NAME --zip-file fileb://$ZIP_NAME --handler $MODULE_NAME.lambda_handler --runtime python3.14 --role $ROLE_ARN --timeout 120
  25 PAYLOAD=$( echo "{ \"first_name\": \"Bob\",\"last_name\":\"Squarepants\" }"  | base64 )
  26 
  27 echo "Invoke lambda"
  28 aws lambda invoke --function-name $LAMBDA_FUNCTION_NAME --payload $PAYLOAD response.json 
  29 
  30 echo "Show response"
  31 cat response.json | jq .