Isolating DB.test DB Sessions in Flask-SQLAlchemy

Question

I am trying to create a Flask application with Flask-SQLAlchemy; I use pytest to test the db. One of the problems is to create separate database sessions between different tests.

I have prepared a minimal, complete example to highlight the problem, note that test_user_schema1() and test_user_schema2() same.

File name: 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 

If db is really isolated between my tests, both tests must pass. However, the last test always fails, because I did not find a way to do rollbacks of db sessions correctly.

conftest.py uses the following, based on what I saw in Alex Michael's blog post , but this device code breaks because it seems t to isolate db sessions between devices.

 @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() 

For the purposes of this question, I built a gist that contains everything you need to reproduce; you can clone it with git clone https://gist.github.com/34fa8d274fc4be240933.git .

I use the following packages ...

 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 

Two questions:

• Why is the status quo violated? This same py.test tool seemed to work for someone else.
• How can I fix this to work correctly?

Answer 1

1.

According to Session Basics - SQLAlchemy documentation :

commit() used to commit the current transaction. It always issues flush () in advance to clear any remaining state to the database; it does not depend on the setting of "autorun" .....

So transaction.rollback() in the function of the session device does not take effect, because the transaction has already been completed.

---

2.

Change the fixture area to function instead of session so that db is cleared every time.

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

---

BTW. If you use the sqlite database in memory, you do not need to delete the db files, and it will be faster:

 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() 

Answer 2

The method introduced in the Alex Michael blog post does not work because it is incomplete. According to the sqlalchemy documentation for joining sessions , the Alex solution only works if there is no rollback rollback. Another difference is that sqla docs uses the vanilla Session object compared to the session on the Alex blog.

In the case of flask-sqlalchemy, a session with a limited validity period is automatically deleted at the request of teardown . A session.remove call is made, which returns a rollback under the hood. To support rollbacks as part of tests, use SAVEPOINT :

 @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, :coolguy:. """ 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() 

Your database should support SAVEPOINT though.