ChatGPT先生と挑むFastAPI #8｜pytestでAPIテスト

それではSTEP8を始めましょう！

STEP8では以下の内容を学習します。

pytestは、Pythonでテストを作成・実行するためのフレームワークです。

例えば、商品登録APIを作成した場合、これまではSwagger UIからリクエストを送信し、正しく商品が登録できるか確認していました。

pytestを利用すると、その確認作業をプログラムとして記述し、自動実行できるようになります。

FastAPIでは、TestClientという仕組みを使用できます。

これは、FastAPIアプリケーションに対して、テストコードからHTTPリクエストと同じ形式で呼び出すためのものです。

pytest：テストを実行する
      ↓
TestClient：APIへリクエストを送信する
      ↓
FastAPI：ルーティング → API処理 → レスポンス
      ↓
assert：実際の結果と期待値を比較する
      ↓
PASS または FAIL

ここで重要なのは、通常のTestClientによるテストでは、UvicornでAPIサーバーを別途起動する必要がないことです。

TestClientがPythonのプログラム内でFastAPIアプリケーションを呼び出します。

それぞれの役割を整理すると以下のようになります。

今回は、現在のFastAPIプロジェクトの構成と、テストに必要なライブラリが導入されているかを確認します。

STEP 1で必要な開発ツールをまとめて導入しているため、まずは既存環境を確認し、不足している場合だけ追加します。

VS CodeでFastAPIのプロジェクトを開き、Dev Containerに接続してください。

プロジェクトのルートディレクトリで、ターミナルから以下を実行します。

＜プロジェクト構成を確認＞

現在のPythonファイルの配置や、テスト用ディレクトリが既に存在するかを確認します。

pwd

find . -maxdepth 3 -type f \
  \( -name "*.py" -o -name "requirements*.txt" -o -name "pyproject.toml" -o -name "pytest.ini" \) \
  | sort

＜ライブラリの導入状況を確認＞

python -m pip show pytest httpx

導入済みであれば、バージョンなどの情報が表示されます。

確認しました！ STEP 8のテスト環境は、必要なライブラリが導入されている状態です。

現時点で追加のインストールは不要です。

ここから実際にテスト環境を構築していきます。

今回の目的は、テストを何度実行しても、普段使用している開発用DBのデータに影響を与えない仕組みを作ることです。

STEP 6では、Depends()を使用してAPIにSQLAlchemyのSessionを渡す仕組みを実装しました。

今回は、その仕組みをテストでも活用します。

＜通常実行とテスト実行の違い＞

＜通常実行＞
FastAPI
    ↓
get_db()：通常のSessionを取得
    ↓
開発用 PostgreSQL

＜テスト実行＞
FastAPI + TestClient
    ↓
テスト用のDB依存関数：get_db()をテスト時だけ差し替える
    ↓
テスト用 PostgreSQL

FastAPIには、app.dependency_overridesという仕組みがあります。

これを利用すると、通常はget_db()から取得しているSessionを、テスト時だけ別のSessionに差し替えられます。

ただし、差し替えただけでDBが分離されるわけではありません。 接続先をテスト専用DBに設定する必要があります。

今回のコードでは、次のようになっています。

products.py：Depends(get_db)でSessionを取得
     ↓
dependencies.py：get_db()を定義
     ↓
database.py：engineとSessionLocalを定義
     ↓
PostgreSQL

まずはDBを分離します。

現在のcompose.yamlでは、PostgreSQLが1つ起動しています。このPostgreSQLの中に、テスト専用DBを追加します。

今回はDB名をfastapi_step8_testとします。

別のPostgreSQLコンテナを起動する必要はありません。また、今のcompose.yamlも変更不要です。

Dev Containerのターミナルで、/workspaceから以下を実行してください。

python - <<'PY'
from sqlalchemy import text
from sqlalchemy.engine import make_url

from app.config import settings
from app.database import engine

# テスト専用のDB名
test_db_name = "fastapi_step8_test"

# 誤って開発用DBと同じ名前を指定しないように確認する
current_db_name = make_url(settings.database_url).database

if current_db_name == test_db_name:
    raise RuntimeError("開発用DBとテスト用DBが同じ名前です")

# CREATE DATABASEをトランザクションの外で実行する
with engine.connect().execution_options(
    isolation_level="AUTOCOMMIT"
) as conn:

    # 同名のDBが既に存在するか確認する
    exists = conn.execute(
        text("SELECT 1 FROM pg_database WHERE datname = :name"),
        {"name": test_db_name},
    ).scalar()

    if exists:
        print("同名のDBが既に存在します。内容を確認するまで使用しません。")
    else:
        # テスト専用DBを新規作成する
        conn.exec_driver_sql("CREATE DATABASE fastapi_step8_test")
        print("テスト用DBを作成しました")
PY

