Sqlalchemy transaction not rollback

For my integration test, I wrote my own base class from unittest.TestCase .

 def initialize_sql(engine, dbsession): dbsession.configure(bind=engine) Base.metadata.bind = engine Base.metadata.drop_all(engine) # ensure the database is clean! Base.metadata.create_all(engine) try: populate(dbsession) except IntegrityError: transaction.abort() class HeavyTestBaseCase(unittest.TestCase): __TEST__ = False test_ini = 'test.ini' @classmethod def setUpClass(cls): TEST_ROOT = os.path.dirname(os.path.abspath(__file__)) settings = appconfig('config:' + os.path.join(TEST_ROOT, cls.test_ini)) cls.engine = engine_from_config(settings, prefix='sqlalchemy.') print 'Creating the tables on the test database %s' % cls.engine cls.dbsession = scoped_session(sessionmaker( extension=ZopeTransactionExtension())) config = Configurator(settings=settings) initialize_sql(cls.engine, cls.dbsession) def tearDown(self): transaction.abort() # strange name for rollback ... @classmethod def tearDownClass(cls): Base.metadata.drop_all(cls.engine) 

Now here is a test case:

 from mock import Mock, patch from aurum.models import User from aurum.user import register_user class TestRegisterUserIntegration(HeavyTestBaseCase): __TEST__ = True @classmethod def setUpClass(cls): cls.uid = 'uid1234' cls.username = 'user01' cls.password = 'password01' cls.masteru = 'masteru' cls.masterp = 'masterp' cls.gcs_patcher = patch('aurum.user.GCS', autospec=True) cls.gcs = cls.gcs_patcher.start() cls.gcs.return_value.register.return_value = cls.uid super(TestRegisterUserIntegration, cls).setUpClass() def test_register_user01_successful_return_useid_and_shared_key(self): result = register_user(self.username, self.password, self.masteru, self.masterp) self.assertEqual(result.keys(), ['user_id', 'shared_key']) self.assertEqual(result['user_id'], self.uid) def test_register_user01_successful_write_to_database_query_is_not_none(self): register_user(self.username, self.password, self.masteru, self.masterp) result = self.dbsession.query(User).filter_by(username=self.username).first() self.assertTrue(result is not None) 

But the rollback did nothing. One of the tests will fail due to duplication of key restrictions, which means that the commit did not hit.

In the actual code, before returning, I wrote transaction.commit() to commit the changes.

Any idea what is going on? Thanks


Adapt @zzzek advise, here is the trace: http://pastebin.com/K9fin7ZH

Actual code about this:

 def add_user(dbsession, username, password): with transaction.manager: user = User(...) dbsession.add(user) 
+4
source share

All Articles