[Zodb-checkins] CVS: Zope/lib/python/ZODB/tests - testZODB.py:1.6.6.1

[Shane Hathaway]

Update of /cvs-repository/Zope/lib/python/ZODB/tests
In directory cvs.zope.org:/tmp/cvs-serv13538/tests

Modified Files:
      Tag: shane-local-transactions-branch
	testZODB.py 
Log Message:
Enabled decoupling of ZODB connections and transactions, a feature
requested often on the zodb-dev list.

The Connection class now has a method setLocalTransaction().
(Suggestions for a better method name accepted.)  Once your app calls
this, your app is expected to use connection.getTransaction() or
ob._p_jar.getTransaction() instead of the get_transaction() function
to commit or abort the transaction.  Doing this enables your
application to have multiple writable connections open in a single
thread, which is especially usefully in GUI applications.

If you don't call setLocalTransaction(), ZODB behaves as it always
has, with transactions bound to threads.

This feature is experimental, but the implementation seems about
right.  Unfortunately I had to change the CommitVersion, AbortVersion,
and TransactionalUndo classes so that instances don't automatically
register themselves with the transaction, but they should not have
been doing that in the first place, and hopefully these classes aren't
being used directly by applications.  Most apps will see no change.



=== Zope/lib/python/ZODB/tests/testZODB.py 1.6 => 1.6.6.1 ===
--- Zope/lib/python/ZODB/tests/testZODB.py:1.6	Mon Nov 18 18:17:40 2002
+++ Zope/lib/python/ZODB/tests/testZODB.py	Tue Dec  3 11:03:13 2002
@@ -98,29 +98,57 @@
         # Make sure the changes to make empty transactions a no-op
         # still allow things like abortVersion().  This should work
         # because abortVersion() calls tpc_begin() itself.
-        r = self._db.open("version").root()
-        r[1] = 1
-        get_transaction().commit()
+        conn = self._db.open("version")
+        try:
+            r = conn.root()
+            r[1] = 1
+            get_transaction().commit()
+        finally:
+            conn.close()
         self._db.abortVersion("version")
         get_transaction().commit()
 
-def test_suite():
-    return unittest.makeSuite(ZODBTests, 'check')
-
-def main():
-    alltests=test_suite()
-    runner = unittest.TextTestRunner()
-    runner.run(alltests)
+    def checkLocalTransactions(self):
+        # Test of transactions that apply to only the connection,
+        # not the thread.
+        conn1 = self._db.open()
+        conn2 = self._db.open()
+        try:
+            conn1.setLocalTransaction()
+            conn2.setLocalTransaction()
+            r1 = conn1.root()
+            r2 = conn2.root()
+            if r1.has_key('item'):
+                del r1['item']
+                conn1.getTransaction().commit()
+            r1.get('item')
+            r2.get('item')
+            r1['item'] = 1
+            conn1.getTransaction().commit()
+            self.assertEqual(r1['item'], 1)
+            # r2 has not seen a transaction boundary,
+            # so it should be unchanged.
+            self.assertEqual(r2.get('item'), None)
+            conn2.sync()
+            # Now r2 is updated.
+            self.assertEqual(r2['item'], 1)
 
-def debug():
-    test_suite().debug()
+            # Now, for good measure, send an update in the other direction.
+            r2['item'] = 2
+            conn2.getTransaction().commit()
+            self.assertEqual(r1['item'], 1)
+            self.assertEqual(r2['item'], 2)
+            conn1.sync()
+            conn2.sync()
+            self.assertEqual(r1['item'], 2)
+            self.assertEqual(r2['item'], 2)
+        finally:
+            conn1.close()
+            conn2.close()
 
-def pdebug():
-    import pdb
-    pdb.run('debug()')
+def test_suite():
+    return unittest.makeSuite(ZODBTests, 'check')
 
 if __name__=='__main__':
-    if len(sys.argv) > 1:
-        globals()[sys.argv[1]]()
-    else:
-        main()
+    unittest.main(defaultTest='test_suite')
+