非同期テスト¶
🌐 AI と人間による翻訳
この翻訳は、人間のガイドに基づいて AI によって作成されました。🤝
原文の意図を取り違えていたり、不自然な表現になっている可能性があります。🤖
AI LLM をより適切に誘導するのを手伝う ことで、この翻訳を改善できます。
これまでに、提供されている TestClient を使って FastAPI アプリケーションをテストする方法を見てきました。ここまでは、async 関数を使わない同期テストのみでした。
テストで非同期関数を使えると、たとえばデータベースへ非同期にクエリする場合などに便利です。非同期データベースライブラリを使いながら、FastAPI アプリにリクエストを送り、その後バックエンドが正しいデータをデータベースに書き込めたかを検証したい、といったケースを想像してください。
その方法を見ていきます。
pytest.mark.anyio¶
テスト内で非同期関数を呼び出したい場合、テスト関数自体も非同期である必要があります。AnyIO はこれを実現するための便利なプラグインを提供しており、特定のテスト関数を非同期で呼び出すことを指定できます。
HTTPX¶
FastAPI アプリケーションが通常の def 関数を使っていても、その内側は依然として async アプリケーションです。
TestClient は、標準の pytest を使って通常の def のテスト関数から非同期の FastAPI アプリを呼び出すための「おまじない」を内部で行います。しかし、その「おまじない」はテスト関数自体が非同期の場合には機能しません。テストを非同期で実行すると、テスト関数内で TestClient は使えなくなります。
TestClient は HTTPX を基に作られており、幸いなことに API のテストには HTTPX を直接利用できます。
例¶
簡単な例として、大きなアプリケーション と テスト で説明したものに似たファイル構成を考えます:
.
├── 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 を使っている点に注意してください。リクエストは非同期です。
注意
アプリケーションが lifespan イベントに依存している場合、AsyncClient はそれらのイベントをトリガーしません。確実にトリガーするには、florimondmanca/asgi-lifespan の LifespanManager を使用してください。
その他の非同期関数呼び出し¶
テスト関数が非同期になったので、FastAPI アプリへのリクエスト送信以外の async 関数も、コードの他の場所と同様に呼び出して（await して）使えます。
豆知識
テストに非同期関数呼び出しを統合した際に（例: MongoDB の MotorClient 使用時）、RuntimeError: Task attached to a different loop に遭遇した場合は、イベントループを必要とするオブジェクトは非同期関数内でのみインスタンス化するようにしてください。例えば @app.on_event("startup") コールバック内で行います。