Pythonのpytestを使用してデータベースアクセスも含むテストを書く時の実装メモ
例えば下記の様なメソッドをテストしたい場合
from sqlalchemy.ext.asyncio import AsyncSession
from {user_crudのインポート} import user_crud
async def get_user_list(db: AsyncSession):
type = 1
# 指定したユーザーのタイプでDBからリストを取得
user_list = await user_crud.get_list(type, db)
return user_list
テストコード
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import NullPool
# 使用するデータベースURL設定
TEST_DATABASE_URL ="postgresql+asyncpg://postgres:{パスワード}@db:5432/{DB名}"
# エンジン設定
engine_test = create_async_engine(
TEST_DATABASE_URL, poolclass=NullPool, echo=True
)
# セッションの作成
TestingSessionLocal = sessionmaker(
autocommit=False, autoflush=False, bind=engine_test, class_=AsyncSession
) # type: ignore
@pytest_asyncio.fixture
async def db():
async with TestingSessionLocal() as session:
yield session
@pytest.mark.asyncio
async def test_get_user_list(
db
) -> None:
"""テストユーザー数を取得テスト"""
# ここで、対象となる関数を呼び出す
result = await get_user_list(db)
# 結果の検証
assert len(result) = 1
細かい説明は省きますが、テスト用のDBのセッションを作成して渡すという感じになります。
今回は非同期処理させる為に「pytest_asyncio」を使っています。