I'm writing tests for my fastapi application that uses asynchronous posgtres connection:
# backend/database/session.py
from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
from config import config
engine = create_async_engine(
config.db_uri,
echo=config.db_echo,
# Connection pool configuration for scalability
# Number of connections to maintain in pool
pool_size=config.db_pool_size,
# Additional connections when pool is full
max_overflow=config.db_max_overflow,
# Validate connections before use
pool_pre_ping=True,
# Recycle connections after specified time
pool_recycle=config.db_pool_recycle,
# Timeout waiting for available connection
pool_timeout=config.db_pool_timeout,
# Reset connection state on return
pool_reset_on_return='commit',
# Performance optimizations
# Don't log pool operations (set to True for debugging)
echo_pool=False,
# Connection arguments
connect_args={
"ssl": config.db_ssl
}
)
async def init_db():
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
async def get_session() -> AsyncSession: # type: ignore
Session = sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False,
autocommit=False,
autoflush=False
)
async with Session() as session:
yield session
My app, db connection and api routes work fine. My tests however are not. To be more precise all tests that require a db_session do not work correctly.
I run my tests effectively in a python file via the subprocess module because I'm updating the env variables before each run to point to a different test database and run the migrations.
# backend/tests/run_tests.py
# <Update env and run alembic upgrade head>
# Run the integration tests
def run_tests(env):
print("Running unit tests...\n")
try:
# subprocess.run("ls")
subprocess.run("pytest -s --color=yes",
shell=True, check=True, text=True, env=env)
except subprocess.CalledProcessError as e:
print(f"Error when running tests: {e}")
pass
print("\nTests completed.")
# <Run alembic downgrade base>
Here is one test that requires the db session and will fail:
# backend/tests/user/test_signup.py
import pytest
from httpx import AsyncClient
from httpx._transports.asgi import ASGITransport
from main import app
@pytest.mark.asyncio
async def test_signup_successful():
"""Test user signup with valid data"""
# Use ASGITransport explicitly
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
# Define the request payload
payload = {
"first_name": "Test",
"last_name": "User",
"email": "integration_testuser@example.com",
"password": "Strongpassword123-"
}
# Perform POST request
response = await client.post("/user/signup", json=payload)
# Assertions
assert response.status_code == 201
data = response.json()
assert data["email"] == payload["email"]
This is the traceback I get (cut down to the last 40% of the actual traceback due to stack character limit):
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/core/user/helper.py", line 52, in _get_users
result = await session.exec(statement)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlmodel/ext/asyncio/session.py", line 81, in exec
result = await greenlet_spawn(
^^^^^^^^^^^^^^^^^^^^^
...<7 lines>...
)
^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 201, in greenlet_spawn
result = context.throw(*sys.exc_info())
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlmodel/orm/session.py", line 66, in exec
results = super().execute(
statement,
...<4 lines>...
_add_event=_add_event,
)
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/orm/session.py", line 2365, in execute
return self._execute_internal(
~~~~~~~~~~~~~~~~~~~~~~^
statement,
^^^^^^^^^^
...<4 lines>...
_add_event=_add_event,
^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/orm/session.py", line 2241, in _execute_internal
conn = self._connection_for_bind(bind)
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/orm/session.py", line 2110, in _connection_for_bind
return trans._connection_for_bind(engine, execution_options)
~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "<string>", line 2, in _connection_for_bind
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/orm/state_changes.py", line 137, in _go
ret_value = fn(self, *arg, **kw)
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/orm/session.py", line 1189, in _connection_for_bind
conn = bind.connect()
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/engine/base.py", line 3277, in connect
return self._connection_cls(self)
~~~~~~~~~~~~~~~~~~~~^^^^^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/engine/base.py", line 143, in __init__
self._dbapi_connection = engine.raw_connection()
~~~~~~~~~~~~~~~~~~~~~^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/engine/base.py", line 3301, in raw_connection
return self.pool.connect()
~~~~~~~~~~~~~~~~~^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/pool/base.py", line 447, in connect
return _ConnectionFairy._checkout(self)
~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/pool/base.py", line 1363, in _checkout
with util.safe_reraise():
~~~~~~~~~~~~~~~~~^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/util/langhelpers.py", line 224, in __exit__
raise exc_value.with_traceback(exc_tb)
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/pool/base.py", line 1301, in _checkout
result = pool._dialect._do_ping_w_event(
fairy.dbapi_connection
)
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/engine/default.py", line 728, in _do_ping_w_event
return self.do_ping(dbapi_connection)
~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 1169, in do_ping
dbapi_connection.ping()
~~~~~~~~~~~~~~~~~~~~~^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 813, in ping
self._handle_exception(error)
~~~~~~~~~~~~~~~~~~~~~~^^^^^^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 794, in _handle_exception
raise error
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 811, in ping
_ = self.await_(self._async_ping())
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 132, in await_only
return current.parent.switch(awaitable) # type: ignore[no-any-return,attr-defined] # noqa: E501
~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 196, in greenlet_spawn
value = await result
^^^^^^^^^^^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 820, in _async_ping
await tr.start()
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/asyncpg/transaction.py", line 146, in start
await self._connection.execute(query)
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/asyncpg/connection.py", line 349, in execute
result = await self._protocol.query(query, timeout)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "asyncpg/protocol/protocol.pyx", line 375, in query
RuntimeError: Task <Task pending name='starlette.middleware.base.BaseHTTPMiddleware.__call__.<locals>.call_next.<locals>.coro' coro=<BaseHTTPMiddleware.__call__.<locals>.call_next.<locals>.coro() running at /Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/starlette/middleware/base.py:144> cb=[TaskGroup._spawn.<locals>.task_done() at /Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/anyio/_backends/_asyncio.py:794]> got Future <Future pending cb=[BaseProtocol._on_waiter_completed()]> attached to a different loop
During handling of the above exception, another exception occurred:
session = <sqlalchemy.orm.session.AsyncSession object at 0x1100ebe00>
@pytest.mark.asyncio
async def test_signup_successful(session):
"""Test user signup with valid data"""
# Use ASGITransport explicitly
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
# Define the request payload
payload = {
"first_name": "Test",
"last_name": "User",
"email": "integration_testuser@example.com",
"password": "Strongpassword123-"
}
# Perform POST request
> response = await client.post("/user/signup", json=payload)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
tests/user/test_user_signup.py:35:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.venv/lib/python3.13/site-packages/httpx/_client.py:1859: in post
return await self.request(
.venv/lib/python3.13/site-packages/httpx/_client.py:1540: in request
return await self.send(request, auth=auth, follow_redirects=follow_redirects)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/httpx/_client.py:1629: in send
response = await self._send_handling_auth(
.venv/lib/python3.13/site-packages/httpx/_client.py:1657: in _send_handling_auth
response = await self._send_handling_redirects(
.venv/lib/python3.13/site-packages/httpx/_client.py:1694: in _send_handling_redirects
response = await self._send_single_request(request)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/httpx/_client.py:1730: in _send_single_request
response = await transport.handle_async_request(request)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/httpx/_transports/asgi.py:170: in handle_async_request
await self.app(scope, receive, send)
.venv/lib/python3.13/site-packages/fastapi/applications.py:1054: in __call__
await super().__call__(scope, receive, send)
.venv/lib/python3.13/site-packages/starlette/applications.py:113: in __call__
await self.middleware_stack(scope, receive, send)
.venv/lib/python3.13/site-packages/starlette/middleware/errors.py:186: in __call__
raise exc
.venv/lib/python3.13/site-packages/starlette/middleware/errors.py:164: in __call__
await self.app(scope, receive, _send)
.venv/lib/python3.13/site-packages/starlette/middleware/base.py:182: in __call__
with recv_stream, send_stream, collapse_excgroups():
^^^^^^^^^^^^^^^^^^^^
/opt/homebrew/Cellar/python@3.13/3.13.2/Frameworks/Python.framework/Versions/3.13/lib/python3.13/contextlib.py:162: in __exit__
self.gen.throw(value)
.venv/lib/python3.13/site-packages/starlette/_utils.py:83: in collapse_excgroups
raise exc
.venv/lib/python3.13/site-packages/starlette/middleware/base.py:184: in __call__
response = await self.dispatch_func(request, call_next)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
middleware.py:27: in execution_timer
response = await call_next(request)
^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/starlette/middleware/base.py:159: in call_next
raise app_exc
.venv/lib/python3.13/site-packages/starlette/middleware/base.py:144: in coro
await self.app(scope, receive_or_disconnect, send_no_error)
.venv/lib/python3.13/site-packages/starlette/middleware/trustedhost.py:36: in __call__
await self.app(scope, receive, send)
.venv/lib/python3.13/site-packages/starlette/middleware/cors.py:85: in __call__
await self.app(scope, receive, send)
.venv/lib/python3.13/site-packages/starlette/middleware/exceptions.py:63: in __call__
await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
.venv/lib/python3.13/site-packages/starlette/_exception_handler.py:53: in wrapped_app
raise exc
.venv/lib/python3.13/site-packages/starlette/_exception_handler.py:42: in wrapped_app
await app(scope, receive, sender)
.venv/lib/python3.13/site-packages/starlette/routing.py:716: in __call__
await self.middleware_stack(scope, receive, send)
.venv/lib/python3.13/site-packages/starlette/routing.py:736: in app
await route.handle(scope, receive, send)
.venv/lib/python3.13/site-packages/starlette/routing.py:290: in handle
await self.app(scope, receive, send)
.venv/lib/python3.13/site-packages/starlette/routing.py:78: in app
await wrap_app_handling_exceptions(app, request)(scope, receive, send)
.venv/lib/python3.13/site-packages/starlette/_exception_handler.py:53: in wrapped_app
raise exc
.venv/lib/python3.13/site-packages/starlette/_exception_handler.py:42: in wrapped_app
await app(scope, receive, sender)
.venv/lib/python3.13/site-packages/starlette/routing.py:75: in app
response = await f(request)
^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/fastapi/routing.py:302: in app
raw_response = await run_endpoint_function(
.venv/lib/python3.13/site-packages/fastapi/routing.py:213: in run_endpoint_function
return await dependant.call(**values)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/slowapi/extension.py:734: in async_wrapper
response = await func(*args, **kwargs) # type: ignore
^^^^^^^^^^^^^^^^^^^^^^^^^^^
api/user/router.py:50: in signup
user_exists = await service.user_exists(email=email, session=session)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
core/user/service.py:67: in user_exists
user = await self.get_user_by_email(email, session)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
core/user/service.py:39: in get_user_by_email
return await service_helper._get_users(session=session, where_clause=User.email == email, include_roles=include_roles, include_permissions=include_permissions)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
core/user/helper.py:52: in _get_users
result = await session.exec(statement)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/sqlmodel/ext/asyncio/session.py:81: in exec
result = await greenlet_spawn(
.venv/lib/python3.13/site-packages/sqlalchemy/util/_concurrency_py3k.py:201: in greenlet_spawn
result = context.throw(*sys.exc_info())
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/sqlmodel/orm/session.py:66: in exec
results = super().execute(
.venv/lib/python3.13/site-packages/sqlalchemy/orm/session.py:2365: in execute
return self._execute_internal(
.venv/lib/python3.13/site-packages/sqlalchemy/orm/session.py:2241: in _execute_internal
conn = self._connection_for_bind(bind)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/sqlalchemy/orm/session.py:2110: in _connection_for_bind
return trans._connection_for_bind(engine, execution_options)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
<string>:2: in _connection_for_bind
???
.venv/lib/python3.13/site-packages/sqlalchemy/orm/state_changes.py:137: in _go
ret_value = fn(self, *arg, **kw)
^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/sqlalchemy/orm/session.py:1189: in _connection_for_bind
conn = bind.connect()
^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/sqlalchemy/engine/base.py:3277: in connect
return self._connection_cls(self)
^^^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/sqlalchemy/engine/base.py:143: in __init__
self._dbapi_connection = engine.raw_connection()
^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/sqlalchemy/engine/base.py:3301: in raw_connection
return self.pool.connect()
^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/sqlalchemy/pool/base.py:447: in connect
return _ConnectionFairy._checkout(self)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/sqlalchemy/pool/base.py:1363: in _checkout
with util.safe_reraise():
^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/sqlalchemy/util/langhelpers.py:224: in __exit__
raise exc_value.with_traceback(exc_tb)
.venv/lib/python3.13/site-packages/sqlalchemy/pool/base.py:1301: in _checkout
result = pool._dialect._do_ping_w_event(
.venv/lib/python3.13/site-packages/sqlalchemy/engine/default.py:728: in _do_ping_w_event
return self.do_ping(dbapi_connection)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py:1169: in do_ping
dbapi_connection.ping()
.venv/lib/python3.13/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py:813: in ping
self._handle_exception(error)
.venv/lib/python3.13/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py:794: in _handle_exception
raise error
.venv/lib/python3.13/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py:811: in ping
_ = self.await_(self._async_ping())
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/sqlalchemy/util/_concurrency_py3k.py:132: in await_only
return current.parent.switch(awaitable) # type: ignore[no-any-return,attr-defined] # noqa: E501
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/sqlalchemy/util/_concurrency_py3k.py:196: in greenlet_spawn
value = await result
^^^^^^^^^^^^
.venv/lib/python3.13/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py:820: in _async_ping
await tr.start()
.venv/lib/python3.13/site-packages/asyncpg/transaction.py:146: in start
await self._connection.execute(query)
.venv/lib/python3.13/site-packages/asyncpg/connection.py:349: in execute
result = await self._protocol.query(query, timeout)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E RuntimeError: Task <Task pending name='starlette.middleware.base.BaseHTTPMiddleware.__call__.<locals>.call_next.<locals>.coro' coro=<BaseHTTPMiddleware.__call__.<locals>.call_next.<locals>.coro() running at /Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/starlette/middleware/base.py:144> cb=[TaskGroup._spawn.<locals>.task_done() at /Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/anyio/_backends/_asyncio.py:794]> got Future <Future pending cb=[BaseProtocol._on_waiter_completed()]> attached to a different loop
asyncpg/protocol/protocol.pyx:375: RuntimeError
--------------------------------------------------------------------- Captured log call ----------------------------------------------------------------------
ERROR sqlalchemy.pool.impl.AsyncAdaptedQueuePool:base.py:376 Exception terminating connection <AdaptedConnection <asyncpg.connection.Connection object at 0x1101dc500>>
Traceback (most recent call last):
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/pool/base.py", line 372, in _close_connection
self._dialect.do_terminate(connection)
~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 1136, in do_terminate
dbapi_connection.terminate()
~~~~~~~~~~~~~~~~~~~~~~~~~~^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 907, in terminate
self.await_(asyncio.shield(self._connection.close(timeout=2)))
~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 132, in await_only
return current.parent.switch(awaitable) # type: ignore[no-any-return,attr-defined] # noqa: E501
~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 196, in greenlet_spawn
value = await result
^^^^^^^^^^^^
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/asyncpg/connection.py", line 1504, in close
await self._protocol.close(timeout)
File "asyncpg/protocol/protocol.pyx", line 627, in close
File "asyncpg/protocol/protocol.pyx", line 660, in asyncpg.protocol.protocol.BaseProtocol._request_cancel
File "/Users/user/Documents/Programming/Python/Visual Studio Code/rag-sample/backend/.venv/lib/python3.13/site-packages/asyncpg/connection.py", line 1673, in _cancel_current_command
self._cancellations.add(self._loop.create_task(self._cancel(waiter)))
~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^
File "/opt/homebrew/Cellar/python@3.13/3.13.2/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/base_events.py", line 466, in create_task
self._check_closed()
~~~~~~~~~~~~~~~~~~^^
File "/opt/homebrew/Cellar/python@3.13/3.13.2/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/base_events.py", line 556, in _check_closed
raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
================================================================== short test summary info ===================================================================
FAILED tests/user/test_user_signup.py::test_signup_successful - RuntimeError: Task <Task pending name='starlette.middleware.base.BaseHTTPMiddleware.__call__.<locals>.call_next.<locals>.coro' coro=<BaseHTTPMiddleware._...
================================================================ 1 failed, 11 passed in 1.28s ================================================================
<sys>:0: RuntimeWarning: coroutine 'Connection._cancel' was never awaited
Error when running tests: Command 'pytest -s --color=yes' returned non-zero exit status 1.
Here is also the GitHub repo if you need more reference.
I also saw this GitHub issue but couldn't really make it work and I'm also not 100% sure if that is the exact same error but it seems likely.
I finally found a working solution myself thanks to wwfyde that nudged me into the right direction.
TLDR: The reason why I got this error is due to the fact that indiviual tests (at least with my config) can not share a single database connection pool or rather a single db engine unlike the FastAPI app instance. To fix this one method is to dispose the engine after every test route.
While just going with wwfyde's solution seems to work at first glance it is not as straightforward as you might think. You should never dispose the db engine in your get_session() method that is used by your backend application since this will absolutely kill performance and scalability. This will enforce a new db connection with every request creating a ton of overhead.
Instead you should create a dedicated test_db_session() method that includes the engine disposal and only use this one for your tests. You can then use this test_db_session() in your tests by overrding the FastAPI dependencies.
Long story short this is the updated session.py file:
# backend/database/session.py
from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
from config import config
engine = create_async_engine(
config.db_uri,
echo=config.db_echo,
# Connection pool configuration for scalability
# Number of connections to maintain in pool
pool_size=config.db_pool_size,
# Additional connections when pool is full
max_overflow=config.db_max_overflow,
# Validate connections before use
pool_pre_ping=True,
# Recycle connections after specified time
pool_recycle=config.db_pool_recycle,
# Timeout waiting for available connection
pool_timeout=config.db_pool_timeout,
# Reset connection state on return
pool_reset_on_return='commit',
# Performance optimizations
# Don't log pool operations (set to True for debugging)
echo_pool=False,
# Connection arguments
connect_args={
"ssl": config.db_ssl
}
)
async def init_db():
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
async def get_session() -> AsyncSession: # type: ignore
Session = sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False,
autocommit=False,
autoflush=False
)
async with Session() as session:
yield session
async def get_test_session() -> AsyncSession: # type: ignore
"""Get a database session specifically for tests.
This version disposes the engine after each session to prevent
asyncio loop conflicts in tests, but should NOT be used in production.
"""
Session = sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False,
autocommit=False,
autoflush=False
)
async with Session() as session:
yield session
# Only dispose in test environment to prevent loop conflicts
await engine.dispose()
Then you need to create a conftest.py file so that you only have to do this once to make it work for all test routes:
# backend/tests/conftest.py
import pytest
import pytest_asyncio
import httpx
from httpx._transports.asgi import ASGITransport
from main import app
from database.session import get_session, get_test_session
@pytest.fixture(scope="session", autouse=True)
def override_get_session():
"""Override the get_session dependency for all tests."""
# Set the override before any tests run
app.dependency_overrides[get_session] = get_test_session
yield
# Clean up after all tests
app.dependency_overrides.clear()
@pytest_asyncio.fixture
async def client():
"""HTTP client fixture that uses test-specific database session."""
transport = ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
yield client
@pytest_asyncio.fixture
async def db_session():
"""Direct database session fixture for tests that need direct DB access."""
async for session in get_test_session():
yield session
Now you can write your tests like this:
# backend/tests/user/test_user_signup.py
import pytest
import uuid
from sqlalchemy.sql import text
@pytest.mark.asyncio
async def test_signup_successful(client, db_session):
"""Test user signup with valid data"""
# Generate unique email for each test run
unique_email = f"test_user_{uuid.uuid4().hex[:8]}@example.com"
# Define the request payload
payload = {
"first_name": "Test",
"last_name": "User",
"email": unique_email,
"password": "Strongpassword123-"
}
# Perform POST request
response = await client.post("/user/signup", json=payload)
# Assertions
assert response.status_code == 201
data = response.json()
assert data["email"] == payload["email"]
assert data["success"]
# Verify the user exists in the database
statement = text(
f"SELECT email FROM users WHERE email = '{payload['email']}'")
result = await db_session.exec(statement)
user = result.scalar()
assert user is not None