複数人で開発するプロジェクトなら、テスト用DBの構築手順もコードとしてGitで管理し、誰でも同じ環境を再現できるようにするべきです。

先ほど提示した方法は、一度だけDBを作成するための作業用コードでした。今回の学習では実務で使える構成を目指しているので、あの方法は変更しましょう。

先ほどのターミナルで直接Pythonを実行する方法は採用しません。

代わりに、次の構成を実装する方針にします。

• compose.yaml：テスト専用PostgreSQLを追加
• tests/conftest.py：DB接続の差し替えとテスト前後の処理を共通化
• Alembic：テスト用DBにも既存のテーブル定義を反映

今回の作業では、既存のcompose.yamlにテスト専用のPostgreSQLを追加します。

開発用DBとは別のコンテナ・保存領域を用意することで、テストによるデータ変更が開発用DBに影響しないようにします。

まずはcompose.yamlを修正します。

既存のservices内に、dbと同じ階層で次の設定を追加してください。

  # テスト専用のPostgreSQL
  test-db:
    image: postgres:17

    environment:
      POSTGRES_DB: fastapi_step8_test
      POSTGRES_USER: ${POSTGRES_USER}
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}

    # 開発用DBとは別の保存領域を使用する
    volumes:
      - postgres_test_data:/var/lib/postgresql/data

    # DBが接続可能になるまでの状態を確認する
    healthcheck:
      test:
        [
          "CMD-SHELL",
          "pg_isready -U ${POSTGRES_USER} -d fastapi_step8_test"
        ]
      interval: 5s
      timeout: 5s
      retries: 10

続いて、ファイル末尾にあるvolumesを変更します。

volumes:
  postgres_data:
  postgres_test_data:

＜変更内容＞

重要なのはPOSTGRES_DBです。

PostgreSQLのDockerイメージには、初回起動時に指定された名前のデータベースを自動作成する仕組みがあります。

そのため、先ほどのようにPythonコードからCREATE DATABASEを実行する必要はありません。

ただし、自動作成されるのは保存領域が空の初期化時だけです。

また、今回はテスト用DBのポートをホストPCに公開していません。Dev Containerなど、同じDocker Composeネットワーク内のコンテナからtest-db:5432で接続する想定です。

次に、.devcontainer/devcontainer.jsonのrunServicesに"test-db"を追加してください。

"runServices": [
  "db",
  "test-db"
],

runServicesは、Dev Containers起動時に起動するDocker Composeのサービスを指定する設定です。

今回の設定では、開発用のdbに加え、テスト用のtest-dbも起動対象になります。

設定を保存したら、VS Codeで以下を実行してください。

1. Ctrl + Shift + Pでコマンドパレットを開く。
2. Dev Containers: Rebuild Containerを選択する。
3. 再構築が完了するまで待つ。

今回作成するのは、次の2ファイルです。

• conftest.py：テスト用DBへの接続設定を共通化
• test_database.py：接続先がテスト用DBであることを検証

conftest.pyにはpytestのfixtureという仕組みを定義します。fixtureはテストで共通利用する準備・後片付けの処理をまとめるためのものです。（公式ドキュメント）

Dev Containerの/workspace直下にtestsディレクトリを作成し、conftest.pyを追加してください。

from collections.abc import Iterator

import pytest
from sqlalchemy import create_engine
from sqlalchemy.engine import Engine, make_url

from app.config import settings

# Docker Composeで定義したテスト用DBの接続先
TEST_DB_HOST = "test-db"
TEST_DB_NAME = "fastapi_step8_test"


@pytest.fixture(scope="session")
def test_engine() -> Iterator[Engine]:
    # 既存のDB接続設定を取得する
    development_url = make_url(settings.database_url)

    # 開発用DBの設定がテスト用DBを指していないことを確認する
    if (
        development_url.host == TEST_DB_HOST
        or development_url.database == TEST_DB_NAME
    ):
        raise RuntimeError("開発用DBの接続設定を確認してください")

    # ホスト名とDB名をテスト用に変更する
    test_url = development_url.set(
        host=TEST_DB_HOST,
        port=5432,
        database=TEST_DB_NAME,
    )

    # テスト専用のEngineを作成する
    engine = create_engine(test_url)

    try:
        # テスト側へEngineを渡す
        yield engine
    finally:
        # テスト終了後に接続プールを破棄する
        engine.dispose()

ここでの重要なポイントは、@pytest.fixture(scope="session")を指定すると、このfixtureは一度のpytest実行中で共有される点です。

また、yieldでテストにEngineを渡し、テスト終了後にfinallyでEngineの接続プールを破棄しています。

これはDBのデータを削除する処理ではありません。

このfixtureは、今後のAPIテストでも再利用します。

続いて、接続確認用のテストを作成します。testsディレクトリにtest_database.pyを追加してください。

from sqlalchemy import text
from sqlalchemy.engine import Engine


