[Python] pytest-asyncio で非同期処理をテスト、非同期フィクスチャを実装

pytest で非同期処理をテストしたい

pytest で asyncio を使った非同期処理をテストしたい場合、単純に async キーワードを付与してテストメソッドを作ってもうまくいきません。

仮に async def test_asyncio():　みたいなテストを作ってテストを実行すると以下のように async def のテストメソッドはネイティブでサポートされていないためスキップされました。

test_sample.py::test_asyncio SKIPPED (async def function and no async plugin installed (see war...)
..
PytestUnhandledCoroutineWarning: async def functions are not natively supported and have been skipped.
  You need to install a suitable plugin for your async framework, for example:
    - anyio
    - pytest-asyncio
    - pytest-tornasync
    - pytest-trio
    - pytest-twisted
    warnings.warn(PytestUnhandledCoroutineWarning(msg.format(nodeid)))

asyncio のテストメソッドをテストしたい場合には、上の警告の通り何かしらのプラグインが必要です。今回は pytest-asyncio というプラグインを導入して対応します。

pytest-asyncio をインストール

• pytest-dev/pytest-asyncio: Pytest support for asyncio.

$ pip install pytest-asyncio

pytest-asyncio の使い方

pytest-asyncio をインストールすると、非同期処理をテストするためのマーカーとフィクスチャが提供されます。

@pytest.mark.asyncio で非同期テストを実装

@pytest.mark.asyncio をマーカーとして付与すれば async を付けたテストを実装できます。

import pytest
import asyncio

# 1秒後に加算結果を返す
async def add_async(x, y, event_loop=None):
    await asyncio.sleep(1)
    return x + y


@pytest.mark.asyncio
async def test_add_1():
    # 普通に await する
    res = await add_async(1, 2)
    assert res == 3

マーカーを付与するだけで普段通りに async, await してテストを実装することができます。

@pytest.mark.asyncio マーカーは、event_loop フィクスチャで提供されるイベントループを使って非同期タスクを実行します。

もちろん影響を受けるのはマーカーが付いてるテストのみで通常のテストには影響しません。

event_loop フィクスチャを使って非同期テストを実装

pytest-asyncio が提供する event_loop フィクスチャでイベントループを使うことができます。event_loop を使って event_loop.run_until_complete で処理を待機することも可能です。

def test_add_2(event_loop):
    res = event_loop.run_until_complete(add_async(1, 2))
    assert res == 3

フィクスチャを使わなくても以下のようにイベントループを取得して実行もできます。

def test_add_3():
    event_loop = asyncio.get_event_loop()
    res = event_loop.run_until_complete(add_async(1, 2))
    assert res == 3

こうすれば pytest-asyncio を使わなくても事項できますが、違いはよくわかりません！

非同期フィクスチャを使う

フィクスチャを使った初期化処理で非同期処理を使うには以下のようにします。

import pytest
import asyncio

@pytest.fixture
async def async_fixture():
    await asyncio.sleep(0.01)
    return 1

def test_async(async_fixture):
    assert async_fixture == 1

非同期処理を使うフィクスチャの定義には特別なことをしなくても普通に async を付けたメソッドとして実装すればよいです。あとは pytest-async がいい感じに処理してくれます。上のコードだと test_async に渡されるは '1' になります。

以上。