FastAPIアプリケーションで行う様々なテストパターンのまとめ

kobayashi.m

2024.05.05

はじめに

データアナリティクス事業本部のkobayashiです。

FastAPIで作成したエンドポイントに対するPytestでテストコードはリクエストの送り方で書き方が変わりますのでそのパターンをまとめたいと思います。

FastAPI TestClientを使ってテストコードを作成する

FastAPIではテスト用のクラスであるTestClientが用意されていてこれを使用することでrequestsパッケージなどを使用してリクエストを送ってテスト作成するのではなく直接FastAPIのメソッドを実行してFastAPIのアプリケーションをテストできます。

環境

  • Python: 0.111.0

TestClientのメソッド

TestClientにはさまざまなメソッドが用意されています。FastAPIでエンドポイントを作成する場合には@app.get()デコレータなどでRESTエンドポイントを作成します。したがって、例えばテスト対象がRESTのリクエストメソッドgetであればTestClientで使うのもこれに合わせたgetメソッドとなります。

では早速そのパターンをまとめたいと思います。

単純なリクエストパターン

パラメータが不要なエンドポイントのテスト

以下のようなGETのリクエストメソッドに対するテストを記述します。

main.py

from fastapi import FastAPI

app = FastAPI()


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

GETなのでTestClientのメソッドもgetメソッドを使って記述します。

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"}

パスパラメータを使ったリクエストのテスト

GETのリクエストメソッドに対するテストを記述しますが先程と違いパスパラメータを引数として取るエンドポイントに対するテストを記述します。

main.py

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/{item_id}")
async def read_item(item_id):
    return {"item_id": item_id}

先ほどと同じGETなのでTestClientのメソッドもgetメソッドを使って記述しますが、パスパラメータの箇所が違うのでここにテストしたい値を入れます。

import pytest
from fastapi.testclient import TestClient

from main import app

client = TestClient(app)

@pytest.mark.parametrize(
    ["item_id"],
    [
        pytest.param(
            "hogehoge",
        ),
    ],
)
def test_read_main(item_id: str):
    response = client.get(f"/items/{item_id}")
    assert response.status_code == 200
    assert response.json() == {"item_id": item_id}

クエリパラメータを使ったリクエストのテスト

次はクエリパタメータを使ったエンドポイントに対するテストを記述します。

main.py

from fastapi import FastAPI

app = FastAPI()

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
    return fake_items_db[skip : skip + limit]

クエリパタメータを使ったエンドポイントに対するリクエストにはTestClientの各メソッドのparamsパラメータを使います。

import pytest
from fastapi.testclient import TestClient

from main import app

client = TestClient(app)

@pytest.mark.parametrize(
    ["skip","limit"],
    [
        pytest.param(
            0,2
        ),
    ],
)
def test_read_item(skip: int, limit: int):
    response = client.get(
        "/items/",
        params={"skip": skip, "limit": limit},
    )
    assert response.status_code == 200
    assert response.json() == [{'item_name': 'Foo'}, {'item_name': 'Bar'}]

一応paramsパラメータを使わず原始的に以下のような記述も可能ですが、paramsパラメータを使ったほうが拡張性は高いのであまりこの記述は使いません。

import pytest
from fastapi.testclient import TestClient

from main import app

client = TestClient(app)

@pytest.mark.parametrize(
    ["skip","limit"],
    [
        pytest.param(
            0,2
        ),
    ],
)
def test_read_item(skip: int, limit: int):
    response = client.get(
        f"/items/?skip={skip}&limit={limit}",
    )
    assert response.status_code == 200
    assert response.json() == [{'item_name': 'Foo'}, {'item_name': 'Bar'}]

リクエストボディを使ったリクエストのテスト

POSTのリクエストメソッドでリクエストボディを使ったエンドポイントに対するテストを記述します。

main.py

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel


class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None


app = FastAPI()


@app.post("/items/")
async def create_item(item: Item):
    return item

POSTなのでTestClientのメソッドもpostメソッドを使って記述し、リクエストボディを使ったエンドポイントに対するリクエストにはTestClientの各メソッドのjsonパラメータを使います。

import pytest
from fastapi.testclient import TestClient

from main import app

client = TestClient(app)

@pytest.mark.parametrize(
    ["item"],
    [
        pytest.param(
            {
                "name": "Foo",
                "description": "An optional description",
                "price": 45.2,
                "tax": 3.5
            }
        ),
    ],
)
def test_create_item(item: dict):
    response = client.post(
        "/items/",
        json=item,
    )
    assert response.status_code == 200
    assert response.json() == item