def test_connect_to_test_database(test_engine: Engine) -> None:
    # テスト用DBに接続する
    with test_engine.connect() as conn:
        # 実際に接続しているデータベース名を取得する
        db_name = conn.execute(
            text("SELECT current_database()")
        ).scalar_one()

    # テスト専用DBへ接続できていることを検証する
    assert db_name == "fastapi_step8_test"

テスト関数の引数にtest_engineを指定すると、pytestが同名のfixtureを探して実行し、その結果を渡してくれます。

今回は実際のPostgreSQLへ接続し、DB名が期待値と一致することを確認しています。

pytestで動作確認をします。Dev Container内の/workspaceで、次のコマンドを実行してください。

python -m pytest tests/test_database.py -v

※-sを付けるとprint文の標準出力もターミナルに表示されるようになります。

期待する結果は以下です。

tests/test_database.py::test_connect_to_test_database PASSED

================== 1 passed ==================

これで、テスト用DBへの接続確認がpytestで再実行できるようになります。

なお、現段階ではテスト用DBにテーブルは作成しておらず、FastAPIのget_db()もまだ差し替えていません。今回のテストは接続確認のみです。

次は、テスト専用DBに商品テーブルを作成します。

STEP 6ではAlembicを使用してマイグレーションを実装しましたね。

今回もその仕組みを利用し、開発用DBと同じマイグレーションをテスト用DBに適用する構成にします。

現在のmigrations/env.pyには次の処理があります。

config.set_main_option(
    "sqlalchemy.url",
    os.environ["DATABASE_URL"],
)

これはAlembicの接続先を環境変数のDATABASE_URLに設定する処理です。

現在のままalembic upgrade headを実行すると、開発用DBにマイグレーションが適用されます。

そこで、通常実行では従来の接続先を使用し、pytestから呼び出す場合だけテスト用DBのConnectionを渡せるようにします。（公式ドキュメント）

migrations/env.pyのrun_migrations_online()を、以下のコードに置き換えてください。

def run_migrations_online() -> None:
    """オンラインモードでマイグレーションを実行する。"""

    # pytestなどの外部プログラムからConnectionが渡されているか確認する
    connection = config.attributes.get("connection")

    if connection is not None:
        # 渡されたConnectionを使用してマイグレーションする
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
        )

        with context.begin_transaction():
            context.run_migrations()

        return

    # Connectionが渡されていなければ、通常のDB接続を使用する
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
        )

        with context.begin_transaction():
            context.run_migrations()

これにより、動作が次のように変わります。

＜通常の実行＞
alembic upgrade head
     ↓
環境変数のDATABASE_URLを使用：開発用DB

＜pytestから実行＞
テスト用ConnectionをAlembicに渡す
     ↓
渡されたConnectionを使用：テスト用DB

ポイントは、この部分です。

connection = config.attributes.get("connection")

外部からConnectionが渡されていれば、新しくEngineを作成せず、そのConnectionを使用します。

１．tests/conftest.pyを修正

まず、既存のimportに以下を追加してください。

from pathlib import Path

from alembic import command
from alembic.config import Config
from sqlalchemy import text

次に、既存のtest_engine()の下へ新しいfixtureを追加します。

@pytest.fixture(scope="session")
def migrated_engine(test_engine: Engine) -> Engine:
    # テスト用DBへ接続し、トランザクションを開始する
    with test_engine.begin() as connection:

        # 実際の接続先DB名を取得する
        db_name = connection.execute(
            text("SELECT current_database()")
        ).scalar_one()

        # 開発用DBへの誤適用を防止する
        if (
            test_engine.url.host != TEST_DB_HOST
            or db_name != TEST_DB_NAME
        ):
            raise RuntimeError("テスト用DB以外にはマイグレーションできません")

        # プロジェクトルートのalembic.iniを読み込む
        alembic_ini = Path(__file__).resolve().parents[1] / "alembic.ini"
        alembic_config = Config(str(alembic_ini))

        # Alembicへテスト用DBのConnectionを渡す
        alembic_config.attributes["connection"] = connection

        # テスト用DBを最新のマイグレーションまで更新する
        command.upgrade(alembic_config, "head")

    # マイグレーション適用済みのEngineを返す
    return test_engine

ポイントは、fixture同士を組み合わせていることです。

• test_engine：テスト用DBのEngineを作成する。
• migrated_engine：そのEngineを利用してマイグレーションを適用する。

scope="session"なので、pytestの1回の実行につき、最初にこのfixtureが必要になった際に1回実行されます。

また、command.upgrade()は既存のマイグレーション履歴に基づいて未適用の変更を適用するため、通常はテスト実行のたびにテーブルを作り直すわけではありません。

２．tests/test_database.pyにテストを追加

現在の接続確認テストは残したまま、次のコードを追加してください。

importにinspectを追加します。

from sqlalchemy import inspect, text

続いて、テスト関数を追加します。

