httpx.AsyncClientFastAPI を使用して pytest と sqlalchemy AsyncSessionをセットアップしようとしています。非同期のものを除いて、すべてが実質的にFastAPI Fullstack repoのテストを模倣しています。
CRUD 単体テストに問題はありません。この問題は、httpx lib から AsyncClient を使用して API テストを実行するときに発生します。
問題は、クライアントによって行われたリクエストは、クライアント フィクスチャを初期化 (セットアップ) する前に作成されたユーザー (私の場合) にしかアクセスできないことです。
私のpytest conftest.pyセットアップは次のようなものです：
from typing import Dict, Generator, Callable
import asyncio
from fastapi import FastAPI
import pytest
# from sqlalchemy.orm import Session
from sqlalchemy.ext.asyncio import AsyncSession
from httpx import AsyncClient
import os
import warnings
import sqlalchemy as sa
from alembic.config import Config
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
async def get_test_session() -> Generator:
test_engine = create_async_engine(
settings.SQLALCHEMY_DATABASE_URI + '_test',
echo=False,
)
# expire_on_commit=False will prevent attributes from being expired
# after commit.
async_sess = sessionmaker(
test_engine, expire_on_commit=False, class_=AsyncSession
)
async with async_sess() as sess, sess.begin():
yield sess
@pytest.fixture(scope="session")
async def async_session() -> Generator:
test_engine = create_async_engine(
settings.SQLALCHEMY_DATABASE_URI + '_test',
echo=False,
pool_size=20, max_overflow=0
)
# expire_on_commit=False will prevent attributes from being expired
# after commit.
async_sess = sessionmaker(
test_engine, expire_on_commit=False, class_=AsyncSession
)
yield async_sess
@pytest.fixture(scope="session")
async def insert_initial_data(async_session:Callable):
async with async_session() as session, session.begin():
# insert first superuser - basic CRUD ops to insert data in test db
await insert_first_superuser(session)
# insert test.superuser@example.com
await insert_first_test_user(session)
# inserts test.user@example.com
@pytest.fixture(scope='session')
def app(insert_initial_data) -> FastAPI:
return FastAPI()
@pytest.fixture(scope='session')
async def client(app: FastAPI) -> Generator:
from app.api.deps import get_session
app.dependency_overrides[get_session] = get_test_session
async with AsyncClient(
app=app, base_url="http://test",
) as ac:
yield ac
# reset dependencies
app.dependency_overrides = {}
したがって、この場合、API テストの実行中に使用できるのはスーパーユーザーtest.superuser@example.comと通常ユーザーのみです。test.user@example.comたとえば、以下のコードはアクセス トークンを正常に取得できます。
async def authentication_token_from_email(
client: AsyncClient, session: AsyncSession,
) -> Dict[str, str]:
"""
Return a valid token for the user with given email.
"""
email = 'test.user@example.com'
password = 'test.user.password'
user = await crud.user.get_by_email(session, email=email)
assert user is not None
data = {"username": email, "password": password}
response = await client.post(f"{settings.API_V1_STR}/auth/access-token",
data=data)
auth_token = response.cookies.get('access_token')
assert auth_token is not None
return auth_token
ただし、以下の変更されたコードはそうではありません。ここでは、新しいユーザーを挿入してから、ログインしてアクセストークンを取得しようとしています。
async def authentication_token_from_email(
client: AsyncClient, session: AsyncSession,
) -> Dict[str, str]:
"""
Return a valid token for the user with given email.
If the user doesn't exist it is created first.
"""
email = random_email()
password = random_lower_string()
user = await crud.user.get_by_email(session, email=email)
if not user:
user_in_create = UserCreate(email=email,
password=password)
user = await crud.user.create(session, obj_in=user_in_create)
else:
user_in_update = UserUpdate(password=password)
user = await crud.user.update(session, db_obj=user, obj_in=user_in_update)
assert user is not None
# works fine up to this point, user inserted successfully
# now try to send http request to fetch token, and user is not found in the db
data = {"username": email, "password": password}
response = await client.post(f"{settings.API_V1_STR}/auth/access-token",
data=data)
auth_token = response.cookies.get('access_token')
# returns None.
return auth_token
ここで何が起こっているのですか？どんな助けにも感謝します！