fix: use NullPool for db engine under pytest to avoid cross-event-loop asyncpg errors
Task 3's conftest.py (per plan) reuses the module-level app.db.engine singleton across every test. pytest-asyncio 0.24 gives each test function its own event loop by default, and asyncpg connections are bound to the loop they were opened on. Pooling a connection from a prior test's loop made subsequent tests fail with "got Future attached to a different loop" as soon as more than one DB-touching test ran in the same session. NullPool is applied only when running under pytest (detected via sys.modules), so production keeps normal connection pooling and only the test suite pays the cost of a fresh connection per checkout.
This commit is contained in:
@@ -1,9 +1,23 @@
|
||||
import sys
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
from sqlalchemy.orm import DeclarativeBase
|
||||
from sqlalchemy.pool import NullPool
|
||||
|
||||
from app.config import settings
|
||||
|
||||
engine = create_async_engine(settings.database_url, echo=False)
|
||||
# pytest-asyncio gives each test function its own event loop by default, but
|
||||
# this engine (and its connection pool) is a module-level singleton created
|
||||
# once and reused across every test. asyncpg connections are bound to the
|
||||
# event loop they were opened on, so pooling them across per-test loops
|
||||
# raises "got Future attached to a different loop". NullPool sidesteps this
|
||||
# by opening a fresh connection per checkout instead of reusing one from a
|
||||
# prior loop. Only applied under pytest; production keeps normal pooling.
|
||||
_engine_kwargs = {"echo": False}
|
||||
if "pytest" in sys.modules:
|
||||
_engine_kwargs["poolclass"] = NullPool
|
||||
|
||||
engine = create_async_engine(settings.database_url, **_engine_kwargs)
|
||||
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user