def test_products_table_exists(migrated_engine: Engine) -> None:
    # マイグレーション適用済みDBのテーブル情報を取得する
    inspector = inspect(migrated_engine)

    # productsテーブルが存在することを検証する
    assert inspector.has_table("products")

これにより、テスト用DBへのマイグレーションが正しく実行され、productsテーブルが存在するかを確認できます。

３．動作確認

Dev Container内の/workspaceで実行してください。

python -m pytest tests/test_database.py -v

期待する結果は以下です。

tests/test_database.py::test_connect_to_test_database PASSED
tests/test_database.py::test_products_table_exists PASSED

================== 2 passed ==================

成功すれば、テスト用DBへの接続と、productsテーブルの存在の確認ができたことになります。

今回は、tests/conftest.pyにpytestの共通fixtureを追加します。

目指すのは次の2点です。

• テスト実行時だけ、FastAPIのget_db()をテスト用Sessionに差し替える。
• API内でdb.commit()しても、テスト終了後に変更をロールバックする。

FastAPIのdependency_overridesと、SQLAlchemyのトランザクション管理を組み合わせて実現します。

1. 仕組みを理解する

pytest：テスト用Sessionを作成
      ↓
dependency_overrides：get_db()を差し替える
      ↓
TestClient → FastAPI：テスト用SessionでAPIを実行
      ↓
テスト終了：変更をロールバック

特に重要なのがロールバックです。

商品登録APIにはdb.commit()がありますが、今回は外側に別のトランザクションを用意し、SessionにはSAVEPOINTを利用させます。

これにより、API側でcommit()を実行しても、テスト終了時に外側のトランザクションをロールバックして変更を取り消せます。

2. tests/conftest.pyを修正

既存のコードは残し、次のimportを追加してください。

from fastapi.testclient import TestClient
from sqlalchemy.orm import Session

from app.dependencies import get_db
from app.main import app

続いて、ファイル末尾に2つのfixtureを追加します。

@pytest.fixture
def db_session(migrated_engine: Engine) -> Iterator[Session]:
    # テスト用DBへ接続する
    with migrated_engine.connect() as connection:

        # テスト全体を囲むトランザクションを開始する
        transaction = connection.begin()

        try:
            # API内のcommit()が外側のトランザクションを
            # 確定しないようにSAVEPOINTを使用する
            with Session(
                bind=connection,
                join_transaction_mode="create_savepoint",
            ) as session:
                yield session

        finally:
            # テストで変更したデータを元に戻す
            transaction.rollback()


@pytest.fixture
def client(db_session: Session) -> Iterator[TestClient]:

    # 通常のget_db()の代わりにテスト用Sessionを返す
    def override_get_db() -> Iterator[Session]:
        yield db_session

    # FastAPIのDB依存性をテスト用に差し替える
    app.dependency_overrides[get_db] = override_get_db

    try:
        # テスト用のAPIクライアントを作成する
        with TestClient(app) as test_client:
            yield test_client

    finally:
        # テスト終了後、今回設定した依存性の差し替えを解除する
        app.dependency_overrides.pop(get_db, None)

3. 今回の重要なポイント

@pytest.fixture
def client(db_session: Session) -> Iterator[TestClient]:

今回は、商品登録APIを使って次の動作を確認します。

• FastAPIがテスト用Sessionを使用していること。
• API内でdb.commit()しても、外側のトランザクションは確定されないこと。
• テスト終了後に登録データが残らないこと。

create_savepointは、外側のトランザクションを保持したままSessionのcommit()を扱うための設定です。テスト終了時には外側のトランザクションをロールバックします。

1. テストコードを作成する

tests/test_products.pyを新規作成し、以下を記述してください。

from fastapi.testclient import TestClient
from sqlalchemy import select
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session

from app.models.product import Product


def test_product_create_rollback(
    client: TestClient,
    db_session: Session,
    migrated_engine: Engine,
) -> None:
    # ロールバック確認用の商品名
    name = "pytest_rollback_check"

    # 前回のテストデータが残っていないことを確認する
    with migrated_engine.connect() as connection:
        result = connection.execute(
            select(Product.id).where(Product.name == name)
        ).first()

        assert result is None

    # API経由で商品を登録する
    response = client.post(
        "/products",
        json={
            "name": name,
            "price": 1000,
        },
    )

    assert response.status_code == 201

    product_id = response.json()["id"]

    # テスト用Sessionから登録データを取得できることを確認する
    product = db_session.get(Product, product_id)

    assert product is not None
    assert product.name == name

    # 別のConnectionからは未確定データが見えないことを確認する
    with migrated_engine.connect() as connection:
        result = connection.execute(
            select(Product.id).where(Product.id == product_id)
        ).first()

        assert result is None

2. このテストで確認している仕組み

＜データの見え方＞
テスト用Session：登録した商品を取得できる。
別のConnection：未確定の登録データは取得できない。
テスト終了後：fixtureが外側のトランザクションをロールバックする。

