Motivation to write below blog:

I used to write the lambda functions . like each lambda will work as a one micro service, it will have the API Gateway and S3 bucket.

But every time I hit the API it will call the lambda then the records will be created in the dynamo db and as well as it will create events then store the records in the S3 bucket.

Every time I test it will create new resource in the AWS and I need to destroy it manually using aws cli or pragmatically if it is incorrect .

package:

pip install moto

Points to be noted:

It will support nearly 90% of all web services
Every service will be mocked
We can integrate with Pytest and unittest

To create s3 :

import boto3


def create_bucket(s3_client, bucket_name):
    s3_client.create_bucket(Bucket=bucket_name)
    available_buckets = s3_client.list_buckets()
    return available_buckets

s3_client = boto3.client("s3")
bucket_name = "test_bucket"
create_b = create_bucket(s3_client, bucket_name)

If we use above code it will actually create a bucket in the AWS s3 . like if we run 20 times with some code change it will that many time s3 buckets .

We know everything in AWS is chargeable, whatever the service it is .

Using the Moto:

#test_mock_s3.py

from moto import mock_s3, mock_dynamodb2
import boto3
import pytest
import os


@pytest.fixture(scope='function')
def aws_credentials():
    """Mocked AWS Credentials for moto."""
    os.environ['AWS_ACCESS_KEY_ID'] = 'testing'
    os.environ['AWS_SECRET_ACCESS_KEY'] = 'testing'
    os.environ['AWS_SECURITY_TOKEN'] = 'testing'
    os.environ['AWS_SESSION_TOKEN'] = 'testing'
    os.environ["aws_region"] = "us-east-1"


@pytest.fixture(scope="function")
def dynamodb_client(aws_credentials):
    with mock_dynamodb2():
        yield boto3.client('dynamodb', region_name="us-east-1")


@pytest.fixture(scope='function')
def s3(aws_credentials):
    with mock_s3():
        yield boto3.client('s3', region_name='us-east-1')


def test_mock_s3(s3):
    bucket_name = "test_bucket"
    available_buckets = create_bucket(s3, bucket_name)
    for bucket in available_buckets["Buckets"]:
        assert bucket["Name"] == bucket_name


def create_bucket(s3_client, bucket_name):
    s3_client.create_bucket(Bucket=bucket_name)
    available_buckets = s3_client.list_buckets()
    return available_buckets

if we run using the pytest then it will call the test_mock_s3 and create a resource in virtually . for the actual resource creation whatever the response we get the same response we will get here.

pytest  test_mock_s3.py -s

Requirements.txt :

pytest
moto

Resources:

Happy coding.

Venkatesh Blog