💻

【pytest】AWS Lambda のテストコードを書くよ

2025/03/07に公開

概要

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