Commit 6b525a6e authored by Tres Seaver's avatar Tres Seaver

Consolidate example code.

parent de6a7506
......@@ -6,9 +6,9 @@ Simple Data Manager
.. 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.
......
......@@ -6,9 +6,9 @@ Simple Resource Manager
.. 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.
......
##############################################################################
#
# 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
......@@ -13,11 +13,89 @@
##############################################################################
"""Sample objects for use in tests
Used by the 'resourcemanager' chapter in the Sphinx docs.
"""
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
......
......@@ -307,7 +307,7 @@ class MiscellaneousTests(unittest.TestCase):
# 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 @@ class HoserJar(BasicJar):
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())
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment