TLConnection insures that it is used to create a transaction via the session when begin() is called, so that it has proper transactional context, + unittests

1 files changed, 32 insertions, 0 deletions

diff --git a/test/engine/transaction.py b/test/engine/transaction.py
index b8f8af96e..5f429761e 100644
--- a/test/engine/transaction.py
+++ b/test/engine/transaction.py
@@ -157,6 +157,38 @@ class TLTransactionTest(testbase.PersistTest):
         finally:
             external_connection.close()
 
+    @testbase.unsupported('mysql')
+    def testrollback_off_conn(self):
+        conn = tlengine.contextual_connect()
+        trans = conn.begin()
+        conn.execute(users.insert(), user_id=1, user_name='user1')
+        conn.execute(users.insert(), user_id=2, user_name='user2')
+        conn.execute(users.insert(), user_id=3, user_name='user3')
+        trans.rollback()
+
+        external_connection = tlengine.connect()
+        result = external_connection.execute("select * from query_users")
+        try:
+            assert len(result.fetchall()) == 0
+        finally:
+            external_connection.close()
+
+    @testbase.unsupported('mysql')
+    def testcommit_off_conn(self):
+        conn = tlengine.contextual_connect()
+        trans = conn.begin()
+        conn.execute(users.insert(), user_id=1, user_name='user1')
+        conn.execute(users.insert(), user_id=2, user_name='user2')
+        conn.execute(users.insert(), user_id=3, user_name='user3')
+        trans.commit()
+
+        external_connection = tlengine.connect()
+        result = external_connection.execute("select * from query_users")
+        try:
+            assert len(result.fetchall()) == 3
+        finally:
+            external_connection.close()
+        
     @testbase.unsupported('mysql', 'sqlite')
     def testnesting(self):
         """tests nesting of tranacstions"""