[Checkins] SVN: transaction/branches/sphinx/ Consolidate example code.
Tres Seaver
cvs-admin at zope.org
Mon Dec 17 20:28:55 UTC 2012
Log message for revision 128708:
Consolidate example code.
Changed:
_U transaction/branches/sphinx/
U transaction/branches/sphinx/docs/datamanager.rst
U transaction/branches/sphinx/docs/resourcemanager.rst
D transaction/branches/sphinx/transaction/tests/SampleDataManager.py
D transaction/branches/sphinx/transaction/tests/SampleResourceManager.py
A transaction/branches/sphinx/transaction/tests/examples.py
U transaction/branches/sphinx/transaction/tests/test_transaction.py
-=-
Modified: transaction/branches/sphinx/docs/datamanager.rst
===================================================================
--- transaction/branches/sphinx/docs/datamanager.rst 2012-12-17 20:28:55 UTC (rev 128707)
+++ transaction/branches/sphinx/docs/datamanager.rst 2012-12-17 20:28:55 UTC (rev 128708)
@@ -6,9 +6,9 @@
.. doctest::
- >>> from transaction.tests.SampleDataManager import DataManager
+ >>> from transaction.tests.examples import DataManager
-This :class:`transaction.tests.SampleDataManager.DataManager` class
+This :class:`transaction.tests.examples.DataManager` class
provides a trivial data-manager implementation and docstrings to illustrate
the the protocol and to provide a tool for writing tests.
Modified: transaction/branches/sphinx/docs/resourcemanager.rst
===================================================================
--- transaction/branches/sphinx/docs/resourcemanager.rst 2012-12-17 20:28:55 UTC (rev 128707)
+++ transaction/branches/sphinx/docs/resourcemanager.rst 2012-12-17 20:28:55 UTC (rev 128708)
@@ -6,9 +6,9 @@
.. doctest::
- >>> from transaction.tests.SampleResourceManager import ResourceManager
+ >>> from transaction.tests.examples import ResourceManager
-This :class:`transaction.tests.SampleResourceManager.ResourceManager`
+This :class:`transaction.tests.examples.ResourceManager`
class provides a trivial resource-manager implementation and doc
strings to illustrate the protocol and to provide a tool for writing
tests.
Deleted: transaction/branches/sphinx/transaction/tests/SampleDataManager.py
===================================================================
--- transaction/branches/sphinx/transaction/tests/SampleDataManager.py 2012-12-17 20:28:55 UTC (rev 128707)
+++ transaction/branches/sphinx/transaction/tests/SampleDataManager.py 2012-12-17 20:28:55 UTC (rev 128708)
@@ -1,89 +0,0 @@
-##############################################################################
-#
-# Copyright (c) 2004 Zope Foundation and Contributors.
-# All Rights Reserved.
-#
-# This software is subject to the provisions of the Zope Public License,
-# Version 2.1 (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.
-#
-##############################################################################
-"""Sample objects for use in tests
-
-Used by the 'datamanager' chapter in the Sphinx docs.
-"""
-
-class DataManager(object):
- """Sample data manager
- """
- def __init__(self):
- self.state = 0
- self.sp = 0
- self.transaction = None
- self.delta = 0
- self.prepared = False
-
- def inc(self, n=1):
- self.delta += n
-
- def prepare(self, transaction):
- if self.prepared:
- raise TypeError('Already prepared')
- self._checkTransaction(transaction)
- self.prepared = True
- self.transaction = transaction
- self.state += self.delta
-
- def _checkTransaction(self, transaction):
- if (transaction is not self.transaction
- and self.transaction is not None):
- raise TypeError("Transaction missmatch",
- transaction, self.transaction)
-
- def abort(self, transaction):
- self._checkTransaction(transaction)
- if self.transaction is not None:
- self.transaction = None
-
- if self.prepared:
- self.state -= self.delta
- self.prepared = False
-
- self.delta = 0
-
- def commit(self, transaction):
- if not self.prepared:
- raise TypeError('Not prepared to commit')
- self._checkTransaction(transaction)
- self.delta = 0
- self.transaction = None
- self.prepared = False
-
- def savepoint(self, transaction):
- if self.prepared:
- raise TypeError("Can't get savepoint during two-phase commit")
- self._checkTransaction(transaction)
- self.transaction = transaction
- self.sp += 1
- return Rollback(self)
-
-
-class Rollback(object):
-
- def __init__(self, dm):
- self.dm = dm
- self.sp = dm.sp
- self.delta = dm.delta
- self.transaction = dm.transaction
-
- def rollback(self):
- if self.transaction is not self.dm.transaction:
- raise TypeError("Attempt to rollback stale rollback")
- if self.dm.sp < self.sp:
- raise TypeError("Attempt to roll back to invalid save point",
- self.sp, self.dm.sp)
- self.dm.sp = self.sp
- self.dm.delta = self.delta
Deleted: transaction/branches/sphinx/transaction/tests/SampleResourceManager.py
===================================================================
--- transaction/branches/sphinx/transaction/tests/SampleResourceManager.py 2012-12-17 20:28:55 UTC (rev 128707)
+++ transaction/branches/sphinx/transaction/tests/SampleResourceManager.py 2012-12-17 20:28:55 UTC (rev 128708)
@@ -1,103 +0,0 @@
-##############################################################################
-#
-# Copyright (c) 2004 Zope Foundation and Contributors.
-# All Rights Reserved.
-#
-# This software is subject to the provisions of the Zope Public License,
-# Version 2.1 (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.
-#
-##############################################################################
-"""Sample objects for use in tests
-
-Used by the 'resourcemanager' chapter in the Sphinx docs.
-"""
-
-class ResourceManager(object):
-
- def __init__(self):
- self.state = 0
- self.sp = 0
- self.transaction = None
- self.delta = 0
- self.txn_state = None
-
- def _check_state(self, *ok_states):
- if self.txn_state not in ok_states:
- raise ValueError("txn in state %r but expected one of %r" %
- (self.txn_state, ok_states))
-
- def _checkTransaction(self, transaction):
- if (transaction is not self.transaction
- and self.transaction is not None):
- raise TypeError("Transaction missmatch",
- transaction, self.transaction)
-
- def inc(self, n=1):
- self.delta += n
-
- def tpc_begin(self, transaction):
- self._checkTransaction(transaction)
- self._check_state(None)
- self.transaction = transaction
- self.txn_state = 'tpc_begin'
-
- def tpc_vote(self, transaction):
- self._checkTransaction(transaction)
- self._check_state('tpc_begin')
- self.state += self.delta
- self.txn_state = 'tpc_vote'
-
- def tpc_finish(self, transaction):
- self._checkTransaction(transaction)
- self._check_state('tpc_vote')
- self.delta = 0
- self.transaction = None
- self.prepared = False
- self.txn_state = None
-
- def tpc_abort(self, transaction):
- self._checkTransaction(transaction)
- if self.transaction is not None:
- self.transaction = None
-
- if self.txn_state == 'tpc_vote':
- self.state -= self.delta
-
- self.txn_state = None
- self.delta = 0
-
- def savepoint(self, transaction):
- if self.txn_state is not None:
- raise TypeError("Can't get savepoint during two-phase commit")
- self._checkTransaction(transaction)
- self.transaction = transaction
- self.sp += 1
- return SavePoint(self)
-
- def discard(self, transaction):
- pass
-
-
-class SavePoint(object):
-
- def __init__(self, rm):
- self.rm = rm
- self.sp = rm.sp
- self.delta = rm.delta
- self.transaction = rm.transaction
-
- def rollback(self):
- if self.transaction is not self.rm.transaction:
- raise TypeError("Attempt to rollback stale rollback")
- if self.rm.sp < self.sp:
- raise TypeError("Attempt to roll back to invalid save point",
- self.sp, self.rm.sp)
- self.rm.sp = self.sp
- self.rm.delta = self.delta
-
- def discard(self):
- pass
Added: transaction/branches/sphinx/transaction/tests/examples.py
===================================================================
--- transaction/branches/sphinx/transaction/tests/examples.py (rev 0)
+++ transaction/branches/sphinx/transaction/tests/examples.py 2012-12-17 20:28:55 UTC (rev 128708)
@@ -0,0 +1,181 @@
+##############################################################################
+#
+# Copyright (c) 2004 Zope Foundation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.1 (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.
+#
+##############################################################################
+"""Sample objects for use in tests
+
+"""
+
+
+class DataManager(object):
+ """Sample data manager
+
+ Used by the 'datamanager' chapter in the Sphinx docs.
+ """
+ def __init__(self):
+ self.state = 0
+ self.sp = 0
+ self.transaction = None
+ self.delta = 0
+ self.prepared = False
+
+ def inc(self, n=1):
+ self.delta += n
+
+ def prepare(self, transaction):
+ if self.prepared:
+ raise TypeError('Already prepared')
+ self._checkTransaction(transaction)
+ self.prepared = True
+ self.transaction = transaction
+ self.state += self.delta
+
+ def _checkTransaction(self, transaction):
+ if (transaction is not self.transaction
+ and self.transaction is not None):
+ raise TypeError("Transaction missmatch",
+ transaction, self.transaction)
+
+ def abort(self, transaction):
+ self._checkTransaction(transaction)
+ if self.transaction is not None:
+ self.transaction = None
+
+ if self.prepared:
+ self.state -= self.delta
+ self.prepared = False
+
+ self.delta = 0
+
+ def commit(self, transaction):
+ if not self.prepared:
+ raise TypeError('Not prepared to commit')
+ self._checkTransaction(transaction)
+ self.delta = 0
+ self.transaction = None
+ self.prepared = False
+
+ def savepoint(self, transaction):
+ if self.prepared:
+ raise TypeError("Can't get savepoint during two-phase commit")
+ self._checkTransaction(transaction)
+ self.transaction = transaction
+ self.sp += 1
+ return Rollback(self)
+
+
+class Rollback(object):
+
+ def __init__(self, dm):
+ self.dm = dm
+ self.sp = dm.sp
+ self.delta = dm.delta
+ self.transaction = dm.transaction
+
+ def rollback(self):
+ if self.transaction is not self.dm.transaction:
+ raise TypeError("Attempt to rollback stale rollback")
+ if self.dm.sp < self.sp:
+ raise TypeError("Attempt to roll back to invalid save point",
+ self.sp, self.dm.sp)
+ self.dm.sp = self.sp
+ self.dm.delta = self.delta
+
+
+class ResourceManager(object):
+ """ Sample resource manager.
+
+ Used by the 'resourcemanager' chapter in the Sphinx docs.
+ """
+ def __init__(self):
+ self.state = 0
+ self.sp = 0
+ self.transaction = None
+ self.delta = 0
+ self.txn_state = None
+
+ def _check_state(self, *ok_states):
+ if self.txn_state not in ok_states:
+ raise ValueError("txn in state %r but expected one of %r" %
+ (self.txn_state, ok_states))
+
+ def _checkTransaction(self, transaction):
+ if (transaction is not self.transaction
+ and self.transaction is not None):
+ raise TypeError("Transaction missmatch",
+ transaction, self.transaction)
+
+ def inc(self, n=1):
+ self.delta += n
+
+ def tpc_begin(self, transaction):
+ self._checkTransaction(transaction)
+ self._check_state(None)
+ self.transaction = transaction
+ self.txn_state = 'tpc_begin'
+
+ def tpc_vote(self, transaction):
+ self._checkTransaction(transaction)
+ self._check_state('tpc_begin')
+ self.state += self.delta
+ self.txn_state = 'tpc_vote'
+
+ def tpc_finish(self, transaction):
+ self._checkTransaction(transaction)
+ self._check_state('tpc_vote')
+ self.delta = 0
+ self.transaction = None
+ self.prepared = False
+ self.txn_state = None
+
+ def tpc_abort(self, transaction):
+ self._checkTransaction(transaction)
+ if self.transaction is not None:
+ self.transaction = None
+
+ if self.txn_state == 'tpc_vote':
+ self.state -= self.delta
+
+ self.txn_state = None
+ self.delta = 0
+
+ def savepoint(self, transaction):
+ if self.txn_state is not None:
+ raise TypeError("Can't get savepoint during two-phase commit")
+ self._checkTransaction(transaction)
+ self.transaction = transaction
+ self.sp += 1
+ return SavePoint(self)
+
+ def discard(self, transaction):
+ pass
+
+
+class SavePoint(object):
+
+ def __init__(self, rm):
+ self.rm = rm
+ self.sp = rm.sp
+ self.delta = rm.delta
+ self.transaction = rm.transaction
+
+ def rollback(self):
+ if self.transaction is not self.rm.transaction:
+ raise TypeError("Attempt to rollback stale rollback")
+ if self.rm.sp < self.sp:
+ raise TypeError("Attempt to roll back to invalid save point",
+ self.sp, self.rm.sp)
+ self.rm.sp = self.sp
+ self.rm.delta = self.delta
+
+ def discard(self):
+ pass
Modified: transaction/branches/sphinx/transaction/tests/test_transaction.py
===================================================================
--- transaction/branches/sphinx/transaction/tests/test_transaction.py 2012-12-17 20:28:55 UTC (rev 128707)
+++ transaction/branches/sphinx/transaction/tests/test_transaction.py 2012-12-17 20:28:55 UTC (rev 128708)
@@ -307,7 +307,7 @@
# The join method is provided for "backward-compatability" with ZODB 4
# data managers.
from transaction import Transaction
- from transaction.tests.SampleDataManager import DataManager
+ from transaction.tests.examples import DataManager
from transaction._transaction import DataManagerAdapter
# The argument to join must be a zodb4 data manager,
# transaction.interfaces.IDataManager.
@@ -469,23 +469,9 @@
HoserJar.committed += 1
-def hook():
- pass
-
-
def test_suite():
- from doctest import DocTestSuite
- suite = unittest.TestSuite((
- DocTestSuite(),
+ return unittest.TestSuite((
unittest.makeSuite(TransactionManagerTests),
unittest.makeSuite(Test_oid_repr),
unittest.makeSuite(MiscellaneousTests),
))
-
- return suite
-
-# additional_tests is for setuptools "setup.py test" support
-additional_tests = test_suite
-
-if __name__ == '__main__':
- unittest.TextTestRunner().run(test_suite())
More information about the checkins
mailing list