ヘッダーのパラメータを使ったリクエストのテスト

ヘッダーのパラメータを使ったリクエストに対するテストを記述します。このエンドポイントの処理はリクエストヘッダーに含まれるuser_agentの値をそのまま返す処理になります。

main.py

from typing import Union

from fastapi import FastAPI, Header

app = FastAPI()


@app.get("/items/")
async def read_items(user_agent: Union[str, None] = Header(default=None)):
    return {"User-Agent": user_agent}

リクエストヘッダーを使ったエンドポイントに対するリクエストにはTestClientの各メソッドのheadersパラメータを使います

import pytest
from fastapi.testclient import TestClient

from main import app

client = TestClient(app)

@pytest.mark.parametrize(
    ["user_agent"],
    [
        pytest.param(
            "user_agent_sample"
        ),
    ],
)
def test_read_items(user_agent: str):
    response = client.get(
        "/items/",
        headers={"user_agent": user_agent},
    )
    assert response.status_code == 200
    assert response.json() == {"User-Agent": user_agent}

クッキーのパラメータを使ったリクエストのテスト

クッキーのパラメータを使ったリクエストに対するテストを記述します。このエンドポイントの処理はcookieにads_idがある場合にその中身を返すエンドポイントになります。

main.py

from typing import Union

from fastapi import Cookie, FastAPI

app = FastAPI()


@app.get("/items/")
async def read_items(ads_id: Union[str, None] = Cookie(default=None)):
    return {"ads_id": ads_id}

cookieもHTTPリクエストヘッダーの1種類ですがheadersパラメータではなくTestClientの各メソッドのcookiesパラメータを使います。

import pytest
from fastapi.testclient import TestClient

from main import app

client = TestClient(app)

@pytest.mark.parametrize(
    ["ads_id"],
    [
        pytest.param(
            "ads_id_sample"
        ),
    ],
)
def test_read_items(ads_id: str):
    response = client.get(
        "/items/",
        cookies={"ads_id": ads_id},
    )
    assert response.status_code == 200
    assert response.json() == {"ads_id": ads_id}

フォームデータを使ったリクエストのテスト

フォームデータを使ったリクエストに対するテスを記述します。

main.py

from fastapi import FastAPI, Form

app = FastAPI()


@app.post("/login/")
def login(username: str = Form(), password: str = Form()):
    return {"username": username}

フォームデータを使ったエンドポイントに対するリクエストにはTestClientの各メソッドのdataパラメータを使います。dataパラメータはdict形式になりますが、キーはエンドポイントのメソッドでFormクラスで指定した引数名と合わせる必要があります。これを間違えると422 Unprocessable Entityのエラーが出ます。

import pytest
from fastapi.testclient import TestClient

from main import app

client = TestClient(app)

@pytest.mark.parametrize(
    ["username","password"],
    [
        pytest.param(
            "username_1", "password_1"
        ),
    ],
)
def test_login(username: str, password: str):
    response = client.post(
        "/login/",
        data={"username": username, "password": password},
    )
    assert response.status_code == 200
    assert response.json() == {"username": username}

ファイルアップロードを行うリクエストのテスト

FastAPIでファイルを送信するリクエストに対するテストを記述します。

main.py

from typing import Annotated

from fastapi import FastAPI, File

app = FastAPI()


@app.post("/files/")
def create_file(file: Annotated[bytes, File()]):
    return {"file_size": len(file)}

ファイル送信を行うエンドポイントに対するリクエストにはTestClientの各メソッドのfilesパラメータを使います。filesパラメータはdict形式になりますが、キーはエンドポイントのメソッドでFileクラスで指定した引数名と合わせる必要があります。これを間違えると先ほどと同じ様に422 Unprocessable Entityのエラーが出ます。

import os
import tempfile

import pytest
from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


@pytest.mark.parametrize(
    ["file_name", "file_content"],
    [
        pytest.param(
            "file_name_1", "sample text 0123456789"
        ),
    ],
)
def test_login(file_name: str, file_content: str):
    with tempfile.TemporaryDirectory(dir="/tmp/") as tmpdir:
        filename = os.path.join(tmpdir, file_name)
        with open(filename, "w+") as f:
            f.write(file_content)

        file_data = {"file": (file_name, open(filename, "rb"))}

        response = client.post(
            "/files/",
            files=file_data,
        )

        assert response.status_code == 200
        assert response.json() == {"file_size": len(file_content)}

複合リクエストパターン

先ほどまではそれぞれシンプルなリクエストパターンでしたが、FastAPIを使ってエンドポイントを実装していくといくつかのリクエストを組み合わせる処置が必要になってきます。次はこのパターンを考えてみます。

