AWS Lambdaに対するユニットテスト
AWS Lambda に対するユニットテスト(以下、UT)で気をつけていることを言語化して、記録に残します。
結論
AWSサービスに関わらない処理ステップを切り出して、先にテスト駆動開発で実装する、という考え方をしています。
はじめに
AWS Lambda は、言わずと知れた FaaS(Function as a Service) です。 開発単位が関数であるために気軽につくりはじめることができますが、 心のままにコードを書いてしまうと、節々にAWSの要素が入り乱れてしまい、 結果としてUTができない成果物になってしまうことがありました。
こうなってしまうと、取りうる手段は二択です。
- リファクタリングする
- そのままデプロイする
FaaS という性質から、後者になりがちだと思われます。そしてその先には、手元のIDEでデバッグするのとは程遠い生産性が待っていました。 そうした課題を抱えつつ、ひよこさんである僕が試行錯誤をした結果、このように進めていくとよいのでは、というものが形になってきました。それを言語化し記録するため、以下に記載していきます。
実現したいこと
Lambda Function の開発体験を良い感じにしたい
考えたこと
設計段階からデプロイ後のテストまで、以下の流れで進めることを考えました。
- 実現したい処理内容を中間生成物を軸に分類し、自然言語で書き起こします
- AWSサービスが関わる処理と関わらない処理に分類します
- 関わらない処理を、テスト駆動開発の方式で実装します
- 関わる処理は、適当なモック実装に留めておきます
- 関わる処理に関する文章を精査し、関わらない処理として切り出す余地がないかをチェックします
- 見つかった場合は、上述のとおりに実装します
- UTを継続的に実行しつつ、関わる処理を実装します
- AWS SAM Localを使って、Lambda Function 全体のテストを行います
- AWSへデプロイします
- 開発対象のサービス全体のテストを行います
今回のテーマは UT なので、AWS SAM Local についてはまた今後にします。
工夫したこと
UT対象となる関数を実装する位置がポイントかと思います。
例えば Python の場合、マネージメントコンソールや AWS SAM で Function を新規作成すると、デフォルトで lambda_handler 関数を作成します。これはエントリポイントとして機能しますが、この中では、UT対象となるような処理は記述しません。基本的に、別途作成していく関数の呼び出しだけを行います。
def lambda_handler(event, context):
try:
result_01: str = function_01_Non_AWS(event) # UT対象
result_02: int = function_02_AWS(result_01)
result_03: dict = function_03_Non_AWS(result_02) # UT対象
return result_03
except Exception as e:
raise e
def function_01_Non_AWS(event: dict) -> str:
# AWSを使わない処理
def function_02_AWS(result_01: str) -> int:
# AWSを使う処理
def function_03_Non_AWS(result_02: int) -> dict:
# AWSを使わない処理
AWSが提示するサンプルコードのように、 lambda_handler の中で UT 対象である処理ステップを記述してしまうと、UT と "Function 全体のテスト" の区別がつかなくなってしまいます。
具体例
S3 と Lambda Function による、CSV -> Parquet 変換を考えます。
なお、ランタイムは Python 3.6 以上、テストツールは pytest を使います。
処理内容
日本語で書き下すと、以下のように表現できます。
この中で、AWSのサービスが関わらない部分は 1, 3, 4 のステップです。これら3つが、UTの対象です。
ディレクトリ構造
[admin@centos7 aws]$ tree . . ├── src │ └── pq_converter.py └── tests ├── conftest.py ├── data │ ├── c01.csv │ ├── c01.parquet │ └── s3_event.json ├── __init__.py └── test_pq_converter.py
(ubuntuに移行しようかな…)
テストデータ
c01.csv
- 適当に作成
s3_event.json
- Amazon S3 Event Message Structure
- これはイベントメッセージの構造を示している例なので、いくつかのValueをテスト用に編集しています
Lambda Function
pq_converter.py が本体です。
下記のように、前述の日本語文と各関数が対応しています。
- イベント情報から S3 Bucket と File Key の情報を取得する : read_s3_event()
- S3 Bucket からファイルの保持するデータを取得 : get_s3_data()
- 取得したデータを Pandas.DataFrame に変換 : make_df()
- Pandas.DataFrame から Parquetファイルを作成 : create_pq()
- Parquetファイルを S3 Bucket にアップロード : upload_file()
テスト共通処理
テストコードでの事前処理や共通関数は、conftest.py にまとめて記述します。
pq_converter.py をimportするのはこのファイルだけです。
テストコード
test_pq_converter.py で、テスト対象の関数ごとにテストコードをクラス化します。これをすることで、pytest -v でテスト結果をグルーピングして表示できます。
conftest.py に記述した共通関数のうち、必要なものをクラスに fixture で指定して関連付けます。
クラスの中に、テストケース単位で関数を作成します。アサーションルーレットを回避するため、関数とアサーションは 1 : 1 の関係にします。
[admin@centos7 tf_serverless]$ 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 collected 5 items aws/tests/test_pq_converter.py::TestReadS3Event::test_bucket PASSED [ 20%] aws/tests/test_pq_converter.py::TestReadS3Event::test_key PASSED [ 40%] aws/tests/test_pq_converter.py::TestMakeDataframe::test_df_row_length PASSED [ 60%] aws/tests/test_pq_converter.py::TestMakeDataframe::test_df_col_length PASSED [ 80%] aws/tests/test_pq_converter.py::TestCreateParquet::test_is_exist_parquet PASSED [100%] =========================================== 5 passed in 0.96s ============================================
すべて PASS していますが、仮にNGがあったとしてもどのテストケースで出たのか、わかりやすいかと思います。
まとめ
- 設計段階でクラウドサービスに関わるか、そうでないかを分類するといろいろ捗ります
- テスト駆動開発は心の安定につながります
- (書くの忘れてたけど)AWS Glue Job でも同じようなアプローチがとれます
次回は、AWS SAM について書くかもしれません。