今回はPostgreSQLの通常のトランザクション分離設定を前提にしています。

なお、テスト中に別のConnectionからデータが見えないことだけでは、終了時のロールバックまで確認できません。そのため、同じテストを再実行してデータが残っていないことも確認します。

3. 動作確認

Dev Container内の/workspaceで実行してください。

python -m pytest tests/test_database.py tests/test_products.py -v

期待する結果：

test_connect_to_test_database PASSED
test_products_table_exists PASSED
test_product_create_rollback PASSED

3 passed

成功したら、同じコマンドをもう一度実行してください。

2回目も成功すれば、少なくとも今回登録した確認用商品が前回のテストから残っていないことを検証できます。

今回のWarningは2件とも、インストールされているライブラリ内部から出ています。

DeprecationWarningは、現在使っている機能が将来的に変更・廃止される可能性を知らせる警告です。

今回のログにはテストコード自体の失敗はありません。

ただし、実務では警告を放置し続けるのも好ましくありません。将来のライブラリアップデートで互換性の問題が起こる可能性があるためです。

今回はライブラリのバージョンをむやみに変更せず、依存ライブラリ由来の警告として把握しておきましょう。

ここからは、実際のAPIの動作をpytestで検証していきます。

今回は、STEP 6で実装した商品詳細取得APIを対象にします。

GET /products/{product_id}

テスト用DBに商品を登録し、GET APIでその商品を取得できるか確認します。

今回は、テストの前提となる商品データを直接SQLAlchemyで登録します。

これにより、POST APIの正しさに依存せず、GET APIそのものをテストできます。

既存のtests/test_products.pyの末尾に、以下を追加してください。

def test_get_product(
    client: TestClient,
    db_session: Session,
) -> None:
    # テスト用の商品データを作成する
    product = Product(
        name="テスト商品",
        price=1500,
    )

    # テスト用DBへ商品を登録する
    db_session.add(product)
    db_session.commit()
    db_session.refresh(product)

    # 登録した商品のIDを指定してGET APIを呼び出す
    response = client.get(f"/products/{product.id}")

    # HTTPステータスコードを検証する
    assert response.status_code == 200

    # レスポンスの内容を検証する
    assert response.json() == {
        "id": product.id,
        "name": "テスト商品",
        "price": 1500,
    }

ポイント：なぜ商品IDを固定しないのか？

response = client.get(f"/products/{product.id}")

ここでは、実際に登録された商品のIDを使用しています。

DBの採番状況によってIDが変わっても、テストが正しく動作するようにするためです。

Dev Container内の/workspaceで実行してください。

python -m pytest tests/test_products.py -v

期待する結果：

test_product_create_rollback PASSED
test_get_product PASSED

2 passed

正しい商品データをPOSTしたとき、商品が正常に登録されることを検証します。

今回のテストでは、次の3点を確認します。

1. HTTPステータスコードが201であること。
2. レスポンスに登録した商品情報が含まれること。
3. DBに商品が正しく登録されていること。

いずれも「商品登録APIが正常に動作する」という1つの振る舞いに対する検証です。

tests/test_products.pyの末尾に、次のテスト関数を追加してください。

def test_create_product(
    client: TestClient,
    db_session: Session,
) -> None:
    # 商品登録APIへ送信するリクエストデータ
    request_data = {
        "name": "POST正常系テスト商品",
        "price": 2500,
    }

    # 商品登録APIを呼び出す
    response = client.post(
        "/products",
        json=request_data,
    )

    # ステータスコードが201であることを検証する
    assert response.status_code == 201

    # レスポンスの内容を検証する
    response_data = response.json()
    product_id = response_data["id"]

    assert response_data == {
        "id": product_id,
        "name": "POST正常系テスト商品",
        "price": 2500,
    }

    # DBに商品が登録されていることを検証する
    product = db_session.get(Product, product_id)

    assert product is not None
    assert product.name == "POST正常系テスト商品"
    assert product.price == 2500

テスト関数にdb_sessionを指定しているため、APIが使用したSessionを使って登録結果を確認できます。

今回は追加したPOST APIのテストだけを実行します。

Dev Containerの/workspaceで実行してください。

python -m pytest tests/test_products.py::test_create_product -v

::test_create_productを指定すると、そのテスト関数だけを実行できます。

期待する結果：

tests/test_products.py::test_create_product PASSED

1 passed

既存商品の名前と価格をPUT APIで更新したとき、正しく変更されることを検証します。

確認するのは、HTTPステータスコード・レスポンス内容・DBの更新結果です。

今回はPUT APIだけを検証したいので、更新対象の商品はPOST APIではなく、SQLAlchemyで直接準備します。

tests/test_products.pyの末尾に追加してください。