CSRFトークンで保護されたエンドポイントへのテスト

POSTのリクエストメソッドを行うエンドポイントがありますが、CSRF対策でCSRFトークンが必要なエンドポイントに対するテストを記述してみます。

このエンドポイントの処理の想定としてははじめに/csrf_token/エンドポイントにGETリクエストメソッドでリクエストを送りレスポンスとして{"csrf_token":{CSRFトークン}"} が返ってきます。このCSRFトークンを使って/items/エンドポイントにPOSTリクエストメソッドでitemを追加します。

このエンドポイントに対するテストは以下のようになります。

import pytest
from fastapi.testclient import TestClient

from main import app

client = TestClient(app)

@pytest.mark.parametrize(
    ["item"],
    [
        pytest.param(
            {
                "name": "Foo",
                "description": "An optional description",
                "price": 45.2,
                "tax": 3.5
            }
        ),
    ],
)
def test_create_item(item: dict):

    # CSRFトークンを取得するTestClientの処理
    response_csrf = client.get("/csrf_token/")
    response_csrf_body = response_csrf.json()
    csrf_token = response_csrf_body["csrf_token"]


    # itemを追加する処理
    response = client.post(
        "/items/",
        json=item,
        headers={"x-csrf-token": csrf_token}
    )
    assert response.status_code == 200
    assert response.json() == item

ポイントとしては先にCSRFトークンをTestClientのgetメソッドで取得し、その結果を使ってitemを追加するPOSTメソッドのテストを行っています。

Loginが必要なエンドポイントへのテスト

Login処理が必要なエンドポイントに対するテストを記述してみます。

このエンドポイントの処理の想定としてははじめに/login/エンドポイントにPOSTリクエストメソッドでリクエストを送りレスポンスとして{"access_token":{Accessトークン}"} が返ってきます。このAccessトークンを使って/items/エンドポイントにPOSTリクエストメソッドでitemを追加します。さらに先ほどと同じ様にPOSTはCSRF対策がされているとします。

このエンドポイントに対するテストは以下のようになります。

import pytest
from fastapi.testclient import TestClient

from main import app

client = TestClient(app)

@pytest.mark.parametrize(
    ["username", "password", "item"],
    [
        pytest.param(
            "username_1", "password_1"
            {
                "name": "Foo",
                "description": "An optional description",
                "price": 45.2,
                "tax": 3.5
            }
        ),
    ],
)
def test_create_item(username: str, password: str, item: dict):

    # CSRFトークンを取得するTestClientの処理
    response_csrf = client.get("/csrf_token/")
    response_csrf_body = response_csrf.json()
    csrf_token = response_csrf_body["csrf_token"]

    # Accessトークンを取得するTestClientの処理
    response = client.post(
        "/login/",
        data={"username": username, "password": password},
        headers={"x-csrf-token": csrf_token},
    )
    response_csrf_body = response_csrf.json()
    access_token = response_csrf_body["access_token"]


    # CSRFトークンを取得するTestClientの処理(2度目）
    response_csrf = client.get("/csrf_token/")
    response_csrf_body = response_csrf.json()
    csrf_token = response_csrf_body["csrf_token"]

    # itemを追加する処理
    response = client.post(
        "/items/",
        json=item,
        cookies={"access_token": access_token},
        headers={"x-csrf-token": csrf_token}
    )
    assert response.status_code == 200
    assert response.json() == item

ポイントしてしてははじめにCSRFトークンを取得しその結果を使ってLogin処理をします。Login処理ではAccessトークンが返ってくるのでAccessトークンを使うのと再びCSRFトークンを取得してからitemの追加処理のテストをしています。

まとめ

FastAPIで作成したエンドポイントに対するPytestでテストコードの様々なパターンを記述しました。組み合わせパターンで記述したCSRFトークンの取得やログイン処理はより高度なテストを記述するためにこれらをfixtureとして共通化することがポイントかと思います。どなたかの参考になれば幸いです。

最後まで読んで頂いてありがとうございました。

AWS

関連記事

FastAPIのMiddleware、CustomRoute、Dependenciesの実行順序を確認してみる

kobayashi.m

2023.12.28

Hypercorn+FastAPIのコンテナイメージを作成してみる

kobayashi.m

2023.10.20

「動かして学ぶ！Python FastAPI開発入門」著者の中村 翔様へインタビューしてみた。

Tanner

2023.08.31

APSchedulerをFastAPIで使ってみる

kobayashi.m

2023.08.30