Aislando sesiones de base de datos py.test en Flask-SQLAlchemy

Estoy tratando de construir una aplicación Flask con Flask-SQLAlchemy; Yo uso pytest para probar el DB. Uno de los problemas parece ser la creación de sesiones de base de datos aisladas entre diferentes pruebas.

test_user_schema1() un ejemplo mínimo y completo para resaltar el problema, tenga en cuenta que test_user_schema1() y test_user_schema2() son iguales.

Nombre de archivo: test_db.py

 from models import User def test_user_schema1(session): person_name = 'Fran Clan' uu = User(name=person_name) session.add(uu) session.commit() assert uu.id==1 assert uu.name==person_name def test_user_schema2(session): person_name = 'Stan Clan' uu = User(name=person_name) session.add(uu) session.commit() assert uu.id==1 assert uu.name==person_name 

Si el db está realmente aislado entre mis pruebas, ambas pruebas deberían pasar. Sin embargo, la última prueba siempre falla, porque no he encontrado una manera de hacer que las sesiones db se deshagan correctamente.

sqlalchemy_session_fail

conftest.py usa lo siguiente según lo que vi en la publicación del blog de Alex Michael , pero este código de accesorio se rompe porque aparentemente no aísla las sesiones de db entre los accesorios.

 @pytest.yield_fixture(scope='function') def session(app, db): connection = db.engine.connect() transaction = connection.begin() #options = dict(bind=connection, binds={}) options = dict(bind=connection) session = db.create_scoped_session(options=options) yield session # Finalize test here transaction.rollback() connection.close() session.remove() 

Para los propósitos de esta pregunta, construí un gist , que contiene todo lo que necesitas para reproducirlo; puede clonarlo con git clone https://gist.github.com/34fa8d274fc4be240933.git .

Estoy usando los siguientes paquetes …

 Flask==0.10.1 Flask-Bootstrap==3.3.0.1 Flask-Migrate==1.3.0 Flask-Moment==0.4.0 Flask-RESTful==0.3.1 Flask-Script==2.0.5 Flask-SQLAlchemy==2.0 Flask-WTF==0.11 itsdangerous==0.24 pytest==2.6.4 Werkzeug==0.10.1 

Dos preguntas:

  1. ¿Por qué se rompe el status quo? Este mismo accesorio de py.test parecía funcionar para otra persona.
  2. ¿Cómo puedo arreglar esto para que funcione correctamente?

1.

De acuerdo con Session Basics – documentación de SQLAlchemy :

Se usa commit() para confirmar la transacción actual. Siempre emite flush () de antemano para vaciar cualquier estado restante en la base de datos; esto es independiente de la configuración de “autoflush”. ….

Por lo tanto, transaction.rollback() en la función de fijación de sesión no tiene efecto, porque la transacción ya está confirmada.


2.

Cambie el scope de los aparatos para que function lugar de la session modo que la db se borre cada vez.

 @pytest.yield_fixture(scope='function') def app(request): ... @pytest.yield_fixture(scope='function') def db(app, request): ... 

Por cierto, si usa una base de datos sqlite en memoria, no necesita eliminar los archivos db, y será más rápido:

 DB_URI = 'sqlite://' # SQLite :memory: database ... @pytest.yield_fixture(scope='function') def db(app, request): _db.app = app _db.create_all() yield _db _db.drop_all() 

El método introducido en la publicación del blog de Alex Michael no funciona porque está incompleto. De acuerdo con la documentación de sqlalchemy sobre cómo unirse a las sesiones , la solución de Alex funciona solo si no hay llamadas de reversión. Otra diferencia es que un objeto de Session vainilla se utiliza en documentos de sqla, en comparación con una sesión con ámbito en el blog de Alex.

En el caso de flask-sqlalchemy, la sesión con ámbito se elimina automáticamente en el desassembly de la solicitud . Se realiza una llamada a session.remove , que emite una reversión bajo el capó. Para admitir reversiones dentro del scope de las pruebas, use SAVEPOINT :

 import sqlalchemy as sa @pytest.yield_fixture(scope='function') def db_session(db): """ Creates a new database session for a test. Note you must use this fixture if your test connects to db. Here we not only support commit calls but also rollback calls in tests. """ connection = db.engine.connect() transaction = connection.begin() options = dict(bind=connection, binds={}) session = db.create_scoped_session(options=options) session.begin_nested() # session is actually a scoped_session # for the `after_transaction_end` event, we need a session instance to # listen for, hence the `session()` call @sa.event.listens_for(session(), 'after_transaction_end') def resetart_savepoint(sess, trans): if trans.nested and not trans._parent.nested: session.expire_all() session.begin_nested() db.session = session yield session session.remove() transaction.rollback() connection.close() 

Su base de datos debe soportar SAVEPOINT sin embargo.