moto
A python library that allows you to easily mock out tests based on AWS infrastructure.
Install AWS CLI
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 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
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
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 .