pytest, SAM, LocalStackを使ったAWS Lambda結合テスト

import os
import boto3
from botocore.config import Config
from pq_converter import LambdaProcessor

endpoint = os.environ["ENDPOINT"]
endpoint_url = os.environ["ENDPOINT_URL"]

# boto3 session
session = boto3.Session()

if endpoint == "localstack":
    print("Start Testing with Localstack")
    s3 = session.resource("s3", endpoint_url=endpoint_url, config=Config())
else:
    s3 = session.resource("s3")


def lambda_handler(event, context) -> dict:
    processor = LambdaProcessor(event=event, context=context, s3=s3)
    return processor.main()

pytest

conftest.py で LambdaProcessorを実体化して、テストコード全体で利用しています。

./tests/conftest.py

@pytest.fixture(scope="session")
def processor():
    with open("./tests/data/s3_event.json", "r") as f:
        test_event = json.load(f)

    s3 = boto3.resource("s3", endpoint_url="http://localhost:4566")
    processor = LambdaProcessor(event=test_event, context={}, s3=s3)
    return processor

SAM Local

LocalStackの利用においてもCredentialが必要ですので、ダミーの設定を追加してください。

./.aws/credentials

[localstack]
aws_access_key_id = dummy
aws_secret_access_key = dummy

./.aws/config

[profile localstack]
region = us-east-1
output = text

接続先の切り替えには、CloudFormation テンプレート内のグローバル環境変数を使用しています。

./aws/template.yml

Globals:
    Function:
         Environment:
             Variables: 
                 ENDPOINT: ""
                 ENDPOINT_URL: ""

Lambda Function の依存ライブラリは、Lambda Layer を使って読み込むようにしています。

テスト実行

手抜き感のあるシェルスクリプトです。

./aws/test.sh

環境変数 ENDPOINT_IP に、実行環境のIPアドレスを設定してください。localhost や 127.0.0.1 は動作しませんので注意です。これを自動抽出するシェル力はありませんでした…
sam local invoke コマンド実行時に、LocalStack のコンテナに接続するため、DOCKER_NETWORK_ID をセットしています。docker-compose.ymlのある dir名称 + '_default' という Name で作成されますので、この NETWORK_ID を抽出しています。今回の例では、aws_defaultです。
sam local invoke コマンド実行時に、CloudFormation Template で設定したグローバル環境変数に ./vars.json をセットしています。./vars.json は、前述のIPアドレスを別途指定するという事情から、シェルスクリプト内で生成しています。

テスト実行結果

標準出力結果から、pytest -v の実行結果を抽出しました。

======================================================= test session starts ========================================================
platform linux -- Python 3.7.5, pytest-6.2.1, py-1.10.0, pluggy-0.13.1 -- /home/admin/.pyenv/versions/3.7.5/bin/python3.7
cachedir: .pytest_cache
rootdir: /home/admin/project/tf_serverless/aws
collected 7 items                                                                                                                  


tests/test_pq_converter.py::TestReadS3Event::test_bucket PASSED                                                              [ 14%]
tests/test_pq_converter.py::TestReadS3Event::test_key PASSED                                                                 [ 28%]
tests/test_pq_converter.py::TestGetS3Data::test_is_s3_data_found PASSED                                                      [ 42%]
tests/test_pq_converter.py::TestMakeDataframe::test_row_length PASSED                                                        [ 57%]
tests/test_pq_converter.py::TestMakeDataframe::test_column_length PASSED                                                     [ 71%]
tests/test_pq_converter.py::TestCreateParquet::test_is_exist_file PASSED                                                     [ 85%]
tests/test_pq_converter.py::TestUploadFile::test_is_uploading_succeeded PASSED                                               [100%]

======================================================== 7 passed in 0.42s =========================================================

単体テストの範囲を超えて、AWSサービスに関連するメソッドまでカバーできています。

sam local invoke の結果は、なぜか return の結果がうまく出力されないので、test.log に出力するようにしてみました。

START RequestId: 4a66a353-934c-4ff8-94c8-c0455fa747cd Version: $LATEST
Start Testing with Localstack
{"StatusCode": 200, "Bucket": "tf-serverless-tosh2230", "Key": "c01.parquet"}END RequestId: 4a66a353-934c-4ff8-94c8-c0455fa747cd
REPORT RequestId: 4a66a353-934c-4ff8-94c8-c0455fa747cd  Init Duration: 0.42 ms  Duration: 3053.66 ms    Billed Duration: 3100 ms    Memory Size: 256 MB Max Memory Used: 256 MB

無事、正常終了していることを確認できました。しかし SAM Local Invoke の実行結果が見づらいうえに、想定どおりに動作したのかわかりにくいです。これはもっとスマートに結果がわかるように、今後改善していけたらいいなと思います。それと、実行結果の直後に改行が入っていないのが、わたし気になります。