def test_update_product(
    client: TestClient,
    db_session: Session,
) -> None:
    # 更新対象の商品を準備する
    product = Product(
        name="更新前の商品",
        price=1000,
    )

    db_session.add(product)
    db_session.commit()
    db_session.refresh(product)

    product_id = product.id

    # 商品更新APIを呼び出す
    response = client.put(
        f"/products/{product_id}",
        json={
            "name": "更新後の商品",
            "price": 2000,
        },
    )

    # ステータスコードとレスポンスを検証する
    assert response.status_code == 200

    assert response.json() == {
        "id": product_id,
        "name": "更新後の商品",
        "price": 2000,
    }

    # DBから最新の状態を再取得する
    db_session.refresh(product)

    # DBのデータが更新されていることを検証する
    assert product.name == "更新後の商品"
    assert product.price == 2000

db_session.refresh(product)で、SQLAlchemyのSessionが管理しているproductの状態をDBから再取得します。

SQLAlchemyのSessionは取得済みオブジェクトを管理しているため、Python上のオブジェクトの値だけを確認するのではなく、DBから再取得して検証しています。

また、更新前の商品を直接Sessionで準備しているため、POST APIが正常に動作することには依存しません。

Dev Container内の/workspaceで実行してください。

python -m pytest tests/test_products.py::test_update_product -v

期待する結果：

tests/test_products.py::test_update_product PASSED

1 passed

既存の商品をDELETE APIで削除したとき、DBから正しく削除されることを検証します。

今回は、ステータスコード・レスポンス・DBからの削除結果を確認します。

PUT APIのテストと同様に、事前データはSQLAlchemyで直接準備し、DELETE APIだけを検証します。

tests/test_products.pyの末尾に追加してください。

def test_delete_product(
    client: TestClient,
    db_session: Session,
) -> None:
    # 削除対象の商品を準備する
    product = Product(
        name="削除テスト商品",
        price=1000,
    )

    db_session.add(product)
    db_session.commit()
    db_session.refresh(product)

    product_id = product.id

    # 商品削除APIを呼び出す
    response = client.delete(f"/products/{product_id}")

    # ステータスコードとレスポンスを検証する
    assert response.status_code == 200

    assert response.json() == {
        "message": "Product deleted",
    }

    # DBから削除対象の商品を検索する
    deleted_product = db_session.execute(
        select(Product).where(Product.id == product_id)
    ).scalar_one_or_none()

    # 商品がDBに存在しないことを検証する
    assert deleted_product is None

以下の処理では、SQLAlchemyで商品の存在を検索しています。

deleted_product = db_session.execute(
    select(Product).where(Product.id == product_id)
).scalar_one_or_none()

scalar_one_or_none()は、検索結果が1件ならそのオブジェクト、0件ならNoneを返します。複数件の場合は例外になります。

したがって、最後のassertによって、DELETE APIが成功レスポンスを返すだけでなく、実際にDBから商品が削除されていることを検証できます。

Dev Container内の/workspaceで実行してください。

python -m pytest tests/test_products.py::test_delete_product -v

期待する結果：

tests/test_products.py::test_delete_product PASSED

1 passed

正常系では、登録済みの商品をGET APIで取得できることを検証しました。

今回は、存在しない商品IDを指定したとき、APIが適切な404エラーを返すことを検証します。

tests/test_products.pyの末尾に追加してください。

def test_get_product_not_found(
    client: TestClient,
) -> None:
    # 存在しない商品IDを指定してGET APIを呼び出す
    response = client.get("/products/-1")

    # HTTPステータスコードを検証する
    assert response.status_code == 404

    # エラーレスポンスの内容を検証する
    assert response.json() == {
        "detail": "Product not found",
    }

今回はSQLAlchemyの自動採番で作成される通常の正のIDとは異なる-1を指定し、該当商品が存在しないことを前提としています。

FastAPIでは、APIの処理中に以下のようなHTTPExceptionを発生させると、

raise HTTPException(
    status_code=404,
    detail="Product not found",
)

TestClientは、そのHTTPエラーレスポンスを取得できます。

そのため、pytestではステータスコードとレスポンス本文の両方を検証できます。

Dev Container内の/workspaceで実行してください。

python -m pytest tests/test_products.py::test_get_product_not_found -v

期待する結果：

test_get_product_not_found PASSED

1 passed

今回は、商品詳細取得APIのパスパラメータに不正な値を指定した場合をテストします。

商品詳細取得APIでは、以下のように商品IDを整数型で定義しています。

def get_product(
    product_id: int,
    db: Annotated[Session, Depends(get_db)],
):

ここに文字列を指定すると、FastAPIが整数へ変換できず、通常はHTTP 422のバリデーションエラーを返します。

今回はステータスコードだけでなく、どのパラメータでエラーになったのかも検証しましょう。

tests/test_products.pyの末尾に追加してください。

