環境

Python: 3.8
- fastapi==0.68.2
- SQLAlchemy==1.4.25
- SQLAlchemy-Utils==0.37.8
- pytest==6.2.5
- factory-boy==3.2.0

アプリケーション準備

ユーザ認証とブログ管理を行う簡単なアプリケーションを用意する。
※アプリケーション本体のコードは割愛するためリポジトリを参照

※ディレクトリ構成

.
├── Pipfile
├── Pipfile.lock
├── README.md
├── alembic.ini
├── app
│   ├── __init__.py
│   ├── core
│   │   ├── __init__.py
│   │   ├── celery_app.py
│   │   ├── config.py
│   │   └── security.py
│   ├── crud
│   ├── database.py
│   ├── dependencies.py
│   ├── main.py
│   ├── middlewares.py
│   ├── models
│   ├── routers
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   ├── blogs.py
│   │   └── users.py
│   ├── schemas.py
│   ├── tests
│   │   ├── ...
│   └── utils
│       ├── oauth2.py
│       └── token.py
└── migration
│   ├── README
│   ├── env.py
│   ├── script.py.mako
│   └── versions
│       ├── ...

本題

データベース準備

DockerコンテナでPostgreSQLを起動しておく

$ docker run -d --rm \
   --name test_db \
  -p 5433:5432 \
  -e POSTGRES_USER=app \
  -e POSTGRES_PASSWORD=secret \
  -e POSTGRES_DB=app \
  postgres:13.4

フィクスチャ準備

テスト用データベースを使用する設定をconftest.pyに仕込んでおき、データベースをfixtureとして使用できるようにしておく。

※こちらのGitHub Issueも参照されたし
Separating database for tests and dev · Issue #125 · tiangolo/full-stack-fastapi-postgresql

`app/tests/conftest.py`

app/tests/conftest.py

from typing import Generator
import pytest
from sqlalchemy.orm import Session
from sqlalchemy_utils import database_exists, create_database

from app.core.config import settings
from app.database import Base
from tests.utils.overrides import TestingSessionLocal, engine


@pytest.fixture(scope="function")
def db() -> Generator[Session, None, None]:
    if not database_exists(settings.SQLALCHEMY_DATABASE_URI):
        create_database(settings.SQLALCHEMY_DATABASE_URI)

    Base.metadata.create_all(bind=engine)
    session = TestingSessionLocal()
    yield session
    session.close()
    Base.metadata.drop_all(bind=engine)

ポイントは、dbフィクスチャのスコープを"function"としている点。
こうすればテスト関数毎にセッションをクローズ&データベース初期化が行われ、データがクリーンアップされる。

app/tests/conftest.py

...
    session = TestingSessionLocal()
    yield session
    session.close()
    Base.metadata.drop_all(bind=engine)

テスト用データ作成

factory-boyを使いFactoryクラスを作成しておく。
テストケースの中で簡単にインスタンス化できるようにする。

`app/tests/factories.py`

app/tests/factories.py

from factory.alchemy import SQLAlchemyModelFactory

import app.models.user as models
from app.tests.utils.overrides import TestingSessionLocal


class UserFactory(SQLAlchemyModelFactory):
    name = 'skokado'
    email = 'skokado@example.com'
    password = 'MyPassw0rd!'

    class Meta:
        model = models.User
        sqlalchemy_session = TestingSessionLocal()

テストケース

テストケースの中でdbフィクスチャを使用してデータ作成をするのみで良い。

`app/tests/routers/test_auth.py`

app/tests/routers/test_auth.py

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

from app.main import app
import app.crud.users as users_crud
from app.tests import factories

client = TestClient(app)


class TestAuthRouter:

    def test_ログイン成功(self, db: Session):
        user_in = factories.UserFactory()
        users_crud.create(user_in, db)
        response = client.post(
            '/api/login',
            data={
                'grant_type': 'password',
                'username': user_in.name,
                'password': user_in.password
            }
        )
        assert response.status_code == 200

`app/tests/routers/test_blogs.py`

app/tests/routers/test_blogs.py

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

from app.main import app
import app.crud.users as users_crud
from app.core.security import create_access_token
from app.tests import factories

client = TestClient(app)
client.headers.update({
    'Authorization': 'Bearer {}'.format(create_access_token)
})


class TestBlogRouter:
    def test_list_empty_blogs(self, db: Session):
        user_in = factories.UserFactory()
        users_crud.create(user_in, db)
        # ログインする
        response = client.post(
            '/api/login',
            data={
                'grant_type': 'password',
                'username': user_in.name,
                'password': user_in.password
            }
        )
        jwt = response.json()
        headers = {
            'Authorization': f'{jwt["token_type"].capitalize()} {jwt["access_token"]}'
        }

        # アクセストークンを使用してAPIリクエスト
        response = client.get('/api/blog/', headers=headers)
        assert response.status_code == 200
        assert response.json() == []