使用httpx的pytest.AsyncClient找不到新创建的数据库记录

Question

我正在尝试使用httpx.AsyncClient设置pytest，并使用FastAPI设置sqlalchemy AsyncSession。除了异步的东西之外，所有的东西实际上都模仿了FastAPI Fullstack repo中的测试。

CRUD单元测试没有问题。从httpx-lib使用AsyncClient运行API测试时会出现此问题。

问题是，客户端发出的任何请求都只能访问在初始化(设置(客户端设备之前创建的用户(在我的情况下(。

我的pytestconftest.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 [e-mail]
await insert_first_test_user(session)
# inserts [e-mail]
@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测试期间，只有超级用户[e-mail]和普通用户[e-mail]可用。例如，下面的代码可以很好地获取访问令牌：

async def authentication_token_from_email(
client: AsyncClient,  session: AsyncSession,
) -> Dict[str, str]:
"""
Return a valid token for the user with given email.

"""

email = '[e-mail]'
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

这是怎么回事？感谢您的帮助！

Answer 1

事实证明，由于我不理解的原因，我所需要做的就是在客户端设备中定义FastAPI依赖重写函数：

之前

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 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 = {}

之后

@pytest.fixture(scope="function")
async def session(async_session) -> Generator:
async with async_session() as sess, sess.begin():
yield sess


@pytest.fixture
async def client(app: FastAPI, session:AsyncSession) -> Generator:
from app.api.deps import get_session

# this needs to be defined inside this fixture
# this is generate that yields session retrieved from `session` fixture

def get_sess(): 
yield session

app.dependency_overrides[get_session] =  get_sess

async with AsyncClient(
app=app, base_url="http://test", 
) as ac:
yield ac
app.dependency_overrides = {}

我很感激对这种行为的任何解释。谢谢