def test_get_product_invalid_id(
    client: TestClient,
) -> None:
    # 商品IDに整数へ変換できない文字列を指定する
    response = client.get("/products/abc")

    # バリデーションエラーが返ることを検証する
    assert response.status_code == 422

    # エラーが発生した項目を検証する
    errors = response.json()["detail"]

    assert any(
        error["loc"] == ["path", "product_id"]
        for error in errors
    )

error["loc"]は、バリデーションエラーが発生した場所を表します。

error["loc"] == ["path", "product_id"]

• path：パスパラメータ
• product_id：エラーが発生した項目名

今回はエラーメッセージ全文ではなく、エラーの発生箇所を検証しています。

これにより、メッセージの文言が変更されても、今回確認したい振る舞いが変わらなければテストが成功します。

Dev Container内の/workspaceで実行してください。

python -m pytest tests/test_products.py::test_get_product_invalid_id -v

期待する結果：

test_get_product_invalid_id PASSED

1 passed

存在しない商品IDを指定してPUT APIを呼び出した場合に、適切な404エラーが返ることを検証します。

既存のupdate_product()には、次の処理があります。

if product is None:
    raise HTTPException(
        status_code=404,
        detail="Product not found",
    )

今回はこの分岐をテストします。tests/test_products.pyの末尾に追加してください。

def test_update_product_not_found(
    client: TestClient,
) -> None:
    # 存在しない商品IDを指定して更新APIを呼び出す
    response = client.put(
        "/products/-1",
        json={
            "name": "更新テスト商品",
            "price": 2000,
        },
    )

    # 商品が存在しない場合は404が返ることを検証する
    assert response.status_code == 404

    # エラーレスポンスの内容を検証する
    assert response.json() == {
        "detail": "Product not found",
    }

今回も、通常の自動採番では生成されない負のIDを使用しています。

商品が存在しない場合、APIはDBの更新処理に進まず、HTTPExceptionによって404を返します。

Dev Container内の/workspaceで実行してください。

python -m pytest tests/test_products.py::test_update_product_not_found -v

期待する結果：

test_update_product_not_found PASSED

1 passed

存在しない商品IDを指定してDELETE APIを呼び出したとき、HTTP 404が返ることを確認します。

これまでのGET・PUTと同様に、DELETE APIにも存在チェックが実装されています。

tests/test_products.pyの末尾に追加してください。

def test_delete_product_not_found(
    client: TestClient,
) -> None:
    # 存在しない商品IDを指定して削除APIを呼び出す
    response = client.delete("/products/-1")

    # 商品が存在しない場合は404が返ることを検証する
    assert response.status_code == 404

    # エラーレスポンスの内容を検証する
    assert response.json() == {
        "detail": "Product not found",
    }

今回は、DELETE APIにおける「対象商品が存在しない場合の振る舞い」だけを検証します。

Dev Container内の/workspaceで実行してください。

python -m pytest tests/test_products.py::test_delete_product_not_found -v

期待する結果：

test_delete_product_not_found PASSED

1 passed

今回は、POST APIに不正な商品データを送信したときの動作をテストします。

商品登録APIに対し、価格に整数へ変換できない文字列を指定します。

{
  "name": "テスト商品",
  "price": "abc"
}

ProductCreateのpriceが整数型として定義されていれば、FastAPIはリクエストボディの検証時にHTTP 422を返します。

今回は次の内容を検証します。

• HTTPステータスコードが422であること。
• エラーが発生した項目がpriceであること。

tests/test_products.pyの末尾に追加してください。

def test_create_product_invalid_price(
    client: TestClient,
) -> None:
    # 価格に整数へ変換できない文字列を指定する
    response = client.post(
        "/products",
        json={
            "name": "テスト商品",
            "price": "abc",
        },
    )

    # バリデーションエラーが返ることを検証する
    assert response.status_code == 422

    # エラーが発生した項目を検証する
    errors = response.json()["detail"]

    assert any(
        error["loc"] == ["body", "price"]
        for error in errors
    )

前回のパスパラメータのテストとの違いは、locの先頭です。

今回のテストは、商品登録処理へ進む前のリクエスト検証を対象としています。

Dev Container内の/workspaceで実行してください。

python -m pytest tests/test_products.py::test_create_product_invalid_price -v

期待する結果：

test_create_product_invalid_price PASSED

1 passed

これが最後のテストです。

商品登録APIで必須項目のnameを省略した場合に、HTTP 422が返ることを検証します。

先ほどは型が不正なケースでしたが、今回は必須項目が存在しないケースです。

tests/test_products.pyの末尾に追加してください。

def test_create_product_missing_name(
    client: TestClient,
) -> None:
    # 必須項目のnameを省略して商品登録APIを呼び出す
    response = client.post(
        "/products",
        json={
            "price": 1000,
        },
    )

    # バリデーションエラーが返ることを検証する
    assert response.status_code == 422

    # nameがエラーの対象になっていることを検証する
    errors = response.json()["detail"]

    assert any(
        error["loc"] == ["body", "name"]
        for error in errors
    )

