MotoでLambdaコードからLambdaモックを呼び出す（Dockerなし）

はじめに

この記事では、次のコードをMotoでテストする方法を記載します。

• Pythonランタイム上で動作するLambda関数
• Lambda関数内で、別のLambda関数をboto3経由で呼び出す

Motoとは

Motoは、Boto3をモックするパッケージです。
Boto3はPythonのAWS SDKの名称であり、MotoはBoto3が対応する多くのAWSサービスをモックできます。

Lambda関数の実装

まずは、テスト対象となるLambda関数を実装します。今回は以下のコードを使用します。
このLambdaコードでは、別のLambdaを呼び出した結果からdataを抽出し、そのままレスポンスとして返却しています。

import json
import os
from typing import Dict

import boto3

REGION_NAME = os.environ["REGION_NAME"]
LAMBDA_FUNC_NAME = os.environ["LAMBDA_FUNC_NAME"]

lambda_client = boto3.client("lambda", region_name=REGION_NAME)


def lambda_handler(event, context):
    res = lambda_client.invoke(
        FunctionName=LAMBDA_FUNC_NAME,
        InvocationType="RequestResponse",
        Payload=json.dumps({}),
    )
    if res["StatusCode"] != 200:
        print("Error")
        return None
    payload: Dict = json.loads(res["Payload"].read().decode("utf-8"))
    if "data" not in payload:
        print("Error")
        return None
    response = {"data": payload["data"]}
    return response

REGION_NAMEとLAMBDA_FUNC_NAMEは環境変数から取得しており、Lambda関数の環境変数設定及びローカルの.envで指定を前提としています。

Lambdaクライアントへのリージョン指定

Lambda上でコードを実行する場合、クライアントのリージョン名指定がなくても動作しますが、Motoを使う場合は未指定だと動きませんでした。
指定しない場合は下記のエラーが起こります。

KeyError: 'ap-north-east-1'

呼び出し先のLambda関数コード

テスト対象から呼び出すLambda関数コードは、次の通りごくシンプルなものとします。

import json

def lambda_handler(event, context):
    return {
        'data': 1
    }

なお実際にLambdaで呼び出す場合は、テスト対象のLambda関数の実行ロールに対して、呼び出し先のLambda関数のInvokeFunctionを実行できる権限が付与されていないと呼び出しに失敗します。

テストコードの実装

コード全文は次のとおりです。

import os

import requests
from moto import mock_aws

os.environ["REGION_NAME"] = "ap-northeast-1"
os.environ["LAMBDA_FUNC_NAME"] = "lambda-func-name"

from example.test_lambda.lambda_function import (
    lambda_handler,
)

REGION_NAME = os.environ["REGION_NAME"]
LAMBDA_FUNC_NAME = os.environ["LAMBDA_FUNC_NAME"]


@mock_aws(
    config={
        "lambda": {"use_docker": False},
    }
)
def test_lambda_handler():
    expected_results = {
        "results": ["""{"data": 1}"""],
        "region": REGION_NAME,
    }
    resp = requests.post(
        "http://motoapi.amazonaws.com/moto-api/static/lambda-simple/response",
        json=expected_results,
    )
    assert resp.status_code == 201

    response = lambda_handler({}, None)
    assert response["data"] == 1

ここからは詳細を記述します。

インストール

pytestは設定されている前提とします。
MotoはモックするAWSサービスごとにインストールすることが可能です（参考）。Lambdaをモックする場合、次のコマンドを実行します。

pip3 install 'moto[lambda]'

今回はLambdaモックにDockerを使用しない設定を使いますが、使用する場合はMotoのIAMインストールとDocker環境構築も必要となります。

Mypyなどの型チェックツールを利用している場合、検証エラー回避のためtypes-requestsを合わせて追加します。

pip3 install types-requests

テスト関数の作成

デコレータ@mock_awsを指定して、テスト関数を作成します。

import os

import requests
from moto import mock_aws

os.environ["REGION_NAME"] = "ap-northeast-1"
os.environ["LAMBDA_FUNC_NAME"] = "lambda-func-name"

from example.test_lambda.lambda_function import (
    lambda_handler,
)

REGION_NAME = os.environ["REGION_NAME"]
LAMBDA_FUNC_NAME = os.environ["LAMBDA_FUNC_NAME"]


@mock_aws(
    config={
        "lambda": {"use_docker": False},
    }
)
def test_lambda_handler():

環境変数の設定

試したところ、以下の場合にはテスト実行結果がエラーになりました。

• テスト対象コードで環境変数を参照しており、環境変数の定義が.envかテストファイル内にない
• テスト対象コードで環境変数を参照しており、環境変数の定義（os.environ）をテストファイルで行っており、かつテスト対象ファイルのインポートが環境変数の定義より前に記述されている

今回は環境変数をテストファイル内で定義しているため、その後にテスト対象コードのインポートを行っています。.envと同じ値を用いる場合は省略できます。

デコレータの指定

@mock_awsを指定している関数内では、Boto3の動作がMotoによってモックされます。（参考）
configの設定値によって、Lambdaモック時にDockerを使用しないように指定しています。（参考）

Lambda関数レスポンスの設定

MotoのLambdaモックは、デフォルトでDockerを利用します。今回はDockerを利用しない方式で、レスポンスデータをカスタマイズします。
モック用のURL（http://motoapi.amazonaws.com/moto-api/static/lambda-simple/response）に値を送信することで、レスポンスを設定できます。

def test_lambda_handler():
    expected_results = {
        "results": ["""{"data": 1}"""],
        "region": REGION_NAME,
    }
    resp = requests.post(
        "http://motoapi.amazonaws.com/moto-api/static/lambda-simple/response",
        json=expected_results,
    )
    assert resp.status_code == 201

複数回のLambda呼び出しのレスポンスデータ設定

上記では1回分のレスポンスを指定していますが、複数回呼び出す場合は配列形式で指定できます。

    expected_results = {
        "results": ["""1回目のレスポンス""", """2回目のレスポンス"""],
        "region": REGION_NAME,
    }

テスト実行

テスト対象関数を呼び出し、通常の判定を行います。

def test_lambda_handler():
    # (中略)
    response = lambda_handler({}, None)
    assert response["data"] == 1