こんにちは。 ディベロップメントサービス１課の山本です。
たまにYoutubeでライブカメラを見るのが好きで、歌舞伎町見てると楽しいです。
阪神優勝の日も、道頓堀川のライブカメラを眺めてました。
今回は、AWSサービスをモックするのに便利なmotoを利用する際の注意点を説明します。 OKだったテスト結果が一瞬でNG(ClientError)となるサンプルをご紹介します。
はじめに
motoとはAWSのサービスをモックするモジュールとなります。
使い方に関しては、過去に当社が記載したブログがありますのでこちらを参考ください。
blog.serverworks.co.jp blog.serverworks.co.jp
今回はNGサンプルを2つ紹介して、 原因と対処法を説明します。
環境
利用ライブラリ
- boto3
- moto
- pytest
- pytest-env (環境変数のモックのため導入)
- pytest-mock (サンプル2で使用)
テスト対象コード
今回は極限まで簡単にしたコードをテストします。
DynamoDBのテーブル一覧を返却するLambda関数を想定したコードとなります。
sample.py
import boto3 dynamodb = boto3.client('dynamodb') def handler(event, context): response = dynamodb.list_tables() return response["TableNames"]
ディレクトリ構成
srcディレクトリ配下にテスト対象コード、
testsディレクトリ配下にテストコードを格納します。
ディレクトリ構成
/ ├ pyproject.toml ├ src └ sample.py └ tests ├ __init__.py └ test_sample.py
テスト時のコンフィグ設定をpyproject.tomlに記載します。
テストコードのパスと、各種AWSのダミー用の認証設定を格納します。
pyproject.toml
[tool.pytest.ini_options] testpaths = ["tests"] env = [ "AWS_ACCESS_KEY_ID=env", "AWS_SECRET_ACCESS_KEY=env", "AWS_DEFAULT_REGION=ap-northeast-1" ]
NGサンプル1(インポートの順番)
テストコード
テスト用のテーブル「sample1_table」をモックで作成して
テスト対象のコードで同じ値を取得できるか確認するテストコードです。
test_sample.py
import boto3 from src.sample import handler from moto import mock_dynamodb def create_table(client: boto3.client, table_name: str, attribute_name: str) -> None: client.create_table( TableName=table_name, KeySchema=[{"AttributeName": attribute_name, "KeyType": "HASH"}], AttributeDefinitions=[{"AttributeName": attribute_name, "AttributeType": "S"}], BillingMode="PAY_PER_REQUEST", ) @mock_dynamodb def test_handler(): test_event = {} lambda_context = {} # テスト用のテーブルを作成 dynamodb = boto3.client("dynamodb") create_table(dynamodb, "sample1_table", "sample1_ID") # テスト実行 result = handler(test_event, lambda_context) assert result == ["sample1_table"]
ここで問題です。
果たしてこのテスト、なぜNGになるでしょうか。
テスト結果
答えはClientErrorが発生してテストNGとなります。
どうもモックがうまく機能していない模様。
E botocore.exceptions.ClientError: An error occurred (UnrecognizedClientException) when calling the ListTables operation: The security token included in the request is invalid. .venv/lib/python3.11/site-packages/botocore/client.py:983: ClientError =========================== short test summary info ============================ FAILED tests/test_sample.py::test_handler - botocore.exceptions.ClientError: ... ============================== 1 failed in 0.45s =============================== Finished running tests!
モックの動作確認
boto3.clientのエンドポイントを出力して、
モックされているかを確認します。
sample.py
import boto3 dynamodb = boto3.client("dynamodb") def handler(event, context): print(dynamodb.meta.endpoint_url) #エンドポイントのURLを出力 response = dynamodb.list_tables() return response["TableNames"]
出力結果
----------------------------- Captured stdout call ----------------------------- https://dynamodb.ap-northeast-1.amazonaws.com =========================== short test summary info ============================ FAILED tests/test_sample.py::test_handler - botocore.exceptions.ClientError: ... ============================== 1 failed in 0.47s =============================== Finished running tests!
やはり本物のDynamoDBにアクセスしている様です。
そのため、デタラメな認証情報ではClientErrorとなって失敗しています。
エラー原因
ここでタイトルの内容に戻ってくるのですが、
エラーの原因はhandlerとmock_dynamodbのインポートの順番にあります。
motoを利用する際、必ず下記の順番を守る必要があります。
- motoのインポート
- モック対象のclient/resourceの構築
今回はそれが守られていませんでした。
import boto3 from src.sample import handler # ハンドラーインポート時にclient/resourceが構築 from moto import mock_dynamodb # motoのインポート
簡単な理由は下記の通りです。
- moto：インポート段階でHTTPリクエストを差し替えるイベントハンドラーを設定する。
- client/resource：構築時のみ、設定済みのイベントハンドラーを参照する。
そのため、client/resource構築後にいくらモックしても無意味だったということです。
詳細は下記のmotoドキュメントに記載されてますので、参考ください。
Getting Started with Moto — Moto 4.2.9.dev documentation
対処法
handlerより先に、motoのインポートをしましょう。
test_sample.py
import boto3 from moto import mock_dynamodb # 先にmotoのインポートを実施 from src.sample import handler def create_table(client: boto3.client, table_name: str, attribute_name: str) -> None: client.create_table( TableName=table_name, KeySchema=[{"AttributeName": attribute_name, "KeyType": "HASH"}], AttributeDefinitions=[{"AttributeName": attribute_name, "AttributeType": "S"}], BillingMode="PAY_PER_REQUEST", ) @mock_dynamodb def test_handler(): test_event = {} lambda_context = {} # テスト用のテーブルを作成 dynamodb = boto3.client("dynamodb") create_table(dynamodb, "sample1_table", "sample1_ID") # テスト実行 result = handler(test_event, lambda_context) assert result == ["sample1_table"]
テスト結果：OK
================= test session starts ================ platform darwin -- Python 3.11.4, pytest-7.4.3, pluggy-1.3.0 rootdir: /Users/yamamotomasatomo/Documents/02_code/blog_mock configfile: pyproject.toml plugins: env-1.1.1 collected 1 item tests/test_sample.py . [100%] ================= 1 passed in 0.42s =================
NGサンプル2(motoを利用していないテストコードがある場合)
次にテスト対象のコード(contents.py)を1つ追加します。
ディレクトリ構成
/ ├ pyproject.toml ├ src ├ contents.py # 追加 └ sample.py └ tests ├ __init__.py ├ test_contents.py # 追加 └ test_sample.py
追加するテスト対象コード
今度はS3バケット名をリストで返却するコードです。
contents.py
import boto3 s3 = boto3.client("s3") def handler(event, context): response = get_bucket_list() bucket_names = [bucket["Name"] for bucket in response["Buckets"]] return bucket_names def get_bucket_list(): response = s3.list_buckets() return response
追加するテストコード
こちらのテストコードではmotoは利用せずに、mocker.patchを利用して戻り値をモックします。
test_contents.py
from src.contents import handler def test_handler(mocker): test_event = {} lambda_context = {} mock_s3_data = { "Buckets": [ {"Name": "sample1_bucket", "CreationDate": "2021-01-01"}, ], "Owner": {"DisplayName": "sample_owner", "ID": "sample_id"}, } mocker.patch("src.contents.get_bucket_list", return_value=mock_s3_data) # テスト実行 result = handler(test_event, lambda_context) assert result == ["sample1_bucket"]
ここで問題です。
果たして、どのテストがエラーとなるでしょうか。
テスト結果
答えは、先ほどの修正でPassする様になった
test_sample.pyが同じClientErrorでNGになります。
E botocore.exceptions.ClientError: An error occurred (UnrecognizedClientException) when calling the ListTables operation: The security token included in the request is invalid. .venv/lib/python3.11/site-packages/botocore/client.py:983: ClientError =========================== short test summary info ============================ FAILED tests/test_sample.py::test_handler - botocore.exceptions.ClientError: ... ========================= 1 failed, 1 passed in 0.46s ========================== Finished running tests!
エラー原因
エラー原因は、pytestが実行される順番にあります。
ファイル名順に実行されるため、追加したtest_contents.pyが先に実行されます。
- test_contents.pyの実行
- contents.pyのインポート ←ここでclient/resourceを構築
- test_sample.pyの実行 ←ここでmotoでモック
- sample.pyのインポート ←2.で構築したclient/resourceを使い回し
contents.pyにてグローバルでclient/resourceを構築しているので、
test_sample.pyでは、モックされてないリソースを使い回すためエラーが発生します。
対処法
テスト関数内でローカルインポートしましょう。
実は先ほど紹介した、motoドキュメント内にもこの解決法が提案されてます。
Getting Started with Moto — Moto 4.2.9.dev documentation
import boto3 from moto import mock_dynamodb def create_table(client: boto3.client, table_name: str, attribute_name: str) -> None: client.create_table( TableName=table_name, KeySchema=[{"AttributeName": attribute_name, "KeyType": "HASH"}], AttributeDefinitions=[{"AttributeName": attribute_name, "AttributeType": "S"}], BillingMode="PAY_PER_REQUEST", ) @mock_dynamodb def test_handler(): from src.sample import handler # ローカルインポートを実施する test_event = {} lambda_context = {} # テスト用のテーブルを作成 dynamodb = boto3.client("dynamodb") create_table(dynamodb, "sample1_table", "sample1_ID") # テスト実行 result = handler(test_event, lambda_context) assert result == ["sample1_table"]
余談
実はファイル名を変えて、実行順を変えるだけでもテストはPassします。
まとめ
motoを利用して、AWSリソースをモックする際の掟
- motoのインポート -> モック対象のclient/resourceの構築 の順番を守る
- motoを利用してモックする際は、テスト対象のコードをテスト関数内でローカルインポートする
最後に
皆さんもテストの際は是非、motoを利用してAWSリソースの箇所までテストしましょう！
本ブログがどなかたのお役に立てれば幸いです。
