我需要测试我的 fastapi 应用程序（https://github.com/DimTur/menu_app_FastApi）。在我的项目中，我有菜单的 CRUD 文件和视图。在 crud 文件中，我为所有 crud 操作初始化了 async def，并且每个操作都接受输入会话：AsyncSession。在视图中，我初始化了 async def，它接受输入会话：AsyncSession = Depends(db_helper.scoped_session_dependency)。 我有 db_helper.py 和代码：
from asyncio import current_task
from contextlib import asynccontextmanager
from typing import AsyncIterator
from sqlalchemy.ext.asyncio import (
AsyncSession,
create_async_engine,
async_sessionmaker,
async_scoped_session,
AsyncConnection,
)
from core.config import settings
from core.models import Base
class DatabaseHelper:
def __init__(self, url: str, echo: bool = False):
self.engine = create_async_engine(
url=url,
echo=echo,
)
self.session_factory = async_sessionmaker(
bind=self.engine,
autoflush=False,
autocommit=False,
expire_on_commit=False,
)
def get_scoped_session(self):
session = async_scoped_session(
session_factory=self.session_factory,
scopefunc=current_task,
)
return session
async def session_dependency(self) -> AsyncSession:
async with self.session_factory() as session:
yield session
await session.close()
async def scoped_session_dependency(self) -> AsyncSession:
session = self.get_scoped_session()
yield session
await session.close()
@asynccontextmanager
async def connect(self) -> AsyncIterator[AsyncConnection]:
async with self.engine.begin() as connection:
yield connection
async def create_all(self, connection: AsyncConnection):
await connection.run_sync(Base.metadata.create_all)
async def drop_all(self, connection: AsyncConnection):
await connection.run_sync(Base.metadata.drop_all)
db_helper = DatabaseHelper(
url=settings.db.url,
echo=settings.db.echo,
)
在confest.py中：
import asyncio
from typing import AsyncGenerator
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
from core.models.db_helper import db_helper
from main import app
@pytest.fixture(scope="session", autouse=True)
async def prepare_database():
async with db_helper.connect() as conn:
await db_helper.drop_all(conn)
await db_helper.create_all(conn)
async def override_scoped_session_dependency() -> AsyncSession:
session = db_helper.scoped_session_dependency()
yield session
await session.close()
app.dependency_overrides[
db_helper.scoped_session_dependency
] = override_scoped_session_dependency
@pytest.fixture(scope="session")
def event_loop(request):
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="session")
async def async_client() -> AsyncGenerator[AsyncClient, None]:
async with AsyncClient(app=app, base_url="http://test") as client:
yield client
我的测试看起来像：
import pytest
from httpx import AsyncClient
from conftest import async_client
@pytest.mark.asyncio
async def test_add_menu(async_client: AsyncClient):
response = await async_client.post(
"/api/v1/menus/",
json={
"title": "menu2",
"description": "1111111111111111111",
},
)
assert response.status_code == 201
@pytest.mark.asyncio
async def test_get_menus(async_client: AsyncClient):
response = await async_client.get(
"/api/v1/menus/",
)
assert response.status_code == 200
但是我有错误：
(.venv) ubuntu@ubuntu:~/Desktop/practice/menu_app_FastApi$ pytest -v tests/
===================================================================== test session starts =====================================================================
platform linux -- Python 3.11.6, pytest-7.4.4, pluggy-1.3.0 -- /home/ubuntu/Desktop/practice/menu_app_FastApi/.venv/bin/python
cachedir: .pytest_cache
rootdir: /home/ubuntu/Desktop/practice/menu_app_FastApi
configfile: pytest.ini
plugins: asyncio-0.23.3, anyio-4.2.0, dotenv-0.5.2
asyncio: mode=Mode.AUTO
collected 2 items
tests/test_menu.py::test_add_menu FAILED [ 50%]
tests/test_menu.py::test_get_menus FAILED [100%]
========================================================================== FAILURES ===========================================================================
________________________________________________________________________ test_add_menu ________________________________________________________________________
async_client = <httpx.AsyncClient object at 0x7ff073695310>
@pytest.mark.asyncio
async def test_add_menu(async_client: AsyncClient):
> response = await async_client.post(
"/api/v1/menus/",
json={
"title": "menu2",
"description": "1111111111111111111",
},
)
tests/test_menu.py:9:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.venv/lib/python3.11/site-packages/httpx/_client.py:1877: in post
return await self.request(
.venv/lib/python3.11/site-packages/httpx/_client.py:1559: in request
return await self.send(request, auth=auth, follow_redirects=follow_redirects)
.venv/lib/python3.11/site-packages/httpx/_client.py:1646: in send
response = await self._send_handling_auth(
.venv/lib/python3.11/site-packages/httpx/_client.py:1674: in _send_handling_auth
response = await self._send_handling_redirects(
.venv/lib/python3.11/site-packages/httpx/_client.py:1711: in _send_handling_redirects
response = await self._send_single_request(request)
.venv/lib/python3.11/site-packages/httpx/_client.py:1748: in _send_single_request
response = await transport.handle_async_request(request)
.venv/lib/python3.11/site-packages/httpx/_transports/asgi.py:162: in handle_async_request
await self.app(scope, receive, send)
.venv/lib/python3.11/site-packages/fastapi/applications.py:1054: in __call__
await super().__call__(scope, receive, send)
.venv/lib/python3.11/site-packages/starlette/applications.py:123: in __call__
await self.middleware_stack(scope, receive, send)
.venv/lib/python3.11/site-packages/starlette/middleware/errors.py:186: in __call__
raise exc
.venv/lib/python3.11/site-packages/starlette/middleware/errors.py:164: in __call__
await self.app(scope, receive, _send)
.venv/lib/python3.11/site-packages/starlette/middleware/exceptions.py:62: in __call__
await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
.venv/lib/python3.11/site-packages/starlette/_exception_handler.py:64: in wrapped_app
raise exc
.venv/lib/python3.11/site-packages/starlette/_exception_handler.py:53: in wrapped_app
await app(scope, receive, sender)
.venv/lib/python3.11/site-packages/starlette/routing.py:762: in __call__
await self.middleware_stack(scope, receive, send)
.venv/lib/python3.11/site-packages/starlette/routing.py:782: in app
await route.handle(scope, receive, send)
.venv/lib/python3.11/site-packages/starlette/routing.py:297: in handle
await self.app(scope, receive, send)
.venv/lib/python3.11/site-packages/starlette/routing.py:77: in app
await wrap_app_handling_exceptions(app, request)(scope, receive, send)
.venv/lib/python3.11/site-packages/starlette/_exception_handler.py:64: in wrapped_app
raise exc
.venv/lib/python3.11/site-packages/starlette/_exception_handler.py:53: in wrapped_app
await app(scope, receive, sender)
.venv/lib/python3.11/site-packages/starlette/routing.py:72: in app
response = await func(request)
.venv/lib/python3.11/site-packages/fastapi/routing.py:299: in app
raise e
.venv/lib/python3.11/site-packages/fastapi/routing.py:294: in app
raw_response = await run_endpoint_function(
.venv/lib/python3.11/site-packages/fastapi/routing.py:191: in run_endpoint_function
return await dependant.call(**values)
api_v1/menus/views.py:32: in create_menu
return await crud.create_menu(session=session, menu_in=menu_in)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
session = <async_generator object DatabaseHelper.scoped_session_dependency at 0x7ff07368c2b0>
menu_in = MenuCreate(title='menu2', description='1111111111111111111')
async def create_menu(session: AsyncSession, menu_in: MenuCreate) -> Menu:
menu = Menu(**menu_in.model_dump())
> session.add(menu)
E AttributeError: 'async_generator' object has no attribute 'add'
api_v1/menus/crud.py:47: AttributeError
_______________________________________________________________________ test_get_menus ________________________________________________________________________
async_client = <httpx.AsyncClient object at 0x7ff073695310>
@pytest.mark.asyncio
async def test_get_menus(async_client: AsyncClient):
> response = await async_client.get(
"/api/v1/menus/",
)
tests/test_menu.py:22:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.venv/lib/python3.11/site-packages/httpx/_client.py:1786: in get
return await self.request(
.venv/lib/python3.11/site-packages/httpx/_client.py:1559: in request
return await self.send(request, auth=auth, follow_redirects=follow_redirects)
.venv/lib/python3.11/site-packages/httpx/_client.py:1646: in send
response = await self._send_handling_auth(
.venv/lib/python3.11/site-packages/httpx/_client.py:1674: in _send_handling_auth
response = await self._send_handling_redirects(
.venv/lib/python3.11/site-packages/httpx/_client.py:1711: in _send_handling_redirects
response = await self._send_single_request(request)
.venv/lib/python3.11/site-packages/httpx/_client.py:1748: in _send_single_request
response = await transport.handle_async_request(request)
.venv/lib/python3.11/site-packages/httpx/_transports/asgi.py:162: in handle_async_request
await self.app(scope, receive, send)
.venv/lib/python3.11/site-packages/fastapi/applications.py:1054: in __call__
await super().__call__(scope, receive, send)
.venv/lib/python3.11/site-packages/starlette/applications.py:123: in __call__
await self.middleware_stack(scope, receive, send)
.venv/lib/python3.11/site-packages/starlette/middleware/errors.py:186: in __call__
raise exc
.venv/lib/python3.11/site-packages/starlette/middleware/errors.py:164: in __call__
await self.app(scope, receive, _send)
.venv/lib/python3.11/site-packages/starlette/middleware/exceptions.py:62: in __call__
await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
.venv/lib/python3.11/site-packages/starlette/_exception_handler.py:64: in wrapped_app
raise exc
.venv/lib/python3.11/site-packages/starlette/_exception_handler.py:53: in wrapped_app
await app(scope, receive, sender)
.venv/lib/python3.11/site-packages/starlette/routing.py:762: in __call__
await self.middleware_stack(scope, receive, send)
.venv/lib/python3.11/site-packages/starlette/routing.py:782: in app
await route.handle(scope, receive, send)
.venv/lib/python3.11/site-packages/starlette/routing.py:297: in handle
await self.app(scope, receive, send)
.venv/lib/python3.11/site-packages/starlette/routing.py:77: in app
await wrap_app_handling_exceptions(app, request)(scope, receive, send)
.venv/lib/python3.11/site-packages/starlette/_exception_handler.py:64: in wrapped_app
raise exc
.venv/lib/python3.11/site-packages/starlette/_exception_handler.py:53: in wrapped_app
await app(scope, receive, sender)
.venv/lib/python3.11/site-packages/starlette/routing.py:72: in app
response = await func(request)
.venv/lib/python3.11/site-packages/fastapi/routing.py:299: in app
raise e
.venv/lib/python3.11/site-packages/fastapi/routing.py:294: in app
raw_response = await run_endpoint_function(
.venv/lib/python3.11/site-packages/fastapi/routing.py:191: in run_endpoint_function
return await dependant.call(**values)
api_v1/menus/views.py:20: in get_menus
return await crud.get_menus(session=session)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
session = <async_generator object DatabaseHelper.scoped_session_dependency at 0x7ff072766400>
async def get_menus(session: AsyncSession) -> list[Menu]:
stmt = (
select(Menu)
.options(selectinload(Menu.submenus).selectinload(Submenu.dishes))
.order_by(Menu.title)
)
> result: Result = await session.execute(stmt)
E AttributeError: 'async_generator' object has no attribute 'execute'
api_v1/menus/crud.py:19: AttributeError
====================================================================== warnings summary =======================================================================
tests/test_menu.py::test_add_menu
/home/ubuntu/Desktop/practice/menu_app_FastApi/.venv/lib/python3.11/site-packages/pytest_asyncio/plugin.py:749: DeprecationWarning: The event_loop fixture provided by pytest-asyncio has been redefined in
/home/ubuntu/Desktop/practice/menu_app_FastApi/tests/conftest.py:31
Replacing the event_loop fixture with a custom implementation is deprecated
and will lead to errors in the future.
If you want to request an asyncio event loop with a scope other than function
scope, use the "scope" argument to the asyncio mark when marking the tests.
If you want to return different types of event loops, use the event_loop_policy
fixture.
warnings.warn(
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
=================================================================== short test summary info ===================================================================
FAILED tests/test_menu.py::test_add_menu - AttributeError: 'async_generator' object has no attribute 'add'
FAILED tests/test_menu.py::test_get_menus - AttributeError: 'async_generator' object has no attribute 'execute'
================================================================ 2 failed, 1 warning in 0.28s =================================================================
不需要使用 @pytest_asyncio.fixture() ，因为我设置了 asyncio_mode = auto
尝试代替
session = db_helper.scoped_session_dependency()
yield session
下一个代码
async for session in db_helper.scoped_session_dependency():
yield session
break