pytestでのテストでAWSサービスをモック化してみたお話

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

pytest自体のお話は別の記事がございますので、そちらも合わせて読んでみてください！
→[Pytestを用いたAPIテストを初めて実装した時の話](https://qiita.com/inacyc_k/items/4eda8918bcc4699bbe53)
→[Pytestでテストをもっと効率化してみたお話](https://qiita.com/inacyc_k/items/ac3a3650b234099ee1ad)

# AWSサービスのモックツール
今回は`moto`というライブラリを利用します。`moto`は`boto`と連携して動作し、AWSインフラをもとにしたテストを簡単にモック化できるライブラリです。モック上でAWSサービスを操作する方法は[botoドキュメント](https://boto3.amazonaws.com/v1/documentation/api/latest/index.html#)にて確認ください。

まずは[motoドキュメント Getting Started with Moto](https://docs.getmoto.org/en/latest/docs/getting_started.html#getting-started-with-moto)から始めてみましょう！

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

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

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

1. デコレーター
    ```python
    @mock_aws
    def test_my_model_save():
        # 〜モック実装〜
    ```
    関数の一番上にデコレーターとして追加するだけの一番簡単な記述方法です。
    もし、モック関数をフィクスチャとして利用する場合は、`@mock_aws`の後に`@pytest.fixture()`を宣言します。
    
    :::note info
    motoでのモック方法について、少し調べてみるとデコレートの書き方がAWSサービスごと`@mock_ec2`や`@mock_s3`のように記述があります。
    こちらv5以降、すべて`@mock_aws`へ統一されました。利用されるバージョンに十分ご注意ください。 

1. コンテキストマネージャ
    ```python
    def test_my_model_save():
        with mock_aws():
            # 〜モック実装〜
    ```
    モック化したい箇所を`with`文のインデント内に記述する方法です。
    他通信をモック化したいなど、複数モックする場合や、モック外の処理記述が必要な時に有効です。

1. コード直書き
    ```python
    def test_my_model_save():
        mock = mock_aws()
        mock.start()
        # 〜モック実装〜
        mock.stop()
    ```
    起動と停止をコード内で記述する方法です。一番柔軟に利用できます。
    しかし、停止処理をきちんとしないとエラーになるので注意。フィクスチャとして利用する際は`yield`の後に停止処理を忘れずに！
---
以降の実装例では2.コンテキストマネージャを利用した方法を利用しています。

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

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

- create_table
```python
@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
```python
@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
```python
@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
```python
@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
```python
@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
```python
@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](https://docs.getmoto.org/en/latest/docs/multi_account.html#multi-account-support)

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

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

- create_queue
```python
@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
```python
@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の値をテスト関数ごとに変更したいため、独自のマーカーを作成し、関数ごとに異なる値を設定できるよう修正してみましょう。

```python
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サービスでも汎用することが可能です。

:::note info
今回紹介したAWSサービス以外もモック化は可能です。一部motoでは実装されていないAWSサービスもありますので、ご確認ください。
→[motoドキュメント Implemented Services](https://docs.getmoto.org/en/latest/docs/services/index.html#implemented-services)
:::

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