非同期テスト¶

提供されているTestClientを使ってFastAPIアプリケーションをテストする方法はすでに見てきました。これまでは、async関数を使わずに、同期テストを書く方法しか見ていませんでした。

テストで非同期関数を使えることは、例えばデータベースを非同期的にクエリする場合に役立つかもしれません。FastAPIアプリケーションにリクエストを送信し、非同期データベースライブラリを使いながら、バックエンドが正しいデータをデータベースに正常に書き込んだことを検証したい場合を想像してみてください。

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

pytest.mark.anyio¶

テストで非同期関数を呼び出すには、テスト関数を非同期にする必要があります。AnyIOはこれのための素晴らしいプラグインを提供しており、一部のテスト関数を非同期に呼び出すことを指定できます。

HTTPX¶

FastAPIアプリケーションがasync defではなく通常のdef関数を使用している場合でも、その内部ではasyncアプリケーションです。

TestClientは、通常のdefテスト関数内で非同期のFastAPIアプリケーションを呼び出すために、標準のpytestを使用して内部でいくつかの魔法を実行します。しかし、非同期関数内で使用すると、その魔法はもはや機能しません。テストを非同期で実行すると、テスト関数内で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を使用していることに注意してください。リクエストは非同期です。

警告

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

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

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

ヒント

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