TLEngine needed a partial rewrite....

1 files changed, 63 insertions, 0 deletions

diff --git a/test/transaction.py b/test/transaction.py
index d76d7e0f4..244da1649 100644
--- a/test/transaction.py
+++ b/test/transaction.py
@@ -5,6 +5,7 @@ import tables
 db = testbase.db
 from sqlalchemy import *
 
+
 class TransactionTest(testbase.PersistTest):
     def setUpAll(self):
         global users, metadata
@@ -34,6 +35,23 @@ class TransactionTest(testbase.PersistTest):
         assert len(result.fetchall()) == 0
         connection.close()
 
+    def testnesting(self):
+        connection = testbase.db.connect()
+        transaction = connection.begin()
+        connection.execute(users.insert(), user_id=1, user_name='user1')
+        connection.execute(users.insert(), user_id=2, user_name='user2')
+        connection.execute(users.insert(), user_id=3, user_name='user3')
+        trans2 = connection.begin()
+        connection.execute(users.insert(), user_id=4, user_name='user4')
+        connection.execute(users.insert(), user_id=5, user_name='user5')
+        trans2.commit()
+        transaction.rollback()
+        self.assert_(connection.scalar("select count(1) from query_users") == 0)
+
+        result = connection.execute("select * from query_users")
+        assert len(result.fetchall()) == 0
+        connection.close()
+        
 class AutoRollbackTest(testbase.PersistTest):
     def setUpAll(self):
         global metadata
@@ -58,6 +76,51 @@ class AutoRollbackTest(testbase.PersistTest):
         # comment out the rollback in pool/ConnectionFairy._close() to see !
         users.drop(conn2)
         conn2.close()
+
+class TLTransactionTest(testbase.PersistTest):
+    def setUpAll(self):
+        global users, metadata, tlengine
+        tlengine = create_engine(testbase.db_uri, strategy='threadlocal', echo=True)
+        metadata = MetaData()
+        users = Table('query_users', metadata,
+            Column('user_id', INT, primary_key = True),
+            Column('user_name', VARCHAR(20)),
+        )
+        users.create(tlengine)
+    def tearDown(self):
+        tlengine.execute(users.delete())
+    def tearDownAll(self):
+        users.drop(tlengine)
+        tlengine.dispose()
+        
+    @testbase.unsupported('mysql')
+    def testrollback(self):
+        """test a basic rollback"""
+        tlengine.begin()
+        tlengine.execute(users.insert(), user_id=1, user_name='user1')
+        tlengine.execute(users.insert(), user_id=2, user_name='user2')
+        tlengine.execute(users.insert(), user_id=3, user_name='user3')
+        tlengine.rollback()
+
+        result = tlengine.execute("select * from query_users")
+        assert len(result.fetchall()) == 0
+
+    @testbase.unsupported('mysql', 'sqlite')
+    def testnesting(self):
+        """test a basic rollback"""
+        external_connection = tlengine.connect()
+        self.assert_(external_connection.connection is not tlengine.contextual_connect().connection)
+        tlengine.begin()
+        tlengine.execute(users.insert(), user_id=1, user_name='user1')
+        tlengine.execute(users.insert(), user_id=2, user_name='user2')
+        tlengine.execute(users.insert(), user_id=3, user_name='user3')
+        tlengine.begin()
+        tlengine.execute(users.insert(), user_id=4, user_name='user4')
+        tlengine.execute(users.insert(), user_id=5, user_name='user5')
+        tlengine.commit()
+        tlengine.rollback()
+        self.assert_(external_connection.scalar("select count(1) from query_users") == 0)
+        external_connection.close()
         
 if __name__ == "__main__":
     testbase.main()