はじめに

本記事はさまざまなAWSサービスへアクセスするLambda関数に対し、pytestを用いたローカル環境でのテスト実行を行うにあたり、AWSサービスをモック化したお話を技術共有と自身の備忘録もかねて記載してきます。

pytest自体のお話は別の記事がございますので、そちらも合わせて読んでみてください！
→Pytestを用いたAPIテストを初めて実装した時の話
→Pytestでテストをもっと効率化してみたお話

AWSサービスのモックツール

今回はmotoというライブラリを利用します。motoはbotoと連携して動作し、AWSインフラをもとにしたテストを簡単にモック化できるライブラリです。モック上でAWSサービスを操作する方法はbotoドキュメントにて確認ください。

まずはmotoドキュメント Getting Started with Motoから始めてみましょう！

環境構築

まずはmotoをインストールします。以下のように、利用するAWSサービスを指定してインストールができます。もちろん全てを一度にインストールすることも可能です。

pip install 'moto[ec2,s3,..]'
pip install 'moto[all]'
pip install 'moto'

motoの利用方法

motoの利用方法は3つあります。

デコレーター
```
@mock_aws
def test_my_model_save():
    # 〜モック実装〜
```
関数の一番上にデコレーターとして追加するだけの一番簡単な記述方法です。
もし、モック関数をフィクスチャとして利用する場合は、@mock_awsの後に@pytest.fixture()を宣言します。

motoでのモック方法について、少し調べてみるとデコレートの書き方がAWSサービスごと@mock_ec2や@mock_s3のように記述があります。
こちらv5以降、すべて@mock_awsへ統一されました。利用されるバージョンに十分ご注意ください。
コンテキストマネージャ
```
def test_my_model_save():
    with mock_aws():
        # 〜モック実装〜
```
モック化したい箇所をwith文のインデント内に記述する方法です。
他通信をモック化したいなど、複数モックする場合や、モック外の処理記述が必要な時に有効です。
コード直書き
```
def test_my_model_save():
    mock = mock_aws()
    mock.start()
    # 〜モック実装〜
    mock.stop()
```
起動と停止をコード内で記述する方法です。一番柔軟に利用できます。
しかし、停止処理をきちんとしないとエラーになるので注意。フィクスチャとして利用する際はyieldの後に停止処理を忘れずに！

以降の実装例では2.コンテキストマネージャを利用した方法を利用しています。

AWSサービスのモック化

以降では実際に自分自身が利用したサービスのモック方法についてまとめていきます。

DynamoDB

ここでのDynamoDBのモックはテーブルの作成と、テストデータの格納の2つフィクスチャで構成します。
これは、テスト関数ごとに格納するデータが異なったり、格納する必要がなかったりするためです。

create_table

@pytest.fixture()
def setup_dynamoDB():
"""DynamoDBの設定を初期化するフィクスチャー"""
    with mock_aws():
        dynamodb_resource = boto3.resource('dynamodb')
        tables = {}

        # テーブル1作成
        tables["table1"] = dynamodb_resource.create_table(
            TableName="table_1",
            KeySchema=[
                {'AttributeName': 'partition_key', 'KeyType': 'HASH'}  # パーティションキー
            ],
            AttributeDefinitions=[
                {'AttributeName': 'partition_key', 'AttributeType': 'S'}
            ],
            ProvisionedThroughput={
                'ReadCapacityUnits': 1,
                'WriteCapacityUnits': 1
            }
        )
        
        # テーブル2作成
        tables["table2"] = dynamodb_resource.create_table(
            TableName="table2",
            KeySchema=[
                {'AttributeName': 'primary_id', 'KeyType': 'HASH'},  # パーティションキー
                {'AttributeName': 'sort_key', 'KeyType': 'RANGE'}  # ソートキー
            ],
            AttributeDefinitions=[
                {'AttributeName': 'primary_id', 'AttributeType': 'S'},
                {'AttributeName': 'sort_key', 'AttributeType': 'S'},
                {'AttributeName': 'alternative_key', 'AttributeType': 'S'}  # LSI のソートキー
            ],
            LocalSecondaryIndexes=[
                {
                    'IndexName': 'LSI',
                    'KeySchema': [
                        {'AttributeName': 'primary_id','KeyType': 'HASH'},  # 同じパーティションキー
                        {'AttributeName': 'alternative_key','KeyType': 'RANGE'}  # LSI のソートキー
                    ],
                    'Projection': {
                        'ProjectionType': 'ALL'  # 全ての属性を射影
                    }
                }
            ],
            ProvisionedThroughput={
                'ReadCapacityUnits': 1,
                'WriteCapacityUnits': 1
            }
        )
        yield tables

ここではPKのみ設定しているTable1と、セカンダリインデックスの設定も含めたTable2の2つのテーブルを作成しました。一度に複数テーブルを作成し、辞書形式で取得できるように工夫しました。

put_item

@pytest.fixture
def insert_data(setup_dynamoDB):
    """テストデータを挿入するフィクスチャー"""
    # テーブルを取得
    table = setup_dynamoDB['table1']
    # テストデータ定義
    test_data = [{data1},{data2}]
    
    # テストデータを挿入
    for item in test_data:
        table.put_item(Item=item)
        
    yield

複数のテストデータに対し、1件ずつPUT処理を行います。

S3

ここでのS3のモックはバケットの作成と、テストデータの格納の2つフィクスチャで構成します。

create_bucket

@pytest.fixture()
def setup_s3():
    """S3の設定を初期化するフィクスチャー"""
    with mock_aws():
        s3_client = boto3.client('s3')

        # バケット作成
        s3_client.create_bucket(
            Bucket="bucket1",
            CreateBucketConfiguration={"LocationConstraint": 'ap-northeast-1'},
        )

        yield s3_client

