[Zodb-checkins] CVS: Zope3/lib/python/ZODB/tests - TransactionalUndoDB.py:1.1

[Jeremy Hylton]

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

Added Files:
	TransactionalUndoDB.py 
Log Message:
First step towards tests of storage features using the DB.


=== Added File Zope3/lib/python/ZODB/tests/TransactionalUndoDB.py === (468/568 lines abridged)
"""Check transactional undo performed via the database."""

import time
import types

from ZODB import POSException
from ZODB.ZTransaction import Transaction
from ZODB.utils import u64, p64, z64
from ZODB.DB import DB

from ZODB.tests.MinPO import MinPO
from ZODB.tests.StorageTestBase import zodb_pickle, zodb_unpickle

from Persistence import Persistent
from Transaction import get_transaction

class C(Persistent):
    pass

class TransactionalUndoDB:

    def checkSimpleTransactionalUndo(self):
        obj = MinPO(23)
        self._root["obj"] = obj
        get_transaction().note("23")
        get_transaction().commit()

        obj.value = 24
        get_transaction().note("24")
        get_transaction().commit()
        
        obj.value = 25
        get_transaction().note("25")
        get_transaction().commit()

        info = self._db.undoInfo()
        tid = info[0]['id']
        self._db.undo(tid)
        get_transaction().commit()
        self._conn.sync()
        self.assertEqual(obj.value, 24)

        info = self._db.undoInfo()
        tid = info[2]['id']
        self._db.undo(tid)
        get_transaction().commit()
        self._conn.sync()
        self.assertEqual(obj.value, 23)

        # Undo object creation

[-=- -=- -=- 468 lines omitted -=- -=- -=-]

                tid = info[base + j]['id']
                s.transactionalUndo(tid, t)
            s.tpc_vote(t)
            s.tpc_finish(t)
        
        for i in range(BATCHES):
            undo(i)

        # There are now (2 + OBJECTS) * BATCHES transactions:
        #     BATCHES original transactions, followed by
        #     OBJECTS * BATCHES modifications, followed by
        #     BATCHES undos

        iter = s.iterator()
        offset = 0

        eq = self.assertEqual

        for i in range(BATCHES):
            txn = iter[offset]
            offset += 1
            
            tid = p64(i + 1)
            eq(txn.tid, tid)

            L1 = [(rec.oid, rec.serial, rec.data_txn) for rec in txn]
            L2 = [(oid, revid, None) for _tid, oid, revid in orig
                  if _tid == tid]
            
            eq(L1, L2)

        for i in range(BATCHES * OBJECTS):
            txn = iter[offset]
            offset += 1
            eq(len([rec for rec in txn if rec.data_txn is None]), 1)

        for i in range(BATCHES):
            txn = iter[offset]
            offset += 1

            # The undos are performed in reverse order.
            otid = p64(BATCHES - i)
            L1 = [(rec.oid, rec.data_txn) for rec in txn]
            L2 = [(oid, otid) for _tid, oid, revid in orig
                  if _tid == otid]
            L1.sort()
            L2.sort()
            eq(L1, L2)

        self.assertRaises(IndexError, iter.__getitem__, offset)