![[Pytest] 非同期処理のテストがたまに失敗する原因と解決策](https://images.ctfassets.net/ct0aopd36mqt/wp-thumbnail-a873a00568e5285b2051ff6c5aae2c3d/febef60aecfe576a837a7efcb96625eb/python_icatch.png)
[Pytest] 非同期処理のテストがたまに失敗する原因と解決策
2025.08.29こんにちは。サービス開発室の武田です。
Pythonで非同期処理のテストを書いていたら、ローカルでは成功するのにCIでときどき失敗する、というかローカルでもたまに失敗するという現象に遭遇しました。今回は、AWS Lambda関数のテストで実際に遭遇した、asyncio.gather()使用時にモックの順序がたまに入れ替わるという問題とその解決方法について共有します。
テストがたまに失敗する
AWS Configの有効化チェック機能をテストしていた際、次のようなエラーが発生しました。
# テストコード
results = await check_config_enabled(TEST_REGIONS, mock_session)
assert results == [True, False, True]
# エラー
E assert [True, True, False] == [True, False, True]
E At index 1 diff: True != False
同じテストを複数回実行すると、成功したり失敗したりする不安定な状態でした。
テスト対象となっていた問題のコードは、複数リージョンで並列にConfig有効化をチェックする非同期関数でした。
async def check_config_enabled(regions: list[str], session: Any) -> list[bool]:
"""複数リージョンでConfigが有効化されているか並列で確認する"""
credentials = session.get_credentials().get_frozen_credentials()
results = await asyncio.gather(
*(asyncio.to_thread(_check_config_enabled_in_region, region, credentials)
for region in regions)
)
return results
テストコードでは、モックを使って各リージョンの結果を制御しています。
@pytest.mark.asyncio
@patch("utils.config_checker._check_config_enabled_in_region")
async def test_config_enabled_in_multiple_regions(self, mock_check_region):
# テスト用リージョン
TEST_REGIONS = ["ap-northeast-1", "us-east-1", "eu-west-1"]
# side_effectでリストを使用（問題のある実装）
mock_check_region.side_effect = [True, False, True]
results = await check_config_enabled(TEST_REGIONS, mock_session)
# 期待する結果
assert results == [True, False, True]
原因はside_effectリストと並列実行の相性問題
先にasyncio.gather()の動作を確認すると、次の仕様となっています。
- 結果の順序は入力順序と一致する
- ただし、実行順序は保証されない
つまり、3つのリージョンを ["ap-northeast-1", "us-east-1", "eu-west-1"] の順で渡せば、結果も必ずこの順序で返されます。
問題はmock.side_effectの動作にあります。
side_effectにリストを設定すると、呼び出し順でリストの値が消費される- 並列実行では、どのリージョンが最初に実行されるかは不定
- 結果として、リージョンと戻り値の対応が予測できない
つまり次のようなことが起こりえます。
- 実行1回目: us-east-1 → ap-northeast-1 → eu-west-1
- 結果: [False, True, True]
- 実行2回目: ap-northeast-1 → eu-west-1 → us-east-1
- 結果: [True, True, False]
呼び出し順への依存をなくすように修正
この問題の解決策は、呼び出し順への依存をなくすことです。side_effectにはリストではなく、引数（リージョン名）をもとに固定の値を返すようにしましょう。
@pytest.mark.asyncio
@patch("utils.config_checker._check_config_enabled_in_region")
async def test_config_enabled_in_multiple_regions(self, mock_check_region):
TEST_REGIONS = ["ap-northeast-1", "us-east-1", "eu-west-1"]
# リージョン名に基づいて固定の値を返す
def region_based_result(region, credentials):
return {
"ap-northeast-1": True,
"us-east-1": False,
"eu-west-1": True,
}[region]
mock_check_region.side_effect = region_based_result
results = await check_config_enabled(TEST_REGIONS, mock_session)
# 常に期待通りの結果が得られる
assert results == [True, False, True]
まとめ
非同期処理のテストでは、次の点に注意が必要でした。
asyncio.gather()は結果の順序を保証するが、実行順序は保証しないmock.side_effectのリストは実行順序に依存するため、並列実行では予測不能- 入力パラメーターに基づくモックを作成し、安定したテストを実現
並列処理のテストは複雑ですが、適切なモック戦略を使えば、安定して高速なテストを実現できます。特にマルチリージョン対応のAWSアプリケーションなど、並列処理が重要な場面では、このような考慮が必要不可欠です。
