[Zope3-checkins] CVS: Zope3/src/zodb/storage/tests - undo.py:1.11.2.1
Jeremy Hylton
jeremy@zope.com
Wed, 16 Apr 2003 14:14:21 -0400
Update of /cvs-repository/Zope3/src/zodb/storage/tests
In directory cvs.zope.org:/tmp/cvs-serv6066/tests
Modified Files:
Tag: jeremy-new-pack-branch
undo.py
Log Message:
Add undo/pack test.
Covers case where an object is only reachable at the pack time because
of an undo after the pack time.
=== Zope3/src/zodb/storage/tests/undo.py 1.11 => 1.11.2.1 ===
--- Zope3/src/zodb/storage/tests/undo.py:1.11 Thu Apr 10 03:27:30 2003
+++ Zope3/src/zodb/storage/tests/undo.py Wed Apr 16 14:14:20 2003
@@ -21,6 +21,14 @@
class C(Persistent):
pass
+def listeq(L1, L2):
+ """Return True if L1.sort() == L2.sort()"""
+ c1 = L1[:]
+ c2 = L2[:]
+ c1.sort()
+ c2.sort()
+ return c1 == c2
+
class TransactionalUndoStorage:
def _transaction_begin(self):
@@ -389,7 +397,6 @@
eq(zodb_unpickle(data), MinPO(54))
self._iterate()
-
def testNotUndoable(self):
eq = self.assertEqual
# Set things up so we've got a transaction that can't be undone
@@ -540,6 +547,62 @@
eq(o1.obj, o2)
eq(o1.obj.obj, o3)
self._iterate()
+
+ def testPackAfterUndoDeletion(self):
+ db = DB(self._storage)
+ cn = db.open()
+ root = cn.root()
+
+ pack_times = []
+ def set_pack_time():
+ snooze()
+ pack_times.append(time.time())
+
+ root["key0"] = MinPO(0)
+ root["key1"] = MinPO(1)
+ root["key2"] = MinPO(2)
+ txn = get_transaction()
+ txn.note("create 3 keys")
+ txn.commit()
+
+ set_pack_time()
+
+ del root["key1"]
+ txn = get_transaction()
+ txn.note("delete 1 key")
+ txn.commit()
+
+ set_pack_time()
+
+ root._p_deactivate()
+ cn.sync()
+ self.assert_(listeq(root.keys(), ["key0", "key2"]))
+
+ L = db.undoInfo()
+ db.undo(L[0]["id"])
+ txn = get_transaction()
+ txn.note("undo deletion")
+ txn.commit()
+
+ set_pack_time()
+
+ root._p_deactivate()
+ cn.sync()
+ self.assert_(listeq(root.keys(), ["key0", "key1", "key2"]))
+
+ for t in pack_times:
+ self._storage.pack(t)
+
+ from zodb.storage.fsdump import dump
+ dump(self._storage._name)
+
+ root._p_deactivate()
+ cn.sync()
+ self.assert_(listeq(root.keys(), ["key0", "key1", "key2"]))
+ for i in range(3):
+ obj = root["key%d" % i]
+ self.assertEqual(obj.value, i)
+ print root.items()
def testTransactionalUndoIterator(self):
# check that data_txn set in iterator makes sense