motoとpytestでlambdaにおけるS3接続のモックを使ったテスト

root
├── src
│   └── lambda_function.py # テスト対象コード
└── tests
    ├── __init__.py 
    ├── conftest.py    # 事前・事後処理用
    └── test_lambda.py # テストコード

pip install moto[essential]

import boto3


def lambda_handler(event, context):
    """デモハンドラー"""
    # イベント情報からバケット名・オブジェクトキー・リージョンを取得
    bucket_name = event["Records"][0]["s3"]["bucket"]["name"]
    key = event["Records"][0]["s3"]["object"]["key"]
    region = event["Records"][0]["awsRegion"]

    s3_client = boto3.client("s3", region_name=region)

    try:
        # ファイルの中身を取得し返す
        response = s3_client.get_object(Bucket=bucket_name, Key=key)
        body = response.get("Body")
        return body.read()
    except Exception as e:
        print(
            f"Error getting object {key} from bucket {bucket_name}. Make sure "
            + "they exist and your bucket is in the same region as this function."
        )
        raise e

import os

import boto3
import pytest
from moto import mock_s3

dummy_reagion = "us-east-1"
dummy_bucket_name = "test_bucket"
dummy_key = "sample.txt"


@pytest.fixture(scope="function")
def aws_credentials():
    """ダミー接続情報"""
    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_DEFAULT_REGION"] = dummy_reagion


@pytest.fixture(scope="function")
def s3(aws_credentials):
    """テスト用のモックS3接続"""
    with mock_s3():
        yield boto3.client("s3", region_name=dummy_reagion)


@pytest.fixture
def create_bucket(s3):
    """テスト用バケットの作成"""
    boto3.client("s3").create_bucket(Bucket=dummy_bucket_name)


@pytest.fixture
def put_dummy_textfile(create_bucket):
    """テスト用のファイルを設置してテスト後に削除する"""
    s3_client = boto3.client("s3", region_name=dummy_reagion)
    s3_client.put_object(Bucket=dummy_bucket_name, Key=dummy_key, Body="test")
    yield
    s3_client.delete_objects(
        Bucket=dummy_bucket_name, Delete={"Objects": [{"Key": dummy_key}]}
    )

import pytest

from src.lambda_function import lambda_handler

from .conftest import dummy_bucket_name, dummy_key, dummy_reagion

dummy_event = {
    "Records": [
        {
            "awsRegion": dummy_reagion,
            "s3": {
                "bucket": {
                    "name": dummy_bucket_name,
                },
                "object": {
                    "key": dummy_key,
                },
            },
        }
    ]
}


@pytest.mark.parametrize(
    "event, expected",
    [(dummy_event, "test")],
)
def test_lambda_handler(event, expected, put_dummy_textfile):
    # モックされた仮想S3よりテキストを取得して期待データと比較
    assert expected == lambda_handler(event, None).decode("utf-8")