[Python] motoでLambdaやDynamoDBをモックしてunittestを実行

目的

• Pythonで書いたLambdaのAPIを標準ライブラリのunittestを使ってテスト。
• motoを使ってLambdaとDynamoDBをモックすることで、AWSに接続せずにunittestを動かす。
  

 

motoとは？

LambdaやDynamoDBに限らず、S3など様々なAWSサービスをモック出来る優れものです。

ただし、全てのAWSサービスに対応しているわけではないで、motoのドキュメントから対応サービスを確認してください。

ディレクトリ階層

root
├ apis
│ └ handler.py ← Lambdaのソースコード
└ test
   └ test_handler.py ← handler.pyに対するテストコード

ライブラリのインストール

$ pip install boto3
$ pip install docker
$ pip install moto

各バージョン

• Python 3.8.5
• boto3 1.20.7
• docker 5.0.3
• moto 2.2.15
  

Lambdaの実装（handler.py）

リクエストパラメータのnameをDynamoDBのtest-tableという名前のテーブルに登録するだけのAPI。

import boto3
import json
import uuid
from http import HTTPStatus


DYNAMO_DB = boto3.resource('dynamodb', region_name='ap-northeast-1')
TEST_TABLE = DYNAMO_DB.Table('test-table')


def handler(event, context):
    # リクエストパラメータ取得
    body = json.loads(event.get('body') or '{}')
    # DynamoDBに登録
    item = {
        'id': str(uuid.uuid4()),
        'name': body.get('name')
    }
    TEST_TABLE.put_item(Item=item)
    # 処理結果返却
    return {
        'statusCode': HTTPStatus.OK,
        'body': json.dumps({'message': 'DB登録完了'})
    }

テストコードの実装（test_handler.py）

import boto3
import json
import unittest
from apis import handler
from http import HTTPStatus
from moto import mock_lambda, mock_dynamodb2


# テスト対象のテーブル名
TABLE_NAME = 'test-table'


# Lambdaのeventを生成
def create_event(body):
    return {
        'body': json.dumps(body),
        'requestContext': {}
    }


class TestHandler(unittest.TestCase):

    # テスト実行前の処理
    def setUp(self):
        pass

    # テスト実行後の処理
    def tearDown(self):
        pass

    # motoにテーブルを作成
    def create_table(self):
        db = boto3.resource('dynamodb', region_name='ap-northeast-1')
        db.create_table(
            TableName=TABLE_NAME,
            KeySchema=[
                {
                    'KeyType': 'HASH',
                    'AttributeName': 'id'
                },
                {
                    'KeyType': 'RANGE',
                    'AttributeName': 'name'
                }
            ],
            AttributeDefinitions=[
                {
                    'AttributeName': 'id',
                    'AttributeType': 'S'
                },
                {
                    'AttributeName': 'name',
                    'AttributeType': 'S'
                }
            ],
            BillingMode='PAY_PER_REQUEST'
        )
        table = db.Table(TABLE_NAME)
        handler.TEST_TABLE = table
        return table

    # テスト実行
    @mock_dynamodb2
    @mock_lambda
    def test_handler(self):
        # テーブル作成
        table = self.create_table()
        # リクエストパラメータ
        name = 'moto-test'
        request = {
            'name': name,
        }
        # Lambda実行
        response = handler.handler(create_event(request), {})
        body = json.loads(response['body'])
        # HTTPステータスチェック
        assert HTTPStatus.OK == response['statusCode']
        # レスポンスメッセージチェック
        assert body['message'] == 'DB登録完了'
        # DB登録結果確認
        data = table.scan()['Items']
        assert len(data) == 1
        assert data[0]['name'] == name

テスト実施

$ python -m unittest discover test/
.
----------------------------------------------------------------------
Ran 1 test in 0.192s

OK

解説

moto から Lambda と DynamoDB のモックするデコレータをimportする。

from moto import mock_lambda, mock_dynamodb2

テスト関数に Lambda と DynamoDB のモックするデコレータを設定。

このデコレータを設定した関数から呼び出される Lambda と DynamoDB はAWSに接続せず、motoのモックを参照します。

    # テスト実行
    @mock_dynamodb2
    @mock_lambda
    def test_handler(self):
        # テーブル作成
        table = self.create_table()

motoにテーブルを作成する。

AWS上のDynamoDBにテーブルがあったとしても、モックしているmotoになければエラーになるので、実際と同じ定義のテーブルを作成します。

    def create_table(self):
        db = boto3.resource('dynamodb', region_name='ap-northeast-1')
        db.create_table(
            TableName=TABLE_NAME,
            KeySchema=[
                {
                    'KeyType': 'HASH',
                    'AttributeName': 'id'
                },
                {
                    'KeyType': 'RANGE',
                    'AttributeName': 'name'
                }
            ],
            AttributeDefinitions=[
                {
                    'AttributeName': 'id',
                    'AttributeType': 'S'
                },
                {
                    'AttributeName': 'name',
                    'AttributeType': 'S'
                }
            ],
            BillingMode='PAY_PER_REQUEST'
        )
        table = db.Table(TABLE_NAME)
        handler.TEST_TABLE = table
        return table

注意点

モックの方法を間違えるとAWSに繋ぎに行ってしまうため、既存リソースに影響を与えてしまう可能性があります。

なのでmotoを使ってテスト方法を試行錯誤する場合は、こまめにAWSのコンソール画面を確認しに行って、テストデータが作られていないか確認することをオススメします。

まぁ、そもそもローカルからAWSに繋がるような設定をしないのが１番安全ですがね。