- Fixed a regression from 0.7 where the contextmanager feature

of :meth:`.Session.begin_nested` would fail to correctly
roll back the transaction when a flush error occurred, instead
raising its own exception while leaving the session still
pending a rollback.  [ticket:2718]

1 files changed, 63 insertions, 1 deletions

diff --git a/test/orm/test_transaction.py b/test/orm/test_transaction.py
index 64b05a131..2866ab4ab 100644
--- a/test/orm/test_transaction.py
+++ b/test/orm/test_transaction.py
@@ -1,4 +1,4 @@
-
+from __future__ import with_statement
 from sqlalchemy.testing import eq_, assert_raises, \
     assert_raises_message, assert_warnings
 from sqlalchemy import *
@@ -83,6 +83,8 @@ class SessionTransactionTest(FixtureTest):
             conn.close()
             raise
 
+
+
     @testing.requires.savepoints
     def test_heavy_nesting(self):
         users = self.tables.users
@@ -620,6 +622,66 @@ class CleanSavepointTest(FixtureTest):
                                     synchronize_session='fetch')
         self._run_test(update_fn)
 
+class ContextManagerTest(FixtureTest):
+    run_inserts = None
+
+    @testing.requires.savepoints
+    @engines.close_open_connections
+    def test_contextmanager_nested_rollback(self):
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+
+        sess = Session()
+        def go():
+            with sess.begin_nested():
+                sess.add(User())   # name can't be null
+                sess.flush()
+
+        # and not InvalidRequestError
+        assert_raises(
+            sa_exc.DBAPIError,
+            go
+        )
+
+        with sess.begin_nested():
+            sess.add(User(name='u1'))
+
+        eq_(sess.query(User).count(), 1)
+
+    def test_contextmanager_commit(self):
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+
+        sess = Session(autocommit=True)
+        with sess.begin():
+            sess.add(User(name='u1'))
+
+        sess.rollback()
+        eq_(sess.query(User).count(), 1)
+
+    def test_contextmanager_rollback(self):
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+
+        sess = Session(autocommit=True)
+        def go():
+            with sess.begin():
+                sess.add(User())  # name can't be null
+        assert_raises(
+            sa_exc.DBAPIError,
+            go
+        )
+
+        eq_(sess.query(User).count(), 0)
+
+        with sess.begin():
+            sess.add(User(name='u1'))
+        eq_(sess.query(User).count(), 1)
+
+
 class AutoExpireTest(_LocalFixture):
 
     def test_expunge_pending_on_rollback(self):