Estoy tratando de imitar el comportamiento de Django cuando ejecuto pruebas en FastAPI: quiero crear una base de datos de prueba al comienzo de cada prueba y destruirla al final. El problema es que la naturaleza asíncrona de FastAPI está rompiendo todo. Cuando hice una verificación de cordura y puse todo en sincronía, todo funcionó a la perfección. Sin embargo, cuando trato de ejecutar las cosas de forma asíncrona, todo se rompe. Esto es lo que tengo en este momento:
El accesorio:
@pytest.fixture(scope="session") def event_loop(): return asyncio.get_event_loop() @pytest.fixture(scope="session") async def session(): sync_test_db = "postgresql://postgres:postgres@postgres:5432/test" if not database_exists(sync_test_db): create_database(sync_test_db) async_test_db = "postgresql+asyncpg://postgres:postgres@postgres:5432/test" engine = create_async_engine(url=async_test_db, echo=True, future=True) async with engine.begin() as conn: await conn.run_sync(SQLModel.metadata.create_all) Session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) async with Session() as session: def get_session_override(): return session app.dependency_overrides[get_session] = get_session_override yield session drop_database(sync_test_db)La prueba:
class TestSomething: @pytest.mark.asyncio async def test_create_something(self, session): data = {"some": "data"} response = client.post( "/", json=data ) assert response.ok results = await session.execute(select(Something)) # <- This line fails assert len(results.all()) == 1El error:
E sqlalchemy.exc.PendingRollbackError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: Task <Task pending name='anyio.from_thread.BlockingPortal._call_func' coro=<BlockingPortal._call_func() running at /usr/local/lib/python3.9/site-packages/anyio/from_thread.py:187> cb=[TaskGroup._spawn.<locals>.task_done() at /usr/local/lib/python3.9/site-packages/anyio/_backends/_asyncio.py:629]> got Future <Future pending cb=[Protocol._on_waiter_completed()]> attached to a different loop (Background on this error at: https://sqlalche.me/e/14/7s2a) /usr/local/lib/python3.9/site-packages/sqlalchemy/orm/session.py:601: PendingRollbackError¿Alguna idea de lo que podría estar haciendo mal?
Verifique si otras declaraciones en sus casos de prueba que involucran la base de datos pueden fallar antes de que se genere este error.
Para mí, el PendingRollbackError fue causado por un InsertionError que se generó en una prueba anterior.
Todas mis pruebas fueron pruebas unitarias (asincrónicas) que involucraron inserciones de bases de datos en una base de datos postgres. Después de las pruebas, se suponía que la sesión de la base de datos haría una reversión de sus entradas.
El error de InsertionError fue causado por inserciones en la base de datos que fallaron en una restricción única. Todas las pruebas posteriores generaron PendingRollbackError .