Zenn
🧪

Python: 外部APIを呼ぶAWS Lambda関数をローカルでテストする

に公開
Python 3
tech

はじめに

外部APIを呼び出すAWS Lambda関数のテストについて。
APIサーバーの状態やネットワークに依存せず、ローカル環境で高速かつ安定したテストを実行したい。

技術スタック

  • pyenv: Python自体のバージョン管理
  • venv: プロジェクトのライブラリ依存関係管理
  • unittest.mock: Python標準のモックライブラリ

Step 1: クリーンな開発環境の構築 (pyenv + venv)

まず、pyenvでプロジェクトで使うPythonのバージョンを固定し、venvでライブラリ専用の仮想環境を作成する。

# 1. このプロジェクトで使うPythonのバージョンを指定
# (事前に pyenv install 3.12.4 などでインストールしておく)
pyenv local 3.12.4

# 2. 指定したPythonバージョンで仮想環境 `venv` を作成
python -m venv venv

# 3. 仮想環境を有効化
source venv/bin/activate

# (venv)とプロンプトに表示されれば成功
# 4. 必要なライブラリをインストール
pip install requests

Step 2: Lambda関数とテストコードの作成

API Gatewayからのリクエストを想定し、外部の天気予報APIを叩いて結果を返すLambda関数を実装する。

テスト対象のLambda関数 (lambda_function.py)

APIエンドポイントは環境変数から、都市名はクエリパラメータで受け取る仕様。

lambda_function.py
import os
import json
import requests

def lambda_handler(event, context):
    """Lambdaのエントリポイント"""
    # 環境変数からAPIエンドポイントを取得
    api_endpoint = os.environ.get('API_ENDPOINT')
    if not api_endpoint:
        return {
            'statusCode': 500,
            'body': json.dumps({'message': 'API_ENDPOINT environment variable is not set.'})
        }

    # eventオブジェクトからクエリパラメータを取得
    city = event.get('queryStringParameters', {}).get('city')
    if not city:
        return {
            'statusCode': 400,
            'body': json.dumps({'message': 'Query parameter "city" is required.'})
        }

    # 外部APIを呼び出し
    try:
        response = requests.get(f"{api_endpoint}?city={city}")
        response.raise_for_status()
        weather_data = response.json()
    except requests.exceptions.RequestException as e:
        print(f"API request failed: {e}")
        return {
            'statusCode': 502,
            'body': json.dumps({'message': 'Failed to retrieve data from the external API.'})
        }

    # 結果を返す
    return {
        'statusCode': 200,
        'body': json.dumps(weather_data)
    }

スタブを使ったテストコード (test_lambda_function.py)

unittest.mock.patchを使い、Lambda関数の依存関係であるrequests.getとos.environをテスト用のモックに書き換える。

test_lambda_function.py
import unittest
import json
import requests
from unittest.mock import patch, Mock

from lambda_function import lambda_handler

class TestLambdaHandler(unittest.TestCase):

    @patch('lambda_function.os.environ', {'API_ENDPOINT': 'https://api.stub.com'})
    @patch('lambda_function.requests.get')
    def test_lambda_handler_success(self, mock_requests_get):
        """正常系: API呼び出しが成功し、200 OKが返るケース"""
        # 準備: requests.getのモックを設定
        mock_response = Mock()
        mock_response.status_code = 200
        mock_response.json.return_value = {'city': 'Kyoto', 'weather': 'Rainy'}
        mock_requests_get.return_value = mock_response
        test_event = {'queryStringParameters': {'city': 'Kyoto'}}

        # 実行
        response = lambda_handler(test_event, None)

        # 検証
        self.assertEqual(response['statusCode'], 200)
        self.assertEqual(json.loads(response['body']), {'city': 'Kyoto', 'weather': 'Rainy'})
        mock_requests_get.assert_called_once_with('https://api.stub.com?city=Kyoto')

    @patch('lambda_function.os.environ', {'API_ENDPOINT': 'https://api.stub.com'})
    @patch('lambda_function.requests.get')
    def test_lambda_handler_api_failure(self, mock_requests_get):
        """異常系: 外部APIの呼び出しに失敗し、502が返るケース"""
        mock_requests_get.side_effect = requests.exceptions.RequestException("Connection Error")
        test_event = {'queryStringParameters': {'city': 'Osaka'}}

        response = lambda_handler(test_event, None)

        self.assertEqual(response['statusCode'], 502)

    @patch('lambda_function.os.environ', {'API_ENDPOINT': 'https://api.stub.com'})
    def test_lambda_handler_no_city(self):
        """異常系: クエリパラメータにcityがなく、400が返るケース"""
        test_event = {'queryStringParameters': {}}
        response = lambda_handler(test_event, None)
        self.assertEqual(response['statusCode'], 400)
        body = json.loads(response['body'])
        self.assertEqual(body['message'], 'Query parameter "city" is required.')

if __name__ == '__main__':
    unittest.main()

テストコードのポイント

  • @patch('lambda_function.os.environ', ...): os.environを辞書でモックし、環境変数への依存をなくす(すべてのテストケースで必要)。
  • @patch('lambda_function.requests.get'): requests.getをMockオブジェクトに置き換え、実際のHTTPリクエストが発生しないようにする。
  • mock_requests_get.return_value: 成功時のレスポンスを模倣する。
  • mock_requests_get.side_effect: 失敗時に送出される例外をシミュレートする。
  • json.loads(response['body']): レスポンスボディのJSON文字列をパースしてPythonの辞書に戻してから検証する。

Step 3: テストの実行

準備が整ったので、仮想環境が有効化されたターミナルでテストを実行する。

$ python -m unittest test_lambda_function.py
...
----------------------------------------------------------------------
Ran 3 tests in 0.001s

OK

すべてのテストが成功すれば、ローカルでの検証は完了。

まとめ

外部APIを呼び出すLambda関数をテストするにあたって、
重要なポイントは以下の3つ

  1. 環境の分離: pyenvとvenvを使い、Pythonバージョンとライブラリ依存関係をプロジェクトごとに完全に分離する。
  2. 依存のスタブ化: unittest.mock.patchを使い、requestsのような外部I/Oやos.environのような環境要因への依存をテストから排除する。
  3. 構造での検証: JSONレスポンスは、文字列としてではなく、パースして辞書オブジェクトの構造として検証する。

Discussion