Lambdaの拡張機能(Parameter Store)を使ったLambdaコードをUnit Testしてみる

Lambdaの拡張機能(Parameter Store)を使ったLambdaコードをUnit Testしてみる

ローカルAPIサーバーを利用したUnit Testを試してみました。
Clock Icon2024.09.27

Lambdaの拡張機能を使えば、パラメータストアから簡単に値を取得できます。

このLambdaコードに対するUnit Testを試してみました。本記事では、ローカルにAPIサーバを作成しています。

おすすめの方

  • Lambdaの拡張機能(Parameter Store)を使ったLambdaコードに対するUnit Testを作成したい方

LambdaコードとUnit Testを作成する

Lambdaコード

Lambdaの拡張機能を利用して、パラメータストアから値を取得しています。

app.py
import os
import json
import requests
import urllib.parse

LAMBDA_EXTENSIONS_SSM_BASE_URL = "http://localhost:2773/systemsmanager/parameters/get"

def lambda_handler(event, context):
    return {
        "statusCode": 200,
        "body": json.dumps(
            {
                "message": get_parameter("/Blog/Sample/TEST/message"),
            }
        ),
    }

def get_parameter(name):
    headers = {
        "X-Aws-Parameters-Secrets-Token": os.environ.get("AWS_SESSION_TOKEN"),
    }
    url = f"{LAMBDA_EXTENSIONS_SSM_BASE_URL}?name={urllib.parse.quote(name)}"
    resp = requests.get(url, headers=headers)
    return resp.json()["Parameter"]["Value"]

Unit Testコード

シンプルなテストコードを用意しました。pytestを利用します。

test_app.py
import json
import pytest
import app

def test_lambda_handler():

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

    assert ret["statusCode"] == 200
    assert "message" in ret["body"]
    assert data["message"] == "hello world"

Unit Testを実行する

テストを実行すると、ConnectionErrorが発生しました。ここまでは期待通りです。

$ pytest test_app.py

>           raise ConnectionError(e, request=request)
E           requests.exceptions.ConnectionError: HTTPConnectionPool(host='localhost', port=2773): Max retries exceeded with url: /systemsmanager/parameters/get?name=/Blog/Sample/TEST/message (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x1064ebad0>: Failed to establish a new connection: [Errno 61] Connection refused'))

ローカルにAPIサーバを作って、Unit Testを実行する

Uvicornをインストールする

pip install "uvicorn[standard]"

Unit Test用のAPIを作成する

main.pyを作成します。内容的には、「/systemsmanager/parameters/get」に対するリクエスト時に動く関数を作成し、「name」が目的の値と一致した場合にパラメータを返すだけです。

main.py
from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/systemsmanager/parameters/get")
def get_parameter(name):
    if name == "/Blog/Sample/TEST/message":
        return {
            "Parameter": {
                "Value": "hello world",
            },
        }

    # 404は適当です
    raise HTTPException(status_code=404, detail="not found")

ローカルサーバーを起動する

ローカルサーバーを2773ポートで起動します。

$ uvicorn main:app --port 2773

INFO:     Started server process [96666]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:2773 (Press CTRL+C to quit)

改めてUnit Testを実行する

改めてテストを実行すると、PASSしました。バッチリですね。

$ pytest test_app.py

...略...

collected 1 item                                                               

test_app.py .                                                            [100%]

============================== 1 passed in 0.08s ===============================

ローカルAPIサーバーを起動したコンソールには、次のログが表示されていました。問題なくアクセスできたことが分かります。

INFO:     127.0.0.1:63330 - "GET /systemsmanager/parameters/get?name=/Blog/Sample/TEST/message HTTP/1.1" 200 OK

さいごに

ローカルAPIサーバーを利用したUnit Testを試してみました。どなたかの参考になれば幸いです。

参考

この記事をシェアする

facebook logohatena logotwitter logo

© Classmethod, Inc. All rights reserved.