コンテンツへスキップ

テスト

Starletteのおかげで、FastAPIアプリケーションのテストは簡単で楽しいです。

これはHTTPXに基づいており、HTTPXはRequestsに基づいて設計されているため、非常に馴染みやすく直感的に使用できます。

これにより、FastAPIpytestを直接使用できます。

TestClientの使用

情報

TestClientを使用するには、まずhttpxをインストールします。

仮想環境を作成し、有効にしてからインストールします(例:)

$ pip install httpx

TestClientをインポートします。

FastAPIアプリケーションを渡してTestClientを作成します。

名前がtest_で始まる関数を作成します(これは標準的なpytestの規約です)。

httpxと同じようにTestClientオブジェクトを使用します。

必要なチェックを行う標準的なPython式を使用して、シンプルなassert文を書きます(これも標準的なpytestです)。

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()


@app.get("/")
async def read_main():
    return {"msg": "Hello World"}


client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

ヒント

テスト関数は通常のdefであり、async defではないことに注意してください。

また、クライアントへの呼び出しも通常の呼び出しであり、awaitを使用していません。

これにより、pytestを複雑にすることなく直接使用できます。

「技術的な詳細」

from starlette.testclient import TestClientを使用することもできます。

FastAPIは、開発者の便宜のためにfastapi.testclientとしてstarlette.testclientと同じものを提供していますが、これはStarletteから直接提供されています。

ヒント

FastAPIアプリケーションへのリクエストの送信以外に、テスト内で非同期関数(例:非同期データベース関数)を呼び出す場合は、上級者向けチュートリアルの非同期テストを参照してください。

テストの分離

実際のアプリケーションでは、テストは別のファイルに配置するでしょう。

そして、あなたのFastAPIアプリケーションは、複数のファイル/モジュールなどで構成されている場合もあります。

FastAPIアプリファイル

より大規模なアプリケーションで説明されているようなファイル構造があるとしましょう。

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

ファイルmain.pyには、あなたのFastAPIアプリがあります。

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_main():
    return {"msg": "Hello World"}

テストファイル

そして、テストを含むファイルtest_main.pyを持つことができます。これは、同じPythonパッケージ(__init__.pyファイルのある同じディレクトリ)に存在することができます。

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

このファイルは同じパッケージ内にあるため、mainモジュール(main.py)からオブジェクトappをインポートするために、相対インポートを使用できます。

from fastapi.testclient import TestClient

from .main import app

client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

…そして、これまでと同じようにテストのコードを書きます。

テスト:拡張例

では、この例を拡張し、詳細を追加して、さまざまな部分をどのようにテストするかを見ていきましょう。

拡張されたFastAPIアプリファイル

これまでと同じファイル構造で続けます。

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

あなたのFastAPIアプリを含むファイルmain.pyに、他のパス操作があるとしましょう。

エラーを返す可能性のあるGET操作があります。

複数のエラーを返す可能性のあるPOST操作があります。

両方のパス操作には、X-Tokenヘッダーが必要です。

from typing import Annotated

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: str | None = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item
from typing import Annotated, Union

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: Union[str, None] = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item
from typing import Union

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
from typing_extensions import Annotated

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: Union[str, None] = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item

ヒント

可能であれば、Annotatedバージョンを使用することをお勧めします。

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: str | None = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item

ヒント

可能であれば、Annotatedバージョンを使用することをお勧めします。

from typing import Union

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: Union[str, None] = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item

拡張されたテストファイル

その後、拡張されたテストを使用してtest_main.pyを更新できます。

from fastapi.testclient import TestClient

from .main import app

client = TestClient(app)


def test_read_item():
    response = client.get("/items/foo", headers={"X-Token": "coneofsilence"})
    assert response.status_code == 200
    assert response.json() == {
        "id": "foo",
        "title": "Foo",
        "description": "There goes my hero",
    }


def test_read_item_bad_token():
    response = client.get("/items/foo", headers={"X-Token": "hailhydra"})
    assert response.status_code == 400
    assert response.json() == {"detail": "Invalid X-Token header"}


def test_read_nonexistent_item():
    response = client.get("/items/baz", headers={"X-Token": "coneofsilence"})
    assert response.status_code == 404
    assert response.json() == {"detail": "Item not found"}


def test_create_item():
    response = client.post(
        "/items/",
        headers={"X-Token": "coneofsilence"},
        json={"id": "foobar", "title": "Foo Bar", "description": "The Foo Barters"},
    )
    assert response.status_code == 200
    assert response.json() == {
        "id": "foobar",
        "title": "Foo Bar",
        "description": "The Foo Barters",
    }


def test_create_item_bad_token():
    response = client.post(
        "/items/",
        headers={"X-Token": "hailhydra"},
        json={"id": "bazz", "title": "Bazz", "description": "Drop the bazz"},
    )
    assert response.status_code == 400
    assert response.json() == {"detail": "Invalid X-Token header"}


def test_create_existing_item():
    response = client.post(
        "/items/",
        headers={"X-Token": "coneofsilence"},
        json={
            "id": "foo",
            "title": "The Foo ID Stealers",
            "description": "There goes my stealer",
        },
    )
    assert response.status_code == 409
    assert response.json() == {"detail": "Item already exists"}

クライアントがリクエストで情報を渡す必要があり、方法がわからない場合は、httpxで行う方法、またはrequestsで行う方法を(Googleで)検索してください。HTTPXの設計はRequestsの設計に基づいているためです。

そして、テストでも同じことを行います。

例:

  • パスまたはクエリパラメータを渡すには、URL自体に追加します。
  • JSONボディを渡すには、Pythonオブジェクト(例:dict)をjsonパラメータに渡します。
  • JSONの代わりにフォームデータを送信する必要がある場合は、代わりにdataパラメータを使用します。
  • ヘッダーを渡すには、headersパラメータにdictを使用します。
  • クッキーには、cookiesパラメータにdictを使用します。

バックエンドにデータを渡す方法(httpxまたはTestClientを使用)の詳細については、HTTPXドキュメントを参照してください。

情報

TestClientは、JSONに変換できるデータを受け取りますが、Pydanticモデルは受け取りません。

テストにPydanticモデルがあり、テスト中にそのデータをアプリケーションに送信したい場合は、JSON互換エンコーダーで説明されているjsonable_encoderを使用できます。

実行

その後、pytestをインストールするだけです。

仮想環境を作成し、有効にしてからインストールします(例:)

$ pip install pytest

---> 100%

ファイルとテストを自動的に検出し、実行し、結果を報告します。

次のコマンドでテストを実行します。

$ pytest

================ test session starts ================
platform linux -- Python 3.6.9, pytest-5.3.5, py-1.8.1, pluggy-0.13.1
rootdir: /home/user/code/superawesome-cli/app
plugins: forked-1.1.3, xdist-1.31.0, cov-2.8.1
collected 6 items

---> 100%

test_main.py <span style="color: green; white-space: pre;">......                            [100%]</span>

<span style="color: green;">================= 1 passed in 0.03s =================</span>