15

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.

sqlalchemy_session_fail

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?
Improve this question

2 Answers 2

Reset to default
19

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.

Improve this answer
Sign up to request clarification or add additional context in comments.

6 Comments

For people using sqlite and want to use this stackoverflow.com/a/36457404/154607
I've been using a more naive session fixture and dancing around commits in code under test to avoid committing the outer transaction. This answer just vaporized five years of frustration.
@TravisMehlinger I danced for a year or two before as well. For many times I thought I read the SA docs, but still didn't know how to do things..
For anyone else puzzled, like I was, there should be a import sqlalchemy as sa before this function.
Thanks a million for this. Lifesaver. In our setup there was one test failing with sqlalchemy.exc.ResourceClosedError: This Connection is closed. This was solved by saving the original db.session in a temp var, and restoring it at the end of this function. I'm not sure what happened (it was an auth wrapper trying to access the user DB, so perhaps it ran before / after the test wrapper?)
|
7

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()
Improve this answer

6 Comments

Regarding point number 1: here is my quandry; I want to test that User.id's primary_key is written correctly and auto_increments as I expect. However, I can't really do that unless I session.commit(), right? Apologies if this is a dumb question, I'm a bit new to SQLAlchemy
@MikePennington, Sorry, I don't understand your comment. Here's the forked version: gist.github.com/jeong-min-lee/28fd88babe2808afe9ba (only conftest.py is modified) and the test result screenshot:i.imgur.com/LxZnsvw.png
@MikePennington, To be clear, I didn't mean to remove a call to the session.commit.
Thank you, I was trying to avoid scope=function because I didn't want to deal with the time to re-create the db, but an in-memory DB is pretty fast. I am developing in sqlite and will probably deploy on postgres or oracle. An in-memory test for other DBs (such as Oracle) probably won't work but this is good enough
@falsetru about the rollback not taking effect, please also see docs.sqlalchemy.org/en/latest/orm/…
|

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.