非同期テスト¶

既に提供されている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ファイルには以下が含まれます。

from fastapi import FastAPI

app = FastAPI()


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

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

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にこのテスト関数を非同期的に呼び出す必要があることを伝えます。

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を使用して非同期リクエストを送信できます。

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`を使用していることに注意してください。リクエストは非同期です。

警告

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

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

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

ヒント

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