[Zope3-checkins] CVS: Zope3/src/transaction/tests - test_register_compat.py:1.2

Jeremy Hylton jeremy at zope.com
Wed Mar 31 22:57:26 EST 2004


Update of /cvs-repository/Zope3/src/transaction/tests
In directory cvs.zope.org:/tmp/cvs-serv23866/src/transaction/tests

Added Files:
	test_register_compat.py 
Log Message:
Merge the jeremy-txn-branch to the head.

This branch introduces a new transaction API.  The key features are:
  - top-level functions in transaction -- get(), commit(), abort() 
  - explicit transaction manager objects
  - Transaction objects are used for exactly one transaction
  - support for transaction synchronizers

The changes here are still provisional, but we want to get them off an
obscure branch and onto the head for further development.


=== Zope3/src/transaction/tests/test_register_compat.py 1.1 => 1.2 ===
--- /dev/null	Wed Mar 31 22:57:26 2004
+++ Zope3/src/transaction/tests/test_register_compat.py	Wed Mar 31 22:56:56 2004
@@ -0,0 +1,149 @@
+##############################################################################
+#
+# Copyright (c) 2004 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""Test backwards compatibility for resource managers using register().
+
+The transaction package supports several different APIs for resource
+managers.  The original ZODB3 API was implemented by ZODB.Connection.
+The Connection passed persistent objects to a Transaction's register()
+method.  It's possible that third-party code also used this API, hence
+these tests that the code that adapts the old interface to the current
+API works.
+
+These tests use a TestConnection object that implements the old API.
+They check that the right methods are called and in roughly the right
+order.
+
+Common cases
+------------
+
+First, check that a basic transaction commit works.
+
+>>> cn = TestConnection()
+>>> cn.register(Object())
+>>> cn.register(Object())
+>>> cn.register(Object())
+>>> transaction.commit()
+>>> len(cn.committed)
+3
+>>> len(cn.aborted)
+0
+>>> cn.calls
+['begin', 'vote', 'finish']
+
+Second, check that a basic transaction abort works.  If the
+application calls abort(), then the transaction never gets into the
+two-phase commit.  It just aborts each object.
+
+>>> cn = TestConnection()
+>>> cn.register(Object())
+>>> cn.register(Object())
+>>> cn.register(Object())
+>>> transaction.abort()
+>>> len(cn.committed)
+0
+>>> len(cn.aborted)
+3
+>>> cn.calls
+[]
+
+Error handling
+--------------
+
+The tricky part of the implementation is recovering from an error that
+occurs during the two-phase commit.  We override the commit() and
+abort() methods of Object to cause errors during commit.
+
+Note that the implementation uses lists internally, so that objects
+are committed in the order they are registered.  (In the presence of
+multiple resource managers, objects from a single resource manager are
+committed in order.  The order of resource managers depends on
+sortKey().)  I'm not sure if this is an accident of the implementation
+or a feature that should be supported by any implementation.
+
+>>> cn = TestConnection()
+>>> cn.register(Object())
+>>> cn.register(CommitError())
+>>> cn.register(Object())
+>>> transaction.commit()
+Traceback (most recent call last):
+ ...
+RuntimeError: commit
+>>> len(cn.committed)
+1
+>>> len(cn.aborted)
+2
+
+"""
+
+import transaction
+
+class Object(object):
+
+    def commit(self):
+        pass
+
+    def abort(self):
+        pass
+
+class CommitError(Object):
+
+    def commit(self):
+        raise RuntimeError("commit")
+
+class AbortError(Object):
+
+    def abort(self):
+        raise RuntimeError("abort")
+
+class BothError(CommitError, AbortError):
+    pass
+
+class TestConnection:
+
+    def __init__(self):
+        self.committed = []
+        self.aborted = []
+        self.calls = []
+
+    def register(self, obj):
+        obj._p_jar = self
+        transaction.get().register(obj)
+
+    def sortKey(self):
+        return str(id(self))
+
+    def tpc_begin(self, txn, sub):
+        self.calls.append("begin")
+
+    def tpc_vote(self, txn):
+        self.calls.append("vote")
+
+    def tpc_finish(self, txn):
+        self.calls.append("finish")
+
+    def tpc_abort(self, txn):
+        self.calls.append("abort")
+
+    def commit(self, obj, txn):
+        obj.commit()
+        self.committed.append(obj)
+
+    def abort(self, obj, txn):
+        obj.abort()
+        self.aborted.append(obj)
+
+import doctest
+
+def test_suite():
+    return doctest.DocTestSuite()




More information about the Zope3-Checkins mailing list