[Zope3-checkins] CVS: Zope3/lib/python/Transaction/tests - abstestIDataManager.py:1.4

Jeremy Hylton jeremy@zope.com
Wed, 24 Jul 2002 19:02:54 -0400


Update of /cvs-repository/Zope3/lib/python/Transaction/tests
In directory cvs.zope.org:/tmp/cvs-serv1364/Transaction/tests

Modified Files:
	abstestIDataManager.py 
Log Message:
Revise Transaction API based on some discussion on persistence SIG.

The core of the change is that objects register with their data
managers and data managers join the transaction.  This simplifies the
bookkeeping done by the transaction.

The new API is based on the TP monitor API discussed in Transaction
Processing by Gray and Reuter.  The rough correspondence between the
old API and the new API is:

prepare() was tpc_begin() + commit() + tpc_vote()
commit() was tpc_finish()
abort() was tpc_abort()
savepoint() was commit(1)

The abort() and commit() methods of the old API are gone.  It's up to
the data managers to do the per-object accounting.

The basic Transaction package now defines a Transaction object without
the ZODB metadata fields.  ZODB will grown its own Transaction that
extends the base class.

Use an explicit Manager class and subclass that implements policy
associating each thread with a different transaction.


=== Zope3/lib/python/Transaction/tests/abstestIDataManager.py 1.3 => 1.4 ===
 import os
 from unittest import TestCase
 
-from Transaction import Transaction, get_transaction
+from Transaction.Transaction import Transaction
 
-class IDataManagerTests(TestCase):
+class IDataManagerTests(TestCase, object):
     
     def setUp(self):
         self.datamgr = None # subclass should override
         self.obj = None # subclass should define Persistent object
+        self.txn_factory = None
 
     def get_transaction(self):
-        t = Transaction()
-        t.setUser('IDataManagerTests')
-        t.note('dummy note')
-        return t
+        return self.txn_factory()
 
     ################################
     # IDataManager interface tests #
@@ -53,38 +51,16 @@
 
     def testCommitObj(self):
         tran = self.get_transaction()
-        self.datamgr.tpc_begin(tran)
-        self.datamgr.commit(self.obj, tran)
-        self.datamgr.tpc_vote(tran)
-        self.datamgr.tpc_finish(tran)
+        self.datamgr.prepare(tran)
+        self.datamgr.commit(tran)
 
     def testAbortTran(self):
         tran = self.get_transaction()
-        self.datamgr.tpc_begin(tran)
-        self.datamgr.commit(self.obj, tran)
-        self.datamgr.tpc_abort(tran)
+        self.datamgr.prepare(tran)
+        self.datamgr.abort(tran)
 
-    def testAbortObj(self):
+    def testRollback(self):
         tran = self.get_transaction()
-        self.datamgr.tpc_begin(tran)
-        self.datamgr.commit(self.obj, tran)
-        self.datamgr.abort(self.obj, tran)
-        self.datamgr.tpc_abort(tran)
-
-    def testSubCommit(self):
-        tran = self.get_transaction()
-        self.datamgr.tpc_begin(tran, 1)
-        self.datamgr.commit(self.obj, tran)
-        self.datamgr.commit_sub(tran)
-        self.datamgr.tpc_vote(tran)
-        self.datamgr.tpc_finish(tran)
-
-    #This test may not be semantically correct.  It hangs on FreeBSD.
-    #def testSubAbortCommitNull(self):
-    #   tran = self.get_transaction()
-    #   self.datamgr.tpc_begin(tran,1)
-    #   self.datamgr.commit(self.obj,tran)
-    #   self.datamgr.abort_sub(tran)
-    #   self.datamgr.tpc_vote(tran)
-    #   self.datamgr.tpc_finish(tran)
-
+        rb = self.datamgr.savepoint(tran)
+        if rb is not None:
+            rb.rollback()