やりたいこと

LambdaでDynamoDBからデータを取得するプログラムの単体テストをやるため、DynamoDB関連の処理をモック化したい

前提

・Pythonインストール済み
・pipenvインストール済み
（今回は便宜上pipenv内でtoxでテストをやるのですが、ローカルで諸々ライブラリを入れてテストをやっても大丈夫です）

ディレクトリ構成

root 　　   
  ┣━ functions
      ┗━ api
          ┗━ app.py
  ┣━ tests
      ┣━ conftest.py
      ┗━ test_app.py
  ┣━ Pipfile
  ┗━ tox.ini

ソースの中身

ごく簡単の処理ですが、あるDynamoDBテーブルからget_itemをやりたい

app.py

import boto3
import os

# DynamoDB関連情報はLambdaの環境変数から取る
DYNAMO_TABLE_NAME = os.environ.get('DYNAMO_TABLE_NAME')
DYNAMO_REGION = os.environ.get('DYNAMO_REGION', default="ap-southeast-1")
dynamodb = boto3.resource('dynamodb', region_name=DYNAMO_REGION)

def lambda_handler(event):
    # keyの値はリクエストbody値から取る
    id = event['body']['id']
    name = event['body']['name']
    
    key = {
            'id': id,
            'name': name,
        }
    data = dynamodb.Table(DYNAMO_TABLE_NAME).get_item(Key=key)

    return data

テスト時は本物のDynamoDBにアクセスしないように、テストの前準備として、motoでmock化したDynamoDBテーブルを作りDummyのデータをputする関数を用意しておきます。

これは、単体テストを実行するとき、先にmock_tableが実行され、create_authority_table関数を返すような処理です。

conftest.py

import pytest
import boto3


DYNAMO_REGION = 'ap-southeast-1'
DYNAMO_TABLE_NAME = 'auth_table'


@pytest.fixture(scope='package')
def mock_table():
    def create_authority_table():
        ddb = boto3.resource('dynamodb', region_name=DYNAMO_REGION)
        # Dummyテーブルを作る
        table = ddb.create_table(
            TableName=DYNAMO_TABLE_NAME ,
            KeySchema=[
                {
                    'AttributeName': 'id',
                    'KeyType': 'HASH'
                },
                {
                    'AttributeName': 'name',
                    'KeyType': 'RANGE'
                },
            ],
            AttributeDefinitions=[
                {
                    'AttributeName': 'id',
                    'AttributeType': 'S'
                },
                {
                    'AttributeName': 'name',
                    'AttributeType': 'S'
                },
            ],
            ProvisionedThroughput={
                'ReadCapacityUnits': 1,
                'WriteCapacityUnits': 1
            }
        )

        # ちょっと待ち時間を置く
        table.wait_until_exists()

        # Dummyデータ投入
        table.put_item(
                Item={
                    'id': '0001',
                    'name': 'satoh',
                    'company': 'A',
                },
        )

        return ddb.Table(DYNAMO_TABLE_NAME)
    return create_authority_table

@mock_dynamodb()を書くと、本物のDynamoDBを見ずに、mock_tableで作られたテーブルを見に行くことになります。

test_app.py

import importlib
import pytest
from moto import mock_dynamodb


@pytest.mark.describe("Lambdaの単体テスト")
class TestLambda:
    @pytest.mark.it("正常系")
    @mock_dynamodb()
    def test_lambda_handler_success(self, mock_table):
        # テスト対象をimport
        from functions.api import app
        importlib.reload(app)

        # mock_tableの戻り値であるcreate_authority_tableを実行
        # 前処理のほうではmock_tableを実行しておくだけなので
        # create_authority_tableはこちらでやる
        mock_table()

        # テストデータを用意
        event = {'body': {'id': '0001', 'name': 'satoh'}}

        # テスト対象を呼び出す
        data = app.lambda_handler(event)

        # 検証
        assert len(data) != 0

終わりに

今回は実行時のキャプチャーを省略しますが、mock_dynamodbの書き方を試してみました