非同期テスト¶

これまでに、提供されているTestClientを使用してFastAPIアプリケーションをテストする方法を見てきました。これまでは、async関数を使用せずに、同期テストを作成する方法のみを見てきました。

テストで非同期関数を使用できると便利な場合があります。たとえば、データベースを非同期的にクエリしている場合などです。FastAPIアプリケーションにリクエストを送信し、非同期データベースライブラリを使用しながら、バックエンドが正しいデータをデータベースに正常に書き込んだことを確認したいとします。

それがどのように機能するかを見てみましょう。

pytest.mark.anyio¶

テストで非同期関数を呼び出したい場合、テスト関数は非同期でなければなりません。AnyIOはこれに便利なプラグインを提供しており、一部のテスト関数を非同期で呼び出すように指定できます。

HTTPX¶

たとえFastAPIアプリケーションがasync def関数ではなく通常のdef関数を使用している場合でも、その下にはasyncアプリケーションがあります。

TestClientは、標準のpytestを使用して、通常のdefテスト関数内で非同期のFastAPIアプリケーションを呼び出すために内部でいくつかの魔法を行います。しかし、非同期関数内で使用する場合、その魔法はもう機能しません。テストを非同期で実行することにより、テスト関数内でTestClientを使用できなくなります。

TestClientはHTTPXに基づいており、幸いなことに、APIをテストするために直接使用できます。

例¶

簡単な例として、大規模なアプリケーションおよびテストで説明されているようなファイル構造を考えてみましょう。

.
├── app
│   ├── __init__.py
│   ├── main.py
│   └── test_main.py

ファイルmain.pyは次のようになります。

Python 3.8+

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Tomato"}

ファイルtest_main.pyにはmain.pyのテストが含まれ、現在は次のようになります。

Python 3.8+

import pytest
from httpx import ASGITransport, AsyncClient

from .main import app


@pytest.mark.anyio
async def test_root():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Tomato"}

実行¶

テストは通常通り実行できます。

$ pytest

---> 100%

詳細¶

マーカー@pytest.mark.anyioは、このテスト関数が非同期で呼び出されるべきであることをpytestに伝えます。

Python 3.8+

import pytest
from httpx import ASGITransport, AsyncClient

from .main import app


@pytest.mark.anyio
async def test_root():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Tomato"}

ヒント

TestClientを使用していた以前とは異なり、テスト関数は単なるdefではなくasync defになっていることに注意してください。

次に、アプリでAsyncClientを作成し、awaitを使用して非同期リクエストを送信できます。

Python 3.8+

import pytest
from httpx import ASGITransport, AsyncClient

from .main import app


@pytest.mark.anyio
async def test_root():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Tomato"}

これは次のものと同等です。

response = client.get('/')

...TestClientを使用してリクエストを行っていました。

ヒント

新しいAsyncClientでasync/awaitを使用していることに注意してください。リクエストは非同期です。

Warning

アプリケーションがライフスパンイベントに依存している場合、AsyncClientはこれらのイベントをトリガーしません。それらがトリガーされることを確認するには、florimondmanca/asgi-lifespanのLifespanManagerを使用してください。

その他の非同期関数呼び出し¶

テスト関数が非同期になったため、コード内の他の場所で呼び出すのとまったく同じように、FastAPIアプリケーションにリクエストを送信する以外にも、他のasync関数をテスト内で呼び出す（そしてawaitする）ことができます。

ヒント

テストで非同期関数呼び出しを統合する際（例：MongoDBのMotorClientを使用する場合）にRuntimeError: Task attached to a different loopが発生した場合は、イベントループが必要なオブジェクトは非同期関数内でのみインスタンス化するようにしてください（例：@app.on_event("startup")コールバック）。