tl;dr
久しぶりに Python で Lambda ファンクションを書いて, ついてでに AWS リソースにアクセスする部分を moto で置き換えてテストを書いたので振り返っておきます.
お題は直前に作った RSS フィードをパースして Slack に通知するやつ.
モチベーション
Lambda ファンクション自体を作ったモチベーションは以下の記事に書きました.
Lambda ファンクションを書くなかで, AWS リソースにアクセスするタイミングが 2 回あるんですが, この部分を実装していて, これってちゃんと意図した通りに動いてくれるのか不安になったのでテストを書いて動作確認をしてみることにしました.
この Lambda ファンクションでは, DynamoDB と SSM パラメータストアにアクセスすることになりますが, これらのリソースを動作確認の為に用意するのも大げさだなという気持ちがありましたので, 擬似的に DyanamoDB や SSM パラメータストアのレスポンスを返してくれる moto というライブラリを利用してテストを書いていきます.
moto については, 既にいくつかのブログ記事等で紹介されている為, 詳細については割愛させて頂きます. 尚, moto がどのようなレスポンスを返すか (モックするか) については, 以下にリストアップされています.
全てのレスポンスをモックしてくれてるわけではないので注意が必要です.
テストコード
DynamoDB への書き込みに関するテスト
Lambda ファンクション内では, 以下のように DynamoDB のテーブルにデータを書き込んでいます.
def write_table(key, machine, rev): dynamodb = boto3.client('dynamodb', region_name=region) try: dynamodb.put_item( TableName=dynamodb_table, Item={ 'key': {'S': key}, 'machine': {'S': machine}, 'revision': {'S': rev} }, Expected={ 'key': { 'Exists': False } } ) except ClientError as e: if e.response['Error']['Code'] == 'ConditionalCheckFailedException': print('duplicate key') return False else: raise else: return True
この write_table という関数は DynamoDB にmachine と revision, machine と revision を + で結合した文字列を Base64 エンコードした文字列を書き込んで, 書き込みに成功したら関数の呼び出し元に True を返し, ConditionalCheckFailedException の例外が発生したら False を返すことを期待しています.
また, ConditionalCheckFailedException 以外の例外が発生した場合には raise するようにしています. ちなみに, Slack チャンネルに通知済みか否かのフラグとして利用する為に, ConditionalCheckFailedException を個別に補足しています. このような使い方が正しいかどうか置いておいて, この関数を本物の DynamoDB テーブルを使わずに moto のモックを利用して書いたテストが以下のコードです.
import pytest import boto3 import base64 from moto import mock_dynamodb2 from moto import mock_ssm from handler import * class TestHandler: ... @mock_dynamodb2 def test_write_table(self): dynamodb = boto3.client('dynamodb', region_name='ap-northeast-1') dynamodb.create_table( TableName='yamaha-firmware-notify', KeySchema=[ { 'AttributeName': 'key', 'KeyType': 'HASH' } ], AttributeDefinitions=[ { 'AttributeName': 'key', 'AttributeType': 'S' } ] ) enc_message = generate_keystring('RTX1200', 'Rev.10.01.78') result = write_table(enc_message, 'RTX1200', 'Rev.10.01.78') assert True is result ...
デコレータとして @mock_dynamodb2 をつけてあげると boto3 の各関数の振る舞いをモックしてくれるので驚いてしまう.
SSM パラメータストアからパラメータを取得する関数のテスト
SSM パラメータストアからパラメータを取得する以下の関数.
def get_ssm_parameter(param_name): ssm = boto3.client('ssm', region_name=region) try: res = ssm.get_parameter( Name=param_name, WithDecryption=True ) return res except ClientError as e: print('ssm get parameter error.') return ''
get_parameter 関数でパラメータを取得するだけのシンプルな関数です. この関数のテストについても本物のパラメータストアを利用することなく moto の力を借りて以下のように書きました.
...
@mock_ssm
def test_get_ssm_parameter(self):
ssm = boto3.client('ssm', region_name='ap-northeast-1')
ssm.put_parameter(Name='test', Value='string', Type='SecureString')
result = get_ssm_parameter('test')
assert 'string' == result['Parameter']['Value']
@mock_ssm
def test_get_ssm_parameter_not_exists(self):
ssm = boto3.client('ssm', region_name='ap-northeast-1')
ssm.put_parameter(Name='test', Value='string', Type='SecureString')
result = get_ssm_parameter('test1')
assert '' == result
...
DynamoDB と同様にデコレータとして @mock_ssm をつけてあげるだけで, put_parameter や get_parameter の挙動をモックしてくれるようになります.
実際にテストを走らせてみる
手元の端末上でテストを走らせてみます.
$ docker-compose exec myservice pytest --verbose --disable-warnings =============================================================================================== test session starts ================================================================================================ platform linux -- Python 3.8.2, pytest-5.4.1, py-1.8.1, pluggy-0.13.1 -- /usr/local/bin/python cachedir: .pytest_cache rootdir: /work collected 5 items tests/test_handler.py::TestHandler::test_generate_keystring PASSED [ 20%] tests/test_handler.py::TestHandler::test_write_table PASSED [ 40%] tests/test_handler.py::TestHandler::test_write_table_revision_already_exists PASSED [ 60%] tests/test_handler.py::TestHandler::test_get_ssm_parameter PASSED [ 80%] tests/test_handler.py::TestHandler::test_get_ssm_parameter_not_exists PASSED [100%] ========================================================================================== 5 passed, 2 warnings in 4.88s ===========================================================================================
いい感じです.
docker-compose を利用してテストを実行 (pytest を実行) していますが, 実際にテストを走らせる方法は docker-compose に限りません. しかし, Docker コンテナに閉じ込めてしまうことで, 後々, CircleCI や Github Actions 等の CI/CD 環境への取り回しが容易になると考えています.
そして, Github Actions で転がす
初心者が書いた .github/workflows/main.yml
まずはリファレンスを読みましょう.
以下は, 初心者がドキュメントを見よう見真似で書いたワークフローの YAML です.
name: CI on Push on: [push] jobs: pycodestyle: runs-on: ubuntu-latest steps: - name: Checkout code uses: actions/checkout@v1 - name: Build test infrastructure run: | docker-compose build docker-compose up -d - name: Setup dependencies run: | docker-compose exec -T myservice \ pip install -r requirements.txt - name: Run pycodestyle run: | docker-compose exec -T myservice \ pycodestyle ./*.py ./tests/*.py pytest: runs-on: ubuntu-latest steps: - name: Checkout code uses: actions/checkout@v1 - name: Build test infrastructure run: | docker-compose build docker-compose up -d - name: Setup dependencies run: | docker-compose exec -T myservice \ pip install -r requirements.txt - name: Run test run: | docker-compose exec -T myservice \ pytest --verbose --disable-warnings
on: [push] と書いておくと, リポジトリへの push をトリガーとして jobs 以下のジョブが実行されます. 気に入らないのが...
- 各ジョブ間で共通処理をまとめて書いていない
- ジョブ間で環境 (docker-compose build した環境) を共有出来たら嬉しいのに...
CircleCI では同様のことが出来ているので, Github Actions も出来るであろうということで, まだまだ勉強が必要なようです.
実際に転がしてみると
下図は実行結果です.
pycodestyle と pytest のジョブは並列で実行されます. これも並列ではなく, 依存関係を設けることが出来るんだろうなあと思いつつ, まだちゃんと手を動かせていません.
ということで
小さな Lambda ファンクションで, 今後, 大きな改修が行われる可能性は小さいものですが, AWS リソースを叩く部分のテストを書いて継続的な改修に対する安心感みたいなものを得ることが出来ました. また, Github Actions にのっけることで, 誰でもテストを走らせることが出来るし, 結果を共有することが可能となりました. インフラエンジニアであっても, 自分で作った Lambda ファンクションに限らず, ツールやなんやかんやには責任を持つという意味でテストを書くべきだなと思った次第です.
ということで, Python で Lambda ファンクションを書く時には moto は元がとれるくらい便利なので出来るだけ使っていこうと思います (テストを書いていこうと思います)
