英文:
What causes error 'async_generator' object has no attribute 'add'?
问题
我为我的FastAPI异步项目编写pytest测试。测试导致错误：
session.add(user)
AttributeError: 'async_generator'对象没有'add'属性
conftest.py：
SQLALCHEMY_DATABASE_URL = f'postgresql+asyncpg://{USER}:{PASSWORD}@{HOST}/{DB}_test'
engine = asyncio.create_async_engine(SQLALCHEMY_DATABASE_URL)
SessionTesting = sessionmaker(
bind=engine,
autocommit=False,
autoflush=False,
expire_on_commit=False,
class_=asyncio.AsyncSession
)
@pytest.fixture(name='session')
async def session_fixture() -> Generator[SessionTesting, None, None]:
Base.metadata.create_all(engine)
async with SessionTesting(bind=engine.connect()) as session:
yield session
Base.metadata.drop_all(engine)
test_auth.py：
@pytest.mark.asyncio
async def test_get_token(session: SessionTesting, create_user):
user = User(
first_name='fixture',
last_name='fixture',
username='fixture',
password=await get_hash('fixture')
)
session.add(user)
await session.commit()
async with AsyncClient(app=app, base_url='http://localhost:8000') as ac:
credentials = {
'username': 'fixture',
'password': 'fixture'
}
response = await ac.post('/token/get_token', data=credentials, headers={"content-type": "application/x-www-form-urlencoded"})
assert response.status_code == 200
有什么想法？
英文:
I'm writing pytest tests for my FastAPI async project. Tests causes error :
session.add(user)
AttributeError: 'async_generator' object has no attribute 'add'
conftest.py:
SQLALCHEMY_DATABASE_URL = f'postgresql+asyncpg://{USER}:{PASSWORD}@{HOST}/{DB}_test'
engine = asyncio.create_async_engine(SQLALCHEMY_DATABASE_URL)
SessionTesting = sessionmaker(
bind=engine,
autocommit=False,
autoflush=False,
expire_on_commit=False,
class_=asyncio.AsyncSession
)
@pytest.fixture(name='session')
async def session_fixture() -> Generator[SessionTesting, None, None]:
Base.metadata.create_all(engine)
async with SessionTesting(bind=engine.connect()) as session:
yield session
Base.metadata.drop_all(engine)
test_auth.py:
@pytest.mark.asyncio
async def test_get_token(session: SessionTesting, create_user):
# async with AsyncClient(app=app, base_url='http://test') as ac:
user = User(
first_name='fixture',
last_name='fixture',
username='fixture',
password=await get_hash('fixture')
)
session.add(user)
await session.commit()
async with AsyncClient(app=app, base_url='http://localhost:8000') as ac:
credentials = {
'username': 'fixture',
'password': 'fixture'
}
response = await ac.post('/token/get_token', data=credentials, headers={"content-type": "application/x-www-form-urlencoded"})
assert response.status_code == 200
Any ideas?
答案1
得分: 1
你正在在 session_fixture 内部产生 session（似乎也在之前某处定义了 fixture），但你没有将它作为参数传递给 session_fixture。这导致产生 async generator 而不是 session。
@pytest.fixture(name='session')
async def session_fixture(session):
...
应该改为：
@pytest.fixture(name='session')
async def session_fixture(session):
...
还有：
@pytest.mark.asyncio
async def test_get_token(session: SessionTesting, create_user):
应该改为：
@pytest.mark.asyncio
async def test_get_token(session_fixture: SessionTesting, create_user):
英文:
You are yielding session (which seems to also be a fixture defined somewhere above) inside session_fixture, but you are not passing it as an argument to session_fixture. That results in yielding async generator instead of session.
@pytest.fixture(name='session')
async def session_fixture():
...
should be
@pytest.fixture(name='session')
async def session_fixture(session):
...
and also
@pytest.mark.asyncio
async def test_get_token(session: SessionTesting, create_user):
should be
@pytest.mark.asyncio
async def test_get_token(session_fixture: SessionTesting, create_user):
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论