Flask + SQLAlchemy + pytest - not rolling back my session

Question

There are several similar questions on stack overflow, and I apologize in advance if I'm breaking etiquette by asking another one, but I just cannot seem to come up with the proper set of incantations to make this work.

I'm trying to use Flask + Flask-SQLAlchemy and then use pytest to manage the session such that when the function-scoped pytest fixture is torn down, the current transation is rolled back.

Some of the other questions seem to advocate using the db "drop all and create all" pytest fixture at the function scope, but I'm trying to use the joined session, and use rollbacks, since I have a LOT of tests. This would speed it up considerably.

http://alexmic.net/flask-sqlalchemy-pytest/ is where I found the original idea, and Isolating py.test DB sessions in Flask-SQLAlchemy is one of the questions recommending using function-level db re-creation.

I had also seen https://github.com/mitsuhiko/flask-sqlalchemy/pull/249 , but that appears to have been released with flask-sqlalchemy 2.1 (which I am using).

My current (very small, hopefully immediately understandable) repo is here: https://github.com/hoopes/flask-pytest-example

There are two print statements - the first (in example/__init__.py) should have an Account object, and the second (in test/conftest.py) is where I expect the db to be cleared out after the transaction is rolled back.

If you pip install -r requirements.txt and run py.test -s from the test directory, you should see the two print statements.

I'm about at the end of my rope here - there must be something I'm missing, but for the life of me, I just can't seem to find it.

Help me, SO, you're my only hope!

Answer 1

You might want to give pytest-flask-sqlalchemy-transactions a try. It's a plugin that exposes a db_session fixture that accomplishes what you're looking for: allows you to run database updates that will get rolled back when the test exits. The plugin is based on Alex Michael's blog post, with some additional support for nested transactions that covers a wider array of user cases. There are also some configuration options for mocking out connectibles in your app so you can run arbitrary methods from your codebase, too.

For test_accounts.py, you could do something like this:

from example import db, Account


class TestAccounts(object):

    def test_update_view(self, db_session):

        test_acct = Account(username='abc')

        db_session.add(test_acct)
        db_session.commit()

        resp = self.client.post('/update',
                                data={'a':1},
                                content_type='application/json')

        assert resp.status_code == 200

The plugin needs access to your database through a _db fixture, but since you already have a db fixture defined in conftest.py, you can set up database access easily:

@pytest.fixture(scope='session')
def _db(db):
    return db

You can find detail on how to setup and installation in the docs. Hope this helps!

Answer 2

I'm also having issues with the rollback, my code can be found here

After reading some documentation, it seems the begin() function should be called on the session.

So in your case I would update the session fixture to this:

@pytest.yield_fixture(scope='function', autouse=True)
def session(db, request):
    """Creates a new database session for a test."""

    db.session.begin()

    yield db.session

    db.session.rollback()
    db.session.remove()

I didn't test this code, but when I try it on my code I get the following error:

INTERNALERROR> Traceback (most recent call last):
INTERNALERROR>   File "./venv/lib/python2.7/site-packages/_pytest/main.py", line 90, in wrap_session
INTERNALERROR>     session.exitstatus = doit(config, session) or 0
...
INTERNALERROR>   File "./venv/lib/python2.7/site-packages/_pytest/python.py", line 59, in filter_traceback
INTERNALERROR>     return entry.path != cutdir1 and not entry.path.relto(cutdir2)
INTERNALERROR> AttributeError: 'str' object has no attribute 'relto'

Answer 3

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from unittest import TestCase

# global application scope.  create Session class, engine
Session = sessionmaker()

engine = create_engine('postgresql://...')

class SomeTest(TestCase):
    def setUp(self):
        # connect to the database
        self.connection = engine.connect()

        # begin a non-ORM transaction
        self.trans = self.connection.begin()

        # bind an individual Session to the connection
        self.session = Session(bind=self.connection)

    def test_something(self):
        # use the session in tests.

        self.session.add(Foo())
        self.session.commit()

    def tearDown(self):
        self.session.close()

        # rollback - everything that happened with the
        # Session above (including calls to commit())
        # is rolled back.
        self.trans.rollback()

        # return connection to the Engine
        self.connection.close()

sqlalchemy doc has solution for the case