Rollback transactions not working with py.test and Flask

It looks like you are trying to join a Session into an External Transaction and you are using flask-sqlalchemy.

Your code is not working as expected because the session actually ends up using a different connection to the one you are beginning the transaction on.

1. You need to bind the Session to the Connection

As in the example linked above. A quick change to your code in conftest.py should do it:

@pytest.yield_fixture(scope='function')def session(db):    ...    session = db.create_scoped_session(options={'bind':connection})    ...

Unfortunately, due to flask-sqlalchemy's SignallingSession (as at v2.0), your 'bind' argument will be overruled!

This is because SignallingSession set the 'binds' argument such that it will take precedence over our 'bind' argument and it doesn't offer us a nice way to specify our own 'binds' argument.

There is a GitHub pull request from December 2013 where someone else had the same problem.

2. Tweak flask-sqlalchemy

We can subclass SignallingSession to allow us to do what we want:

class SessionWithBinds(SignallingSession):    """The extends the flask-sqlalchemy signalling session so that we may    provide our own 'binds' argument.    """    def __init__(self, db, autocommit=False, autoflush=True, **options):        #: The application that this session belongs to.        self.app = db.get_app()        self._model_changes = {}        #: A flag that controls whether this session should keep track of        #: model modifications.  The default value for this attribute        #: is set from the ``SQLALCHEMY_TRACK_MODIFICATIONS`` config        #: key.        self.emit_modification_signals = \            self.app.config['SQLALCHEMY_TRACK_MODIFICATIONS']        bind = options.pop('bind', None) or db.engine        # Our changes to allow a 'binds' argument        try:            binds = options.pop('binds')        except KeyError:            binds = db.get_binds(self.app)        SessionBase.__init__(self, autocommit=autocommit, autoflush=autoflush,                             bind=bind,                             binds=binds, **options)

And then subclass SQLAlchemy (the main flask-sqlalchemy class) to use our SessionWithBinds in place of SignallingSession

class TestFriendlySQLAlchemy(SQLAlchemy):    """For overriding create_session to return our own Session class"""    def create_session(self, options):        return SessionWithBinds(self, **options)

Now you have to use this class in place of SQLAlchemy:

db = TestFriendlySQLAlchemy()

And finally, back in our conftest.py specify a new 'binds':

@pytest.yield_fixture(scope='function')def session(db):    ...    session = db.create_scoped_session(options={'bind':connection, 'binds':None})    ...

Now your transactions should rollback as expected.

This is all a bit complicated...

Instead of doing all this you could try using Session.begin_nested. It requires that your database supports SQL SAVEPOINTs (PostgreSQL does).

Change your conftest.py fixture:

@pytest.yield_fixture(scope='function')def session(db):    db.session.begin_nested()    yield db.session    db.session.rollback()

More info on using SAVEPOINTs in SQLAlchemy: http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#using-savepoint

This is pretty simple but will work as long as the code you are testing doesn't call rollback itself. If this is a problem take a look at the code here in the SQLAlchemy docs under the title "Supporting Tests with Rollbacks"

The key here is to run your tests within a nested session, and then rollback everything after the execution of each test (this also assumes there are no dependencies across your tests).

I suggest adopting the following approach, by running each of your tests within a nested transaction:

# module conftest.pyimport pytestfrom app import create_appfrom app import db as _dbfrom sqlalchemy import eventfrom sqlalchemy.orm import sessionmaker@pytest.fixture(scope="session")def app(request):    """    Returns session-wide application.    """    return create_app("testing")@pytest.fixture(scope="session")def db(app, request):    """    Returns session-wide initialised database.    """    with app.app_context():        _db.drop_all()        _db.create_all()@pytest.fixture(scope="function", autouse=True)def session(app, db, request):    """    Returns function-scoped session.    """    with app.app_context():        conn = _db.engine.connect()        txn = conn.begin()        options = dict(bind=conn, binds={})        sess = _db.create_scoped_session(options=options)        # establish  a SAVEPOINT just before beginning the test        # (http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#using-savepoint)        sess.begin_nested()        @event.listens_for(sess(), 'after_transaction_end')        def restart_savepoint(sess2, trans):            # Detecting whether this is indeed the nested transaction of the test            if trans.nested and not trans._parent.nested:                # The test should have normally called session.commit(),                # but to be safe we explicitly expire the session                sess2.expire_all()                sess2.begin_nested()        _db.session = sess        yield sess        # Cleanup        sess.remove()        # This instruction rollsback any commit that were executed in the tests.        txn.rollback()        conn.close()

You haven't really said what you use to manage the database, there is no clue what library is behind _db or any of the model classes.

But regardless of that I would suspect that the session.commit() call is likely involved in the reason the transaction gets committed. Ultimately you'll have to read the docs on what session.commit() does in the framework you're using.