I am building an app using FastAPI and SQLAlchemy. Async setup. For some reason I cannot find a proper way to setup async tests. I am running a very basic test setup.
conftest.py:
@pytest_asyncio.fixture(scope='function')
async def client() -> AsyncGenerator[AsyncClient, None]:
async with AsyncClient(app=app, base_url="http://test") as ac:
yield ac
test_event.py:
@pytest.mark.asyncio
async def test_get_public_events(client: AsyncClient) -> None:
response = await client.get("/api/v1/events")
assert response.status_code == 200
The interesting thing is that tests only work if I set
poolclass=NullPool
on my async_engine.
db.py:
async_engine = create_async_engine(
url=settings.POSTGRES_URL.get_secret_value(),
poolclass=NullPool, # TODO: Research this, tests are failing if this is not set
pool_pre_ping=True,
echo=True, # TODO: Handle this in a better way
)
async_session = async_sessionmaker(
bind=async_engine,
autoflush=False,
expire_on_commit=False,
class_=AsyncSession,
)
If I dont then I get an error when running tests:
RuntimeError: Task <Task pending name='Task-4'
coro=<test_get_public_events() running at ....
cb=[_run_until_complete_cb() at ...... got
Future <Future pending> attached to a different loop
Versions:
fastapi = "^0.110.0"
sqlalchemy = {extras = ["asyncio"], version = "^2.0.28"}
asyncpg = "^0.29.0"
pytest = "^8.1.1"
Maybe there is someone more knowledgable about the issues and can explain what is going on here?
Thanks
Managed to fix the issue by adding await async_engine.dispose()
to fixture which prepares my database for testing:
@pytest_asyncio.fixture(scope="session", autouse=True)
async def prepare_database() -> AsyncIterator[None]:
# Creates tables
async with async_engine.begin() as conn:
await conn.run_sync(CustomBaseModel.metadata.create_all)
# Creates test user
async with async_session_factory.begin() as session:
session.add(UserModel(...))
await session.commit()
await async_engine.dispose() # Added this line
yield