Isolating py.test DB sessions in Flask-SQLAlchemy

Question

I'm trying to build a Flask app with Flask-SQLAlchemy; I use pytest to test the DB. One of the problems seems to be creating isolated DB sessions between different tests.

I cooked up a minimal, complete example to highlight the problem, note that test_user_schema1() and test_user_schema2() are the same.

Filename: 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 the db is truly isolated between my tests, both tests should pass. However, the last test always fails, because I haven't found a way to make db sessions rollback correctly.

conftest.py uses the following based on what I saw in Alex Michael's blog post, but this fixture code breaks because it apparently doesn't isolate the db sessions between fixtures.

@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, which contains all you need to reproduce it; you can clone it with git clone https://gist.github.com/34fa8d274fc4be240933.git.

I am using 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:

1. Why is status quo broken? This same py.test fixture seemed to work for someone else.
2. How can I fix this to work correctly?

Answer 1

The method introduced in Alex Michael's blog post is not working because it's incomplete. According to the sqlalchemy documentation on joining sessions, Alex's solution works only if there are no rollback calls. Another difference is, a vanilla Session object is used in sqla docs, compared to a scoped session on Alex's blog.

In the case of flask-sqlalchemy, the scoped session is automatically removed on request teardown. A call to session.remove is made, which issues a rollback under the hood. To support rollbacks within the scope of the tests, 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 restart_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 must support SAVEPOINT though.

Answer 2

1.

According to Session Basics - SQLAlchemy documentation:

commit() is used to commit the current transaction. It always issues flush() beforehand to flush any remaining state to the database; this is independent of the “autoflush” setting. ....

So transaction.rollback() in session fixture function does not take effect, because the transaction is already committed.

---

2.

Change scope of fixtures 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 in-memory sqlite database, you don't 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()