【pytest】motoのインポート順序がClientErrorを引き起こす？

こんにちは。ディベロップメントサービス１課の山本です。

たまにYoutubeでライブカメラを見るのが好きで、歌舞伎町見てると楽しいです。
阪神優勝の日も、道頓堀川のライブカメラを眺めてました。

今回は、AWSサービスをモックするのに便利なmotoを利用する際の注意点を説明します。 OKだったテスト結果が一瞬でNG(ClientError)となるサンプルをご紹介します。

はじめに
環境
NGサンプル1(インポートの順番)
NGサンプル2(motoを利用していないテストコードがある場合)
まとめ
最後に

はじめに

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リソースの箇所までテストしましょう！
本ブログがどなかたのお役に立てれば幸いです。

はじめに

環境

利用ライブラリ

テスト対象コード

ディレクトリ構成

NGサンプル1(インポートの順番)

テストコード

テスト結果

モックの動作確認

エラー原因

対処法

NGサンプル2(motoを利用していないテストコードがある場合)

ディレクトリ構成

追加するテスト対象コード

追加するテストコード

テスト結果

エラー原因

対処法

余談

まとめ

最後に

Amazon EventBridgeで実現するマルチアカウントイベントの集約管理

ACM for Nitro Enclaves を利用してNginxへのHTTPS通信を実装してみた

ISUCON過去問のボトルネックをNew Relic APMで確認する

AWSから抜け出す方法も考えておく【ガバメントクラウド】

AWS BuilderCardsを遊んでみた