From 9f366afdda4b508eb4ef3e626da2fec98ad04773 Mon Sep 17 00:00:00 2001
From: Ants Aasma <ants.aasma@gmail.com>
Date: Sun, 20 Jan 2008 03:22:00 +0000
Subject: - parent transactions weren't started on the connection when adding a
 connection to a nested session transaction. - session.transaction now always
 refers to the innermost active transaction, even when commit/rollback are
 called directly on the session transaction object. - when preparing a
 two-phase transaction fails on one connection all the connections are rolled
 back. - two phase transactions can now be prepared. - session.close() didn't
 close all transactions when nested transactions were used. - rollback()
 previously erroneously set the current transaction directly to the parent of
 the transaction that could be rolled back to. - autoflush for commit() wasn't
 flushing for simple subtransactions.

---
 test/orm/session.py | 118 +++++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 117 insertions(+), 1 deletion(-)

(limited to 'test')

diff --git a/test/orm/session.py b/test/orm/session.py
index 0a6157e0a..930efe07a 100644
--- a/test/orm/session.py
+++ b/test/orm/session.py
@@ -1,6 +1,6 @@
 import testenv; testenv.configure_for_tests()
 from sqlalchemy import *
-from sqlalchemy import exceptions
+from sqlalchemy import exceptions, util
 from sqlalchemy.orm import *
 from sqlalchemy.orm.session import SessionExtension
 from sqlalchemy.orm.session import Session as SessionCls
@@ -355,6 +355,122 @@ class SessionTest(AssertMixin):
         assert len(sess.query(User).all()) == 1
         sess.close()
 
+    @testing.unsupported('sqlite', 'mssql', 'firebird', 'sybase', 'access',
+                         'oracle', 'maxdb')
+    @testing.exclude('mysql', '<', (5, 0, 3))
+    def test_nested_transaction_connection_add(self):
+        class User(object): pass
+        mapper(User, users)
+        
+        sess = create_session(transactional=False)
+        
+        sess.begin()
+        sess.begin_nested()
+        
+        u1 = User()
+        sess.save(u1)
+        sess.flush()
+        
+        sess.rollback()
+        
+        u2 = User()
+        sess.save(u2)
+        
+        sess.commit()
+        
+        self.assertEquals(util.Set(sess.query(User).all()), util.Set([u2]))
+        
+        sess.begin()
+        sess.begin_nested()
+        
+        u3 = User()
+        sess.save(u3)
+        sess.commit() # commit the nested transaction
+        sess.rollback()
+        
+        self.assertEquals(util.Set(sess.query(User).all()), util.Set([u2]))
+        
+        sess.close()
+    
+    @testing.unsupported('sqlite', 'mssql', 'firebird', 'sybase', 'access',
+                         'oracle', 'maxdb')
+    @testing.exclude('mysql', '<', (5, 0, 3))
+    def test_mixed_transaction_control(self):
+        class User(object): pass
+        mapper(User, users)
+        
+        sess = create_session(transactional=False)
+        
+        sess.begin()
+        sess.begin_nested()
+        transaction = sess.begin()
+    
+        sess.save(User())
+        
+        transaction.commit()
+        sess.commit()
+        sess.commit()
+        
+        sess.close()
+        
+        self.assertEquals(len(sess.query(User).all()), 1)
+        
+        t1 = sess.begin()
+        t2 = sess.begin_nested()
+    
+        sess.save(User())
+        
+        t2.commit()
+        assert sess.transaction is t1
+        
+        sess.close()
+    
+    @testing.unsupported('sqlite', 'mssql', 'firebird', 'sybase', 'access',
+                         'oracle', 'maxdb')
+    @testing.exclude('mysql', '<', (5, 0, 3))
+    def test_mixed_transaction_close(self):
+        class User(object): pass
+        mapper(User, users)
+        
+        sess = create_session(transactional=True)
+        
+        sess.begin_nested()
+    
+        sess.save(User())
+        sess.flush()
+        
+        sess.close()
+        
+        sess.save(User())
+        sess.commit()
+        
+        sess.close()
+        
+        self.assertEquals(len(sess.query(User).all()), 1)
+    
+    @testing.unsupported('sqlite', 'mssql', 'firebird', 'sybase', 'access',
+                         'oracle', 'maxdb')
+    @testing.exclude('mysql', '<', (5, 0, 3))
+    def test_error_on_using_inactive_session(self):
+        class User(object): pass
+        mapper(User, users)
+        
+        sess = create_session(transactional=False)
+        
+        try:
+            sess.begin()
+            sess.begin()
+        
+            sess.save(User())
+            sess.flush()
+            
+            sess.rollback()
+            sess.begin()
+            assert False
+        except exceptions.InvalidRequestError, e:
+            self.assertEquals(str(e), "The transaction is inactive due to a rollback in a subtransaction and should be closed")
+        sess.close()
+    
     @engines.close_open_connections
     def test_bound_connection(self):
         class User(object):pass
-- 
cgit v1.2.1