💻
【pytest】AWS Lambda のテストコードを書くよ
概要
Lambda function を AWS SDK for Python（Boto3）で実装している人向け。
Q&A
Q. AWSサービスのモックに特化したmotoは利用しないのですか？
以下の理由から使うのをやめました（あくまで個人の感想です）
-
motoにて作成したリソースのパラメータがboto3最新バージョンで取得できるパラメータと乖離があることがあった -
motoにて利用できないサービスが多い
Q. unittestモジュールではだめですか？
良いです。テストすることが大事です。
Q. カバレッジ100%を目指すべきですか？
何が何でも目指す必要はありません。テストが不要な部分はテスト対象から除外しましょう。
除外したことを設定値やドキュメンテーションなどで残すことはしましょう（ただ除外しただけだと他の人はわかりません）
Q. テストコードを書くタイミングは？
ソースコーディングと並行して実施するのが良いと思います。
関数や条件を作成するたびにテストコードを記述するようにしましょう。
Q. テスト中にAWS環境を変更してしまいますか？
API通信を適切にモックすれば変更することはありません。
前提条件
以下のバージョンで動作しました。
$ lsb_release -d
Description: Ubuntu22.04.3
$ pip list installed
Package Version
------------------ -----------
Python 3.12.4
pip 24.0
boto3 1.37.7
pytest 8.3.5
pytest-cov 6.0.0
pytest-mock 3.14.0
coverage 7.6.12
ファイル構成
$ tree -L 2 boto3/
boto3/
├── rds
│ ├── lambda_function.py # Lambda ソースコード
│ └── test_lambda_function.py # テストコード
├── pytest.ini # pytest 設定ファイル
└── .coveragerc # pytest-cov 設定ファイル
テスト実行方法
# 通常実行
pytest rds/test_lambda_function.py
# 特定のテストのみ実行
pytest rds/test_lambda_function.py -m "t001"
# 標準出力ありで実行
pytest -s rds/test_lambda_function.py
# カバレッジ計測ありで実行
pytest --cov --cov-branch -v rds/test_lambda_function.py --cov-report=term-missing
カバレッジ出力例
カバレッジとは網羅率です。ソースコードがどのくらいテストされているかを理解するのに役立つ指標です。
以下の例は100%ですが、網羅されていないコードがある場合にはMissing列に行数を出してくれます。
$ python --cov -v rds/test_lambda_function.py --cov-report=term-missing
================================================================== test session starts ==================================================================
platform linux -- Python 3.12.4, pytest-8.3.5, pluggy-1.5.0 -- /home/suraud/work/.venv/bin/python
cachedir: .pytest_cache
rootdir: /home/suraud/work/boto3/rds
configfile: pytest.ini
plugins: mock-3.14.0, cov-6.0.0
collected 6 items
rds/test_lambda_function.py::TestLambdaHandler::test_t001 PASSED [ 16%]
rds/test_lambda_function.py::TestLambdaHandler::test_t002 PASSED [ 33%]
rds/test_lambda_function.py::TestLambdaHandler::test_t003 PASSED [ 50%]
rds/test_lambda_function.py::TestLambdaHandler::test_t004 PASSED [ 66%]
rds/test_lambda_function.py::TestLambdaHandler::test_t005 PASSED [ 83%]
rds/test_lambda_function.py::TestLambdaHandler::test_t006 PASSED [100%]
---------- coverage: platform linux, python 3.12.4-final-0 -----------
Name Stmts Miss Branch BrPart Cover Missing
---------------------------------------------------
TOTAL 63 0 18 0 100%
1 file skipped due to complete coverage.
=================================================================== 6 passed in 0.70s ===================================================================
テストコードの紹介
先日諸事情でRDSの再起動・起動確認をするLambda関数を作成したので、このソースコードを例にテストコードの記述を紹介させていただきます。
Lambdaソースコード
lambda_function.py
Lambda引数で指定したアクションを実行します。
アクションは以下の2種類用意しました。
-
action == 'reboot': RDSを再起動 -
action == 'describe_status': RDSのステータスを取得
boto3/rds/lambda_function.py
import os
import json
import boto3
import logging
logger = logging.getLogger()
logger.setLevel(os.environ.get('LOG_LEVEL', 'DEBUG'))
class RDSClient:
def __init__(self):
self.client = boto3.client('rds')
def _check_response(self, response):
""" API 呼び出しのHTTPステータスコードを確認する """
status_code = response['ResponseMetadata']['HTTPStatusCode']
if status_code != 200:
logger.error(f'API呼び出しのHTTPステータスコードが不正です。HTTPステータスコード: {status_code}')
raise
return
def reboot_db_instance(self, rds_id: str):
""" API: RDS再起動 """
try:
response = self.client.reboot_db_instance(
DBInstanceIdentifier = rds_id,
ForceFailover = False # フェイルオーバーしない
)
except Exception as e:
logger.error('RDS再起動処理中にエラーが発生しました。')
raise
self._check_response(response)
return response
def describe_db_instances(self, rds_id: str):
""" API: RDS情報取得 """
try:
response = self.client.describe_db_instances(
DBInstanceIdentifier = rds_id
)
except Exception as e:
logger.error('RDS情報取得処理中にエラーが発生しました。')
raise
self._check_response(response)
return response
def lambda_handler(event, context):
"""
action == reboot: 対象のRDSを再起動します。
action == describe_status: 対象のRDSの起動状態を取得します。
Parameters
----------
event : dict
action : str of action(reboot or describe_status)
rds_id : str of RDS Identifier
Returns
-------
if action == 'reboot':
None
if action == 'describe_status':
dict
rds_status : str of rds status
None
"""
if 'rds_id' not in event or 'action' not in event: # T001
logger.error('引数が不正です。')
raise
rds_id = event['rds_id']
action = event['action']
rds_client = RDSClient()
""" RDS再起動処理 """
if action == 'reboot': # T002
logger.info(f'RDS再起動処理を開始します。\nrds_id: {rds_id}, action: {action}')
response = rds_client.reboot_db_instance(rds_id)
logger.debug(json.dumps(response, default=str, indent=2))
logger.info('RDS再起動処理が完了しました。')
return
""" RDSの起動状態取得 """
if action == 'describe_status':
logger.info(f'RDS起動状態の取得処理を開始します。\nrds_id: {rds_id}, action: {action}')
response = rds_client.describe_db_instances(rds_id)
if len(response['DBInstances']) != 1: # T003
logger.error(f'取得したRDSの情報が不正です。\n{json.dumps(response, default=str, indent=2)}')
raise
rds_status = response['DBInstances'][0]['DBInstanceStatus']
if rds_status == 'available': # T004
logger.info('RDS起動状態の取得処理が完了しました。')
return {'rds_status': rds_status}
else: # rdsステータスがavailable出ない場合はエラー通知、正常終了 # T005
logger.error('所定時間内にRDSが起動状態になりませんでした。\nrds_status: {rds_status}')
return {'rds_status': rds_status}
""" T006 """
logger.error('指定されたアクションは存在しません。')
raise
テストコード
Boto3によるAPI通信が走る部分をモックしています。そのため、実際にAWS環境に変更を与える心配はいりません。
test_lambda_function.py
boto3/rds/test_lambda_function.py
import pytest
import controle_rds
@pytest.fixture(scope='function')
def MockRDSClient(mocker):
return mocker.patch('controle_rds.RDSClient', spec=True)
@pytest.fixture
def t002_client(MockRDSClient):
client = MockRDSClient()
client.reboot_db_instance.return_value = {}
return client
@pytest.fixture
def t003_client(MockRDSClient):
client = MockRDSClient()
client.describe_db_instances.return_value = {
'DBInstances': [
{"DBInstanceStatus": "available"},
{"DBInstanceStatus": "failed"}
]
}
return client
@pytest.fixture
def t004_client(MockRDSClient):
client = MockRDSClient()
client.describe_db_instances.return_value = {
'DBInstances': [
{"DBInstanceStatus": "available"}
]
}
return client
@pytest.fixture
def t005_client(MockRDSClient):
client = MockRDSClient()
client.describe_db_instances.return_value = {
'DBInstances': [
{"DBInstanceStatus": "failed"}
]
}
return client
class TestLambdaHandler():
""" T001 """
@pytest.mark.t001
def test_t001(self, caplog):
print('=============T001 start=============')
params = {
'event': {
'action': 'reboot'
},
'context': None
}
with pytest.raises(Exception):
controle_rds.lambda_handler(**params)
assert '引数が不正です。' in caplog.text
print('=============T001 end=============')
""" T002 """
@pytest.mark.t002
def test_t002(self, t002_client, caplog):
print('=============T002 start=============')
mock_rds_client = t002_client
params = {
'event': {
'action': 'reboot',
'rds_id': 'test-rds-xx'
},
'context': None
}
controle_rds.lambda_handler(**params)
mock_rds_client.reboot_db_instance.assert_called_with('test-rds-xx')
assert mock_rds_client.reboot_db_instance.call_count == 1
assert mock_rds_client.describe_db_instances.call_count == 0
assert 'RDS再起動処理を開始します。' in caplog.text
assert 'RDS再起動処理が完了しました。' in caplog.text
print('=============T002 end=============')
""" T003 """
@pytest.mark.t003
def test_t003(self, t003_client, caplog):
print('=============T003 start=============')
mock_rds_client = t003_client
params = {
'event': {
'action': 'describe_status',
'rds_id': 'test-rds-xx'
},
'context': None
}
with pytest.raises(Exception):
controle_rds.lambda_handler(**params)
mock_rds_client.describe_db_instances.assert_called_with('test-rds-xx')
assert mock_rds_client.reboot_db_instance.call_count == 0
assert mock_rds_client.describe_db_instances.call_count == 1
assert 'RDS起動状態の取得処理を開始します。' in caplog.text
assert '取得したRDSの情報が不正です。'
print('=============T003 end=============')
""" T004 """
@pytest.mark.t004
def test_t004(self, t004_client, caplog):
print('=============T004 start=============')
mock_rds_client = t004_client
params = {
'event': {
'action': 'describe_status',
'rds_id': 'test-rds-xx'
},
'context': None
}
controle_rds.lambda_handler(**params)
mock_rds_client.describe_db_instances.assert_called_with('test-rds-xx')
assert mock_rds_client.reboot_db_instance.call_count == 0
assert mock_rds_client.describe_db_instances.call_count == 1
assert 'RDS起動状態の取得処理を開始します。' in caplog.text
assert 'RDS起動状態の取得処理が完了しました。' in caplog.text
print('=============T004 end=============')
""" T005 """
@pytest.mark.t005
def test_t005(self, t005_client, caplog):
print('=============T005 start=============')
mock_rds_client = t005_client
params = {
'event': {
'action': 'describe_status',
'rds_id': 'test-rds-xx'
},
'context': None
}
controle_rds.lambda_handler(**params)
mock_rds_client.describe_db_instances.assert_called_with('test-rds-xx')
assert mock_rds_client.reboot_db_instance.call_count == 0
assert mock_rds_client.describe_db_instances.call_count == 1
assert 'RDS起動状態の取得処理を開始します。' in caplog.text
assert '所定時間内にRDSが起動状態になりませんでした。' in caplog.text
print('=============T005 end=============')
""" T006 """
@pytest.mark.t006
def test_t006(self, MockRDSClient, caplog):
print('=============T006 start=============')
params = {
'event': {
'action': "describe",
'rds_id': 'test-rds-xx'
},
'context': None
}
with pytest.raises(Exception):
controle_rds.lambda_handler(**params)
assert '指定されたアクションは存在しません。' in caplog.text
print('=============T006 end=============')
pytest 設定ファイル
pytest.ini
boto3/pytest.ini
[pytest]
addopts = -v --cov --cov-branch --cov-report=term-missing
python_files = test_*
markers =
t001: t001
t002: t002
t003: t003
t004: t004
t005: t005
t006: t006
pytest-cov 設定ファイル
.coveragerc
boto3/.coveragerc
[report]
exclude_lines =
class RDSClient:
最後に
ドキュメンテーションを補完する意味でもテストを書いておくと良いと思いました。
Discussion