I need to test my fastapi app. In my project, I have crud file and views for my menus. In crud file I initialized async def for all crud operations, and each accepts as input session: AsyncSession. In views I initialized async def, which accepts as input session: AsyncSession = Depends(db_helper.scoped_session_dependency).
I have db_helper.py with code:
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,
)
In 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.get_scoped_session()
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
My tests looks like:
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
But I have error:
(.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 PASSED [100%]
========================================================================== FAILURES ===========================================================================
________________________________________________________________________ test_add_menu ________________________________________________________________________
async_client = <httpx.AsyncClient object at 0x7f70bc9c1090>
@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)
api_v1/menus/crud.py:48: in create_menu
await session.commit()
.venv/lib/python3.11/site-packages/sqlalchemy/ext/asyncio/scoping.py:471: in commit
return await self._proxied.commit()
.venv/lib/python3.11/site-packages/sqlalchemy/ext/asyncio/session.py:1011: in commit
await greenlet_spawn(self.sync_session.commit)
.venv/lib/python3.11/site-packages/sqlalchemy/util/_concurrency_py3k.py:200: in greenlet_spawn
result = context.throw(*sys.exc_info())
.venv/lib/python3.11/site-packages/sqlalchemy/orm/session.py:1969: in commit
trans.commit(_to_root=True)
<string>:2: in commit
???
.venv/lib/python3.11/site-packages/sqlalchemy/orm/state_changes.py:139: in _go
ret_value = fn(self, *arg, **kw)
.venv/lib/python3.11/site-packages/sqlalchemy/orm/session.py:1256: in commit
self._prepare_impl()
<string>:2: in _prepare_impl
???
.venv/lib/python3.11/site-packages/sqlalchemy/orm/state_changes.py:139: in _go
ret_value = fn(self, *arg, **kw)
.venv/lib/python3.11/site-packages/sqlalchemy/orm/session.py:1231: in _prepare_impl
self.session.flush()
.venv/lib/python3.11/site-packages/sqlalchemy/orm/session.py:4312: in flush
self._flush(objects)
.venv/lib/python3.11/site-packages/sqlalchemy/orm/session.py:4447: in _flush
with util.safe_reraise():
.venv/lib/python3.11/site-packages/sqlalchemy/util/langhelpers.py:146: in __exit__
raise exc_value.with_traceback(exc_tb)
.venv/lib/python3.11/site-packages/sqlalchemy/orm/session.py:4408: in _flush
flush_context.execute()
.venv/lib/python3.11/site-packages/sqlalchemy/orm/unitofwork.py:466: in execute
rec.execute(self)
.venv/lib/python3.11/site-packages/sqlalchemy/orm/unitofwork.py:642: in execute
util.preloaded.orm_persistence.save_obj(
.venv/lib/python3.11/site-packages/sqlalchemy/orm/persistence.py:93: in save_obj
_emit_insert_statements(
.venv/lib/python3.11/site-packages/sqlalchemy/orm/persistence.py:1227: in _emit_insert_statements
result = connection.execute(
.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py:1416: in execute
return meth(
.venv/lib/python3.11/site-packages/sqlalchemy/sql/elements.py:517: in _execute_on_connection
return connection._execute_clauseelement(
.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py:1639: in _execute_clauseelement
ret = self._execute_context(
.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py:1848: in _execute_context
return self._exec_single_context(
.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py:1988: in _exec_single_context
self._handle_dbapi_exception(
.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py:2347: in _handle_dbapi_exception
raise exc_info[1].with_traceback(exc_info[2])
.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py:1969: in _exec_single_context
self.dialect.do_execute(
.venv/lib/python3.11/site-packages/sqlalchemy/engine/default.py:922: in do_execute
cursor.execute(statement, parameters)
.venv/lib/python3.11/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py:580: in execute
self._adapt_connection.await_(
.venv/lib/python3.11/site-packages/sqlalchemy/util/_concurrency_py3k.py:130: in await_only
return current.driver.switch(awaitable) # type: ignore[no-any-return]
.venv/lib/python3.11/site-packages/sqlalchemy/util/_concurrency_py3k.py:195: in greenlet_spawn
value = await result
.venv/lib/python3.11/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py:516: in _prepare_and_execute
await adapt_connection._start_transaction()
.venv/lib/python3.11/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py:850: in _start_transaction
self._handle_exception(error)
.venv/lib/python3.11/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py:799: in _handle_exception
raise error
.venv/lib/python3.11/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py:848: in _start_transaction
await self._transaction.start()
.venv/lib/python3.11/site-packages/asyncpg/transaction.py:146: in start
await self._connection.execute(query)
.venv/lib/python3.11/site-packages/asyncpg/connection.py:350: in execute
result = await self._protocol.query(query, timeout)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E RuntimeError: Task <Task pending name='Task-4' coro=<test_add_menu() running at /home/ubuntu/Desktop/practice/menu_app_FastApi/tests/test_menu.py:9> cb=[_run_until_complete_cb() at /usr/lib/python3.11/asyncio/base_events.py:180]> got Future <Future pending cb=[Protocol._on_waiter_completed()]> attached to a different loop
asyncpg/protocol/protocol.pyx:374: RuntimeError
====================================================================== 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 - RuntimeError: Task <Task pending name='Task-4' coro=<test_add_menu() running at /home/ubuntu/Desktop/practice/menu_app_FastApi/tests/test_menu.py:9> cb=[_...
=========================================================== 1 failed, 1 passed, 1 warning in 0.46s ============================================================
Exception terminating connection <AdaptedConnection <asyncpg.connection.Connection object at 0x7f70bc97f1f0>>
Traceback (most recent call last):
File "/home/ubuntu/Desktop/practice/menu_app_FastApi/.venv/lib/python3.11/site-packages/sqlalchemy/pool/base.py", line 377, in _close_connection
self._dialect.do_terminate(connection)
File "/home/ubuntu/Desktop/practice/menu_app_FastApi/.venv/lib/python3.11/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 1109, in do_terminate
dbapi_connection.terminate()
File "/home/ubuntu/Desktop/practice/menu_app_FastApi/.venv/lib/python3.11/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 902, in terminate
self._connection.terminate()
File "/home/ubuntu/Desktop/practice/menu_app_FastApi/.venv/lib/python3.11/site-packages/asyncpg/connection.py", line 1478, in terminate
self._abort()
File "/home/ubuntu/Desktop/practice/menu_app_FastApi/.venv/lib/python3.11/site-packages/asyncpg/connection.py", line 1505, in _abort
self._protocol.abort()
File "asyncpg/protocol/protocol.pyx", line 607, in asyncpg.protocol.protocol.BaseProtocol.abort
File "/usr/lib/python3.11/asyncio/selector_events.py", line 821, in abort
self._force_close(None)
File "/usr/lib/python3.11/asyncio/selector_events.py", line 891, in _force_close
self._loop.call_soon(self._call_connection_lost, exc)
File "/usr/lib/python3.11/asyncio/base_events.py", line 761, in call_soon
self._check_closed()
File "/usr/lib/python3.11/asyncio/base_events.py", line 519, in _check_closed
raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
The garbage collector is trying to clean up non-checked-in connection <AdaptedConnection <asyncpg.connection.Connection object at 0x7f70bc97f1f0>>, which will be terminated. Please ensure that SQLAlchemy pooled connections are returned to the pool explicitly, either by calling ``close()`` or by using appropriate context managers to manage their lifecycle.
sys:1: SAWarning: The garbage collector is trying to clean up non-checked-in connection <AdaptedConnection <asyncpg.connection.Connection object at 0x7f70bc97f1f0>>, which will be terminated. Please ensure that SQLAlchemy pooled connections are returned to the pool explicitly, either by calling ``close()`` or by using appropriate context managers to manage their lifecycle.
I edit my code and this message. Get - working, but I have new error with post.
Using @pytest_asyncio.fixture() not required, because I set asyncio_mode = auto
try instead of
next code