[Zope3-checkins] CVS: Zope3/src/transaction/tests - test_SampleDataManager.py:1.1.2.1

[Fred L. Drake, Jr.]

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

Added Files:
      Tag: zope3-zodb3-devel-branch
	test_SampleDataManager.py 
Log Message:
checking in so jim can edit


=== Added File Zope3/src/transaction/tests/test_SampleDataManager.py ===
##############################################################################
#
# 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.
#
##############################################################################
"""Sample objects for use in tests

$Id: test_SampleDataManager.py,v 1.1.2.1 2004/01/22 16:32:20 fdrake Exp $
"""

class DataManager(object):
    """Sample data manager

    This class provides a trivial data-manager implementation and doc
    strings to illustrate the the protocol and to provide a tool for
    writing tests.

    Our sample data manager has state that is updated through an inc
    method and through transaction operations.

    When we create a sample data manager:

    >>> dm = DataManager()

    It has two bits of state, state:

    >>> dm.state
    0

    and delta:

    >>> dm.delta
    0

    Both of which are initialized to 0.  state is meant to model
    committed state, while delta represents tentative changes within a
    transaction.  We change the state by calling inc:

    >>> dm.inc()

    which updates delta:

    >>> dm.delta
    1

    but state isn't changed until we commit the transaction:

    >>> dm.state
    0

    To commit the changes, we use 2-phase commit. We execute the first
    stage by calling prepare.  We need to pass a transation. Our
    sample data managers don't really use the transactions for much,
    so we'll be lazy and use integers for transactions:

    >>> t1 = 1
    >>> dm.prepare(t1)

    The sample data manager updates the state when we call prepare:

    >>> dm.state
    1
    >>> dm.delta
    1

    This is mainly so we can detect some affect of calling the methods.

    Now if we call commit:

    >>> dm.commit(t1)

    Our changes are"permanent".  The state reflects the changes and the
    delta has been reset to 0. 

    >>> dm.state
    1
    >>> dm.delta
    0
    """

    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):
        """Prepare to commit data

        >>> dm = DataManager()
        >>> dm.inc()
        >>> dm.prepare(1)
        >>> dm.commit(1)
        >>> dm.state
        1
        >>> dm.inc()
        >>> dm.prepare(2)
        >>> dm.abort(2)
        >>> dm.state
        1

        It is en error to call prepare more than once without an intervening
        commit or abort:

        >>> dm.prepare(1)
Traceback (most recent call last):

TypeError: Already prepared

        >>> dm.prepare(1)
        >>> dm.prepare(2)

        >>> dm.abort(1)

        If there was a preceeding savepoint, the transaction must match:

        >>> rollback = dm.savepoint(1)
        >>> rollback.prepare(2)

        >>> rollback.prepare(1)
        
        """
        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):
        """Abort a transaction

        """
        self._checkTransaction(transaction)
        if self.transaction is not None:
            self.state -= self.delta
            self.transaction = None
        self.delta = 0
        self.prepared = False

    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.state = dm.state
        self.transaction = dm.transaction

    def rollback(self):
        if self.transaction is not self.sp.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.state = self.state
        self.dm.delta = 0


def test_suite():
    import unittest
    from doctest import DocTestSuite
    return unittest.TestSuite((
        DocTestSuite(),
        ))

if __name__ == '__main__': unittest.main()