[AWS] Lambda関数のユニットテストをローカル環境で実行してみよう

前提

今回は、以下のような前提で検証してみたいと思います。

• SAM
• Lambda
• Python3.8
• pytest
• moto（モック）
• DynamoDB

なお、SAMを使用すると、Dockerん環境を使って擬似的にローカル環境にDynamoDBなどを構築することができますが、今回は、ユニットテスト（将来的な自動テスト）を見据えてのテストについて検証してみたいと思います。

プロジェクト作成

では、まず最初にプロジェクトを作成します。
もし、SAMでのプロジェクト作成がよくわからない方は、事前に

• [AWS] Serverless Application Model (SAM) の基本まとめ
• [AWS] Serverless Application Model (SAM) でAPI Gateway + Lambda + DynamoDBなサンプルを作成してみる

に目を通されることをおすすめします。

$ sam init --runtime=python3.8
Which template source would you like to use?
	1 - AWS Quick Start Templates
	2 - Custom Template Location
Choice: 1

Project name [sam-app]:

Cloning app templates from https://github.com/awslabs/aws-sam-cli-app-templates.git

AWS quick start application templates:
	1 - Hello World Example
	2 - EventBridge Hello World
	3 - EventBridge App from scratch (100+ Event Schemas)
	4 - Step Functions Sample App (Stock Trader)
	5 - Elastic File System Sample App
Template selection: 1

-----------------------
Generating application:
-----------------------
Name: sam-app
Runtime: python3.8
Dependency Manager: pip
Application Template: hello-world
Output Directory: .

Next steps can be found in the README file at ./sam-app/README.md

必要なパッケージのインストール

ユニットテストを実行するために必要なPythonのパッケージをインストールします。

$ pipenv install pytest pytest-mock mocker moto --dev

ユニットテスト

以下のコマンドで、Lamnda関数を呼び出して、各Lambda関数のユニットテストを実行できます。
まずは、デフォルトで作成されているHelloWorldのLambda関数のテスト（こちらのテストコードもデフォルトで作成済み）を実行してみましょう。

$ python -m pytest tests
============================= test session starts ==============================
platform darwin -- Python 3.8.5, pytest-6.0.1, py-1.9.0, pluggy-0.13.1
rootdir: /Users/******/aws/github/sam-app
plugins: mock-3.3.0
collected 1 item

tests/unit/test_handler.py .                                             [100%]

============================== 1 passed in 0.02s ===============================

1件のテストが成功したことがわかります。

DynamoDBアクセスコードの追加

今回は、ユニットテストすることだけを目的とするため、特にSAMの設定変更等は行わずに、テストに必要な変更のみを行うようにします。
では、DynamoDBにアクセス（レコード追加）する処理の追加を行ってみます。

hello_world/app.py

import json
import boto3
import os
from datetime import datetime

def lambda_handler(event, context):
    try:
        event_body = json.loads(event["body"])
        dynamodb = boto3.resource("dynamodb")

        table = dynamodb.Table("Demo")
        table.put_item(
            Item={
                "Key": event_body["test"],
                "CreateDate": datetime.utcnow().isoformat()
            }
        )

        return {
            "statusCode": 200,
            "body": json.dumps({
                "message": "hello world",
            }),
        }
    except Exception as e:
        return {
            "statusCode": 500,
            "body": json.dumps({
                "message": e.args
            }),
    }

テストコードに、DynamoDBのモックアップを追加

tests/unit/test_handler.py

import boto3
import json
import pytest
from hello_world import app
from moto import mock_dynamodb2

@pytest.fixture()
def apigw_event():
    """ Generates API GW Event"""

    return {
        "body": '{ "test": "body"}',
        "resource": "/{proxy+}",
        "requestContext": {
            "resourceId": "123456",
            "apiId": "1234567890",
            "resourcePath": "/{proxy+}",
            "httpMethod": "POST",
            "requestId": "c6af9ac6-7b61-11e6-9a41-93e8deadbeef",
            "accountId": "123456789012",
            "identity": {
                "apiKey": "",
                "userArn": "",
                "cognitoAuthenticationType": "",
                "caller": "",
                "userAgent": "Custom User Agent String",
                "user": "",
                "cognitoIdentityPoolId": "",
                "cognitoIdentityId": "",
                "cognitoAuthenticationProvider": "",
                "sourceIp": "127.0.0.1",
                "accountId": "",
            },
            "stage": "prod",
        },
        "queryStringParameters": {"foo": "bar"},
        "headers": {
            "Via": "1.1 08f323deadbeefa7af34d5feb414ce27.cloudfront.net (CloudFront)",
            "Accept-Language": "en-US,en;q=0.8",
            "CloudFront-Is-Desktop-Viewer": "true",
            "CloudFront-Is-SmartTV-Viewer": "false",
            "CloudFront-Is-Mobile-Viewer": "false",
            "X-Forwarded-For": "127.0.0.1, 127.0.0.2",
            "CloudFront-Viewer-Country": "US",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
            "Upgrade-Insecure-Requests": "1",
            "X-Forwarded-Port": "443",
            "Host": "1234567890.execute-api.us-east-1.amazonaws.com",
            "X-Forwarded-Proto": "https",
            "X-Amz-Cf-Id": "aaaaaaaaaae3VYQb9jd-nvCd-de396Uhbp027Y2JvkCPNLmGJHqlaA==",
            "CloudFront-Is-Tablet-Viewer": "false",
            "Cache-Control": "max-age=0",
            "User-Agent": "Custom User Agent String",
            "CloudFront-Forwarded-Proto": "https",
            "Accept-Encoding": "gzip, deflate, sdch",
        },
        "pathParameters": {"proxy": "/examplepath"},
        "httpMethod": "POST",
        "stageVariables": {"baz": "qux"},
        "path": "/examplepath",
    }

