Commit 2b15f06b authored by Jeremy Hylton's avatar Jeremy Hylton

Remove variant copy testTransactionalUndoIterator that depended on

ZODB4.
parent 040dc5a2
...@@ -659,95 +659,6 @@ class TransactionalUndoStorage: ...@@ -659,95 +659,6 @@ class TransactionalUndoStorage:
self.assertEqual(rt["test"].value, 3) self.assertEqual(rt["test"].value, 3)
self.assertEqual(rt["test2"].value, 2) self.assertEqual(rt["test2"].value, 2)
def testTransactionalUndoIterator(self):
# check that data_txn set in iterator makes sense
if not hasattr(self._storage, "iterator"):
return
s = self._storage
BATCHES = 4
OBJECTS = 4
orig = []
for i in range(BATCHES):
t = Transaction()
tid = p64(i + 1)
s.tpcBegin(t, tid)
for j in range(OBJECTS):
oid = s.newObjectId()
obj = MinPO(i * OBJECTS + j)
data, refs = zodb_pickle(obj)
revid = s.store(oid, None, data, refs, '', t)
orig.append((tid, oid, revid))
s.tpcVote(t)
s.tpcFinish(t)
i = 0
for tid, oid, revid in orig:
self._dostore(oid, revid=revid, data=MinPO(revid),
description="update %s" % i)
# Undo the OBJECTS transactions that modified objects created
# in the ith original transaction.
def undo(i):
info = s.undoInfo()
t = Transaction()
s.tpcBegin(t)
base = i * OBJECTS + i
for j in range(OBJECTS):
tid = info[base + j]['id']
s.undo(tid, t)
s.tpcVote(t)
s.tpcFinish(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
fsiter = iter(s.iterator())
offset = 0
eq = self.assertEqual
for i in range(BATCHES):
txn = fsiter.next()
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 = fsiter.next()
offset += 1
eq(len([rec for rec in txn if rec.data_txn is None]), 1)
for i in range(BATCHES):
txn = fsiter.next()
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(StopIteration, fsiter.next)
def checkTransactionalUndoIterator(self): def checkTransactionalUndoIterator(self):
# check that data_txn set in iterator makes sense # check that data_txn set in iterator makes sense
if not hasattr(self._storage, "iterator"): if not hasattr(self._storage, "iterator"):
......
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