今回も、エラーメッセージ全文ではなく、エラーが発生した項目を検証します。

Dev Container内の/workspaceで実行してください。

python -m pytest tests/test_products.py::test_create_product_missing_name -v

期待する結果：

test_create_product_missing_name PASSED

1 passed

これまでは各テストケースを個別に実行してきました。

今回は、すべてのテストをまとめて実行し、テスト同士が影響し合わずに成功するか確認します。

Dev Container内の/workspaceで、以下を実行してください。

python -m pytest tests -v

testsディレクトリを指定すると、pytestが配下のテストを検出してまとめて実行します。

今回作成してきたコードがすべてそろっていれば、13件のテストが実行される想定です。

期待する結果：

13 passed

個別実行では成功していても、一括実行で失敗する場合は、テスト間のデータや設定の分離に問題がある可能性があります。

今回は、単にWarningを非表示にするのではなく、原因となっているライブラリの依存関係を整理し、Gitで管理できる形で対応します。

＜2件のWarningの原因＞

1件目については、Starlette公式がhttpx2への移行を案内しています。（公式ドキュメント）

2件目はStarlette側でも報告されている問題です。AnyIO 4.15.0以降で発生しており、修正がリリースされるまでAnyIO 4.14.2に一時的に固定する方法を検討できます。（Git issues）

Warning解消①｜httpx2の導入

現在のrequirements.txtにhttpx2を追加してください。

変更後：

fastapi[standard]
sqlalchemy
alembic
psycopg[binary]
pydantic-settings
pytest
pytest-asyncio
httpx
httpx2
ruff

既存のhttpxは削除しません。fastapi[standard]の依存関係にも含まれているためです。StarletteのTestClientはhttpx2がインストールされていれば、そちらを優先的に使用します。

requirements.txtを保存したら、VS CodeのコマンドパレットからDev Containers: Rebuild Containerを実行してください。

再構築が完了したら、Dev Container内の/workspaceでテストを一括実行します。

python -m pytest tests -v

今回の確認目標：

• テスト13件が成功する
• httpxのWarningが消える
• BlockingPortalのWarningは残る可能性がある

Warning解消②｜BlockingPortal

残っているのは、次の警告です。

DeprecationWarning:
The anyio.abc.BlockingPortal alias is deprecated,
use anyio.from_thread.BlockingPortal instead.

Starlette 1.6.0が、AnyIO 4.15.0以降で非推奨になった参照方法を使用していることが原因です。

requirements.txtに以下の1行を追加してください。

# Starlette 1.6.0のBlockingPortal非推奨警告を回避するための暫定固定
anyio==4.14.2

4.14.2は、非推奨化が導入された4.15.0より前の正式リリースです。

バージョンを固定する理由もコメントとして残しておくことで、ほかの開発者が意図を把握できます。

ただし、これはStarlette自体の修正ではありません。将来、修正済みのStarletteが利用できるようになったら、固定を解除するか再検討します。

requirements.txtを保存したら、VS CodeのコマンドパレットからDev Containers: Rebuild Containerを実行してください。

再構築後、Dev Containerの/workspaceで実行します。

python -m pip check
python -m pytest tests -v

pip checkはインストール済みパッケージ間の依存関係に問題がないか確認するコマンドです。

今回の確認目標：

• 依存関係に問題がない
• 13 passed
• Warning 0件

Warningがすべて消えました。

今回の対応では、requirements.txtに変更内容を残しているため、Dev Containerを再構築するほかの開発者にも同じ設定を適用できます。

なお、anyio==4.14.2はWarning回避のための暫定固定なので、将来のライブラリ更新時には見直しましょう。

ここからは、今回学んだ内容を使って、自分でテストを設計・実装してみましょう。

STEP 6で作成した、次のAPIを対象にします。

POST /products/{product_id}/duplicate

このAPIは、既存の商品をもとに、名前と価格が同じ新しい商品を作成するものです。

tests/test_products.pyに、以下のテスト関数を新規追加してください。

def test_duplicate_product(
    client: TestClient,
    db_session: Session,
) -> None:
    # ここから自力で実装

＜テストの要件＞

• SQLAlchemyで複製元の商品を準備する
• 商品複製APIを呼び出す
• HTTPステータスコードが201であることを確認する
• 複製元と複製先の商品IDが異なることを確認する
• 複製後の商品名と価格が元の商品と同じであることを確認する
• DBに複製元と複製先の両方が存在することを確認する

今回のテスト目的は、既存商品から、正しい内容の別の商品が複製されることを検証することです。

異常系のテストは今回の課題には含めません。

実装後、Dev Containerの/workspaceで実行してください。

python -m pytest tests/test_products.py::test_duplicate_product -v

期待する結果は1 passedです。

確認しました！ 問題ありません。自力課題の要件を満たしています。