Rollback of transactions not working with py.test and Flask

Question

I use py.test to test my Flask application, but I get IntegrityError because I create the same model in two different tests.

I am using postgreSQL 9.3.5 and Flask-SQLAlchemy 1.0.

EDIT I ​​updated my sessoin fixture with Jeremy Allen's answer and it fixed a lot of errors. However, it seems that when I use the user device, I still get IntegrityErrors

Mistake

E IntegrityError: (IntegrityError) duplicate key value violates unique constraint "ix_users_email" E DETAIL: Key (email)=([e-mail]) already exists. E 'INSERT INTO users (email, username, name, role_id, company_id, password_hash, confirmed, member_since, last_seen) VALUES (%(email)s, %(username)s, %(name)s, %(role_id)s, %(company_id)s, %(password_hash)s, %(confirmed)s, %(member_since)s, %(last_seen)s) RETURNING users.id' {'username': 'not_used', 'confirmed': True, 'name': 'To be Removed', 'member_since': datetime.datetime(2014, 10, 29, 19, 19, 41, 7929), 'company_id': None, 'role_id': 3, 'last_seen': datetime.datetime(2014, 10, 29, 19, 19, 41, 7941), 'email': '[e-mail]', 'password_hash': 'pbkdf2:sha1:1000$cXUh6GbJ$6f38242871cff5e4cce4c1dc49a62c4aea4ba1f3'} 

conftest.py

 @pytest.yield_fixture(scope='session') def app(): app = create_app('testing') app.config['SERVER_NAME'] = 'example.com:1234' ctx = app.app_context() ctx.push() app.response_class = TestResponse app.test_client_class = TestClient yield app ctx.pop() @pytest.fixture(scope='session') def db(app): _db.drop_all() _db.create_all() Permission.insert_initial() Role.insert_initial() Technology.insert_initial() Product.insert_initial() Actor.insert_initial() Industry.insert_initial() DeliveryCategory.insert_initial() DeliveryMethod.insert_initial() user = User(email='[e-mail]', username='admin', confirmed=True, password='admin', name='Admin') user.role = Role.query.filter_by(name='Administrator').first() _db.session.add(user) _db.session.commit() return _db @pytest.yield_fixture(scope='function') def session(db): db.session.begin_nested() yield db.session db.session.rollback() @pytest.yield_fixture(scope='function') def user(session): yield session.query(User).filter_by(email='[e-mail]').first() @pytest.yield_fixture(scope='function') def client(app, user): client = app.test_client() client.auth = 'Basic ' + b64encode((user.email + ':' + 'admin').encode('utf-8')).decode('utf-8') yield client 

Tests that don't work

 def test_edit_agenda_add_company_rep_without_company(session, client, user): user2 = User(name='To be Removed', password='not_used', username='not_used', confirmed=True, email='[e-mail]', role=Role.query.filter_by(name='User').first()) agenda = Agenda(name='Invalid Company Rep', creator=user) session.add(agenda) session.commit() response = client.jput('/api/v1.0/agendas/%s' % agenda.id, data={ 'company_representative': user2.id } ) assert response.status_code == 200 def test_edit_agenda_add_user_already_in_agenda(session, client, user): user2 = User(name='To be Removed', password='not_used', username='not_used', confirmed=True, email='[e-mail]', role=Role.query.filter_by(name='User').first()) agenda = Agenda(name='Invalid Company Rep', creator=user) agenda.users.append(user2) session.add(agenda) session.commit() response = client.jput('/api/v1.0/agendas/%s' % agenda.id, data={ 'users': [user2.id] } ) assert response.status_code == 200 

Tests that pass

 def test_get_agenda_modules_where_agenda_that_does_not_exist(session, app): # Create admin user with permission to create models user = User(email='[e-mail]', username='admin2', confirmed=True, password='admin2') user.role = Role.query.filter_by(name='Administrator').first() session.add(user) session.commit() client = app.test_client() client.auth = 'Basic ' + b64encode( (user.email + ':' + 'admin2').encode('utf-8')).decode('utf-8') response = client.jget('/api/v1.0/agenda-modules/%s/%s' % (5, 4)) assert response.status_code == 404 def test_get_agenda_modules_agenda_modules_does_not_exist(session, app): agenda = Agenda(name='Is tired in the AM') session.add(agenda) # Create admin user with permission to create models user = User(email='[e-mail]', username='admin2', confirmed=True, password='admin2') user.role = Role.query.filter_by(name='Administrator').first() session.add(user) session.commit() client = app.test_client() client.auth = 'Basic ' + b64encode( (user.email + ':' + 'admin2').encode('utf-8')).decode('utf-8') response = client.jget('/api/v1.0/agenda-modules/%s/%s' % (agenda.id, 4)) assert response.status_code == 400 assert response.jdata['message'] == 'AgendaModule does not exist.' 

Answer 1

It looks like you are trying to join a session in an external transaction , and you are using flask-sqlalchemy.

Your code does not work as expected, because the session actually ends up using a different connection from the one you are starting the transaction from.

1. You need to bind a session to a connection

Like in the example above. A quick change to your code in conftest.py should do this:

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

Unfortunately , due to flask-sqlalchemy SignallingSession (as in version 2.0), your 'bind' argument will be canceled!

This is because SignallingSession sets the binds argument in such a way that it takes precedence over our bind argument, and it does not offer us a good way to specify our own binds argument.

In December 2013 there is a request

And then subclass SQLAlchemy (core flask-sqlalchemy class) to use our SessionWithBinds instead 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 should use this class instead of SQLAlchemy:

 db = TestFriendlySQLAlchemy() 

Finally, go back to our conftest.py to specify the new "binds":

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

Your transactions should now roll back as expected.

All this is a little complicated ...

You can try using instead. This requires that your database support SQL SAVEPOINT (PostgreSQL).

Change fixture conftest.py:

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

Additional information on using SAVEPOINT in SQLAlchemy:

This is pretty simple, but will work as long as the code you are testing does not call rollback . If this is a problem, check out the code under the heading "Test Rollback Support"

Answer 2

You really did not say that you use to manage the database, there is no information about which library is behind _db or any of the model classes.

But no matter what, I suspect that the call to session.commit() is probably related to the reason for the transaction. Ultimately, you will need to read the docs about what session.commit() does in the structure used.

Answer 3

The key here is to run the tests in a nested session and then roll back everything after each test (this also assumes no dependency on your tests).

I suggest using the following approach by running each of your tests in a nested transaction:

 # module conftest.py import pytest from app import create_app from app import db as _db from sqlalchemy import event from 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()