バケット名と、リージョンを指定し、バケットを作成します。

put_object

@pytest.fixture
def insert_data_s3(setup_s3):
    """テストデータを挿入するフィクスチャー"""
    # テストデータ定義
    test_data = [{data1},{data2}]

    # テストデータを挿入
    for item in test_data:
        setup_s3.put_object(Bucket="bucket1",Key=item['key'], 
                            Body=item['body'], ContentType=item['content_type'])

    yield

複数のテストデータに対し、1件ずつPUT処理を行います。

EventBridge Scheduler

ここでのEventBridge Schedulerのモックは、スケジュールグループの作成と、スケジュールの作成の2つのフィクスチャで構成します。

create_schedule_group

@pytest.fixture()
def setup_eventbridge():
    """EventBridgeのモックセットアップ"""
    with mock_aws():
        eventbridge_client = boto3.client('scheduler')

        # スケジュールグループを作成
        eventbridge_client.create_schedule_group(
            Name='schedule1',
        )

        yield eventbridge_client

スケジュール名を指定し、スケジュールグループを作成します。

create_schedule

@pytest.fixture
def create_schedule_image(setup_eventbridge):
    """スケジュールを作成するフィクスチャー"""
    # テストデータ定義
    test_data = [{data1},{data2}]

    # テストデータを挿入
    for item in test_data:
        setup_eventbridge.create_schedule(
            ActionAfterCompletion='NONE',
            ClientToken='string',
            FlexibleTimeWindow={
                'MaximumWindowInMinutes': 1,
                'Mode': 'FLEXIBLE'
            },
            Name='1234567890',
            ScheduleExpression='rate(1 minutes)',
            State='ENABLED',
            Target={
                'Arn': 'arn:aws:lambda:ap-northeast-1:123456789012:function:test-function',
                'Input': json.dumps(item),
                'RoleArn': 'arn:aws:iam::123456789012:role/test-role',
            })

    yield setup_eventbridge

ここでは1分ごとに起動するスケジュールを作成しました。

LambdaARNや実行ロールARNは'moto'のデフォルトアカウント123456789012を利用しています。
参照：motoドキュメント Multi-Account support

boto3.client('events')はEventBridgeのイベント駆動ルールへ関するサービス、
boto3.client('scheduler')はEventBridgeのスケジューリングへ関するサービス
です。同じEventBridge内の機能ですが、異なる機能なのでご注意を！

SQS

ここでのSQSのモックは、キューを作成します。

create_queue

@pytest.fixture()
def setup_sqs():
    """SQSキューの設定を初期化するフィクスチャー"""
    with mock_aws():
        sqs_resource = boto3.resource('sqs')
        queues = {}

        # FIFOキューの作成
        queues["receive_line"] = sqs_resource.create_queue(
            QueueName='test-receiveline.fifo',
            Attributes={
                'FifoQueue': 'true',                     # FIFOキューの有効化
                'ContentBasedDeduplication': 'true',     # コンテンツベースの重複排除を有効化
                'VisibilityTimeout': '60',               # 可視性タイムアウト: 1分
                'MessageRetentionPeriod': '60',          # メッセージ保持期間: 1分
                'DelaySeconds': '0',                     # 配信遅延: 0秒
                'ReceiveMessageWaitTimeSeconds': '0'     # メッセージ受信待機時間: 0秒
            }
        )

        yield queues

ここでは1分サイクルのFIFOキューを作成しましたが、さまざまな設定値でのキューを作成することができます。

Secrets Manager

Secrets Managerのモックはシークレットを作成します。

create_secret

@pytest.fixture()
def setup_secrets():
    """Secrets Managerの設定を初期化するフィクスチャー"""
    with mock_aws():
        secrets_manager_client = boto3.client('secretsmanager')
        secrets = {}
        
        # Secrets Managerの作成
        secrets[secret_name] = secrets_manager_client.create_secret(
            Name='secrets',
            SecretString='{"key": "secrets"}'
        )
        
        yield secrets

DynamoDBと同様で、一度に複数Secretsを作成し、辞書形式で取得できるように工夫しました。（上記では1つのみ作成）

また、Secretsの値をテスト関数ごとに変更したいため、独自のマーカーを作成し、関数ごとに異なる値を設定できるよう修正してみましょう。

def pytest_configure(config):
    """カスタムマーカーの登録"""
    config.addinivalue_line(
        "markers",
        "secrets_config(secret_name, secret_string): Secrets Manager設定を指定するマーカー"
    )

@pytest.fixture
def setup_secrets(request):
    # 〜上記に追加し、作成値をマーカーから取得〜
    marker = request.node.get_closest_marker('secrets_config')
    if marker:
        # マーカーが設定されている場合は、新しい設定で上書き
        secret_name = marker.kwargs.get('secret_name', secret_name)
        secret_string = marker.kwargs.get('secret_string', secret_string)

@pytest.mark.secrets_config(
    secret_name='newsecrets',
    secret_string='{"key": "secrets"}'
)
def test1():
    # 〜テスト関数記述〜

このように、'secrets_config'というマーカーを作成し、テスト関数に'@pytest.mark.secrets_config'を追加することで、指定したSecrets値を適用したフィクスチャ処理を実施することができます。
これは他のAWSサービスでも汎用することが可能です。

今回紹介したAWSサービス以外もモック化は可能です。一部motoでは実装されていないAWSサービスもありますので、ご確認ください。
→motoドキュメント Implemented Services

まとめ

今回はさまざまなAWSサービスに対しmotoを用いてモック化する方法をご紹介しました。以前DIを学んでからテストのしやすさをテストのしやすさというものを意識していますが、AWSという外部サービスに対してもモック化することでテストができるようになるのはとても便利だなぁ、と感じました。引き続き頑張っていきます〜