1 files changed, 113 insertions, 0 deletions
diff --git a/test/ext/legacy_objectstore.py b/test/ext/legacy_objectstore.py
new file mode 100644
index 000000000..3aa99a1ae
--- /dev/null
+++ b/test/ext/legacy_objectstore.py
@@ -0,0 +1,113 @@
+from testbase import PersistTest, AssertMixin
+import unittest, sys, os
+from sqlalchemy import *
+import StringIO
+import testbase
+
+from tables import *
+import tables
+
+install_mods('legacy_session')
+
+
+class LegacySessionTest(AssertMixin):
+    def setUpAll(self):
+        db.echo = False
+        users.create()
+        db.echo = testbase.echo
+    def tearDownAll(self):
+        db.echo = False
+        users.drop()
+        db.echo = testbase.echo
+    def setUp(self):
+        objectstore.get_session().clear()
+        clear_mappers()
+        tables.user_data()
+        #db.echo = "debug"
+    def tearDown(self):
+        tables.delete_user_data()
+        
+    def test_nested_begin_commit(self):
+        """tests that nesting objectstore transactions with multiple commits
+        affects only the outermost transaction"""
+        class User(object):pass
+        m = mapper(User, users)
+        def name_of(id):
+            return users.select(users.c.user_id == id).execute().fetchone().user_name
+        name1 = "Oliver Twist"
+        name2 = 'Mr. Bumble'
+        self.assert_(name_of(7) != name1, msg="user_name should not be %s" % name1)
+        self.assert_(name_of(8) != name2, msg="user_name should not be %s" % name2)
+        s = objectstore.get_session()
+        trans = s.begin()
+        trans2 = s.begin()
+        m.get(7).user_name = name1
+        trans3 = s.begin()
+        m.get(8).user_name = name2
+        trans3.commit()
+        s.commit() # should do nothing
+        self.assert_(name_of(7) != name1, msg="user_name should not be %s" % name1)
+        self.assert_(name_of(8) != name2, msg="user_name should not be %s" % name2)
+        trans2.commit()
+        s.commit()  # should do nothing
+        self.assert_(name_of(7) != name1, msg="user_name should not be %s" % name1)
+        self.assert_(name_of(8) != name2, msg="user_name should not be %s" % name2)
+        trans.commit()
+        self.assert_(name_of(7) == name1, msg="user_name should be %s" % name1)
+        self.assert_(name_of(8) == name2, msg="user_name should be %s" % name2)
+
+    def test_nested_rollback(self):
+        """tests that nesting objectstore transactions with a rollback inside
+        affects only the outermost transaction"""
+        class User(object):pass
+        m = mapper(User, users)
+        def name_of(id):
+            return users.select(users.c.user_id == id).execute().fetchone().user_name
+        name1 = "Oliver Twist"
+        name2 = 'Mr. Bumble'
+        self.assert_(name_of(7) != name1, msg="user_name should not be %s" % name1)
+        self.assert_(name_of(8) != name2, msg="user_name should not be %s" % name2)
+        s = objectstore.get_session()
+        trans = s.begin()
+        trans2 = s.begin()
+        m.get(7).user_name = name1
+        trans3 = s.begin()
+        m.get(8).user_name = name2
+        trans3.rollback()
+        self.assert_(name_of(7) != name1, msg="user_name should not be %s" % name1)
+        self.assert_(name_of(8) != name2, msg="user_name should not be %s" % name2)
+        trans2.commit()
+        self.assert_(name_of(7) != name1, msg="user_name should not be %s" % name1)
+        self.assert_(name_of(8) != name2, msg="user_name should not be %s" % name2)
+        trans.commit()
+        self.assert_(name_of(7) != name1, msg="user_name should not be %s" % name1)
+        self.assert_(name_of(8) != name2, msg="user_name should not be %s" % name2)
+
+    def test_true_nested(self):
+        """tests creating a new Session inside a database transaction, in 
+        conjunction with an engine-level nested transaction, which uses
+        a second connection in order to achieve a nested transaction that commits, inside
+        of another engine session that rolls back."""
+#        testbase.db.echo='debug'
+        class User(object):
+            pass
+        testbase.db.begin()
+        try:
+            m = mapper(User, users)
+            name1 = "Oliver Twist"
+            name2 = 'Mr. Bumble'
+            m.get(7).user_name = name1
+            s = objectstore.Session(nest_on=testbase.db)
+            m.using(s).get(8).user_name = name2
+            s.commit()
+            objectstore.commit()
+            testbase.db.rollback()
+        except:
+            testbase.db.rollback()
+            raise
+        objectstore.clear()
+        self.assert_(m.get(8).user_name == name2)
+        self.assert_(m.get(7).user_name != name1)
+
+if __name__ == "__main__":
+    testbase.main()