@mock_dynamodb2
def test_lambda_handler(apigw_event, mocker):
    dynamodb = boto3.resource('dynamodb')
    dynamodb.create_table(
        TableName='Demo',
        KeySchema=[
            {
                'AttributeName': 'Key',
                'KeyType': 'HASH'
            },
            {
                'AttributeName': 'CreateDate',
                'KeyType': 'RANGE'
            }
        ],
        AttributeDefinitions=[
            {
                'AttributeName': 'Key',
                'AttributeType': 'S'
            },
            {
                'AttributeName': 'CreateDate',
                'AttributeType': 'S'
            }
        ],
        ProvisionedThroughput={
            'ReadCapacityUnits': 10,
            'WriteCapacityUnits': 10
        }
    )

    ret = app.lambda_handler(apigw_event, "")
    data = json.loads(ret["body"])

    assert ret["statusCode"] == 200
    assert "message" in ret["body"]
    assert data["message"] == "hello world"
    # assert "location" in data.dict_keys()

主な修正点は、

• importに必要な定義追加
• test_lambda_handler前に@mock_dynamodb2追加
• test_lambda_handlerで、Lambda関数呼び出し前に擬似的にDynamoDBのテーブル作成

です。

ユニットテスト実行

では、早速ユニットテストを実行してみます。

$ python -m pytest tests
============================= test session starts ==============================
platform darwin -- Python 3.8.5, pytest-6.0.1, py-1.9.0, pluggy-0.13.1
rootdir: /Users/******/aws/github/sam-app
plugins: mock-3.3.0
collected 1 item

tests/unit/test_handler.py .                                             [100%]

=============================== warnings summary ===============================
[pytest]
/Users/******/.local/share/virtualenvs/sam-app-3Tr4jFKA/lib/python3.8/site-packages/boto/plugin.py:40
  /Users/******/.local/share/virtualenvs/sam-app-3Tr4jFKA/lib/python3.8/site-packages/boto/plugin.py:40: DeprecationWarning: the imp module is deprecated in favour of importlib; see the module's documentation for alternative uses
    import imp

/Users/******/.local/share/virtualenvs/sam-app-3Tr4jFKA/lib/python3.8/site-packages/moto/cloudformation/parsing.py:407
  /Users/******/.local/share/virtualenvs/sam-app-3Tr4jFKA/lib/python3.8/site-packages/moto/cloudformation/parsing.py:407: DeprecationWarning: Using or importing the ABCs from 'collections' instead of from 'collections.abc' is deprecated since Python 3.3, and in 3.9 it will stop working
    class ResourceMap(collections.Mapping):

-- Docs: https://docs.pytest.org/en/stable/warnings.html

一応、成功していますが、警告が出ています。
これは、現在のPythonのバージョンでは、motoなどで使用している一部機能が使用停止されていることを表しています。
今回のユニットテストには影響のない警告なので、プロジェクトのホームにpytest.iniという名称でファイルを作成し、以下を記述して保存してみましょう。

pytest.ini

[pytest]
filterwarnings =
    ignore::DeprecationWarning

では、再度実行してみましょう。

$ python -m pytest tests
============================= test session starts ==============================
platform darwin -- Python 3.8.5, pytest-6.0.1, py-1.9.0, pluggy-0.13.1
rootdir: /Users/******/aws/github/sam-app, configfile: pytest.ini
plugins: mock-3.3.0
collected 1 item

tests/unit/test_handler.py .                                             [100%]

============================== 1 passed in 2.10s ===============================

今度は警告なく終了しました。

まとめ

ユニットテストは、ソースコード変更時に自動実行されることが望ましいです。
今回は、ユニットテストの実行をローカルで行っているため、ソースコード部分のみの修正に留めていますが、CI/CDのパイプラインの延長で行う場合は、requirements.txtや各定義ファイルにも修正が必要です。
次回は、その辺のパイプラインとの結合も含めた検証を、どこかのタイミングでやってみたいと思います。

サンプルコードリポジトリ