[Zodb-checkins] CVS: Zope3/src/zodb/storage/tests -
test_recover.py:1.2.2.1 base.py:1.15.14.1 basic.py:1.9.20.1
synchronization.py:1.6.20.1 undo.py:1.16.2.1 test_config.py:NONE
Grégoire Weber
zope at i-con.ch
Sun Jun 22 11:24:01 EDT 2003
Update of /cvs-repository/Zope3/src/zodb/storage/tests
In directory cvs.zope.org:/tmp/cvs-serv24874/src/zodb/storage/tests
Modified Files:
Tag: cw-mail-branch
base.py basic.py synchronization.py undo.py
Added Files:
Tag: cw-mail-branch
test_recover.py
Removed Files:
Tag: cw-mail-branch
test_config.py
Log Message:
Synced up with HEAD
=== Added File Zope3/src/zodb/storage/tests/test_recover.py ===
##############################################################################
#
# Copyright (c) 2003 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.
#
##############################################################################
"""Tests of the file storage recovery script."""
import base64
import os
import random
import sys
import tempfile
import unittest
import StringIO
from zodb.db import DB
from zodb.storage.file import FileStorage
from zodb.storage.file.recover import recover
from zodb.storage.file.dump import Dumper
from persistence.dict import PersistentDict
from transaction import get_transaction
class RecoverTest(unittest.TestCase):
level = 2
path = None
def setUp(self):
self.path = tempfile.mktemp(suffix=".fs")
self.storage = FileStorage(self.path)
self.populate()
self.dest = tempfile.mktemp(suffix=".fs")
self.recovered = None
def tearDown(self):
self.storage.close()
if self.recovered is not None:
self.recovered.close()
self.storage.cleanup()
## removefs(self.dest)
def populate(self):
db = DB(self.storage)
cn = db.open()
rt = cn.root()
# create a whole bunch of objects,
# looks like a Data.fs > 1MB
for i in range(50):
d = rt[i] = PersistentDict()
get_transaction().commit()
for j in range(50):
d[j] = "a" * j
get_transaction().commit()
def damage(self, num, size):
self.storage.close()
# Drop size null bytes into num random spots.
for i in range(num):
offset = random.randint(0, self.storage._pos - size)
f = open(self.path, "a+b")
f.seek(offset)
f.write("\0" * size)
f.close()
ITERATIONS = 5
def recover(self, source, dest):
orig = sys.stdout
try:
sys.stdout = StringIO.StringIO()
try:
recover(self.path, self.dest,
verbose=0, partial=1, force=0, pack=1)
except SystemExit:
raise RuntimeError, "recover tried to exit"
finally:
sys.stdout = orig
def testOneBlock(self):
for i in range(self.ITERATIONS):
self.damage(1, 1024)
self.recover(self.path, self.dest)
self.recovered = FileStorage(self.dest)
self.recovered.close()
os.remove(self.path)
os.rename(self.dest, self.path)
def testFourBlocks(self):
for i in range(self.ITERATIONS):
self.damage(4, 512)
self.recover(self.path, self.dest)
self.recovered = FileStorage(self.dest)
self.recovered.close()
os.remove(self.path)
os.rename(self.dest, self.path)
def testBigBlock(self):
for i in range(self.ITERATIONS):
self.damage(1, 32 * 1024)
self.recover(self.path, self.dest)
self.recovered = FileStorage(self.dest)
self.recovered.close()
os.remove(self.path)
os.rename(self.dest, self.path)
def testBadTransaction(self):
# Find transaction headers and blast them.
L = self.storage.undoLog()
r = L[3]
tid = base64.decodestring(r["id"] + "\n")
pos1 = self.storage._txn_find(tid, 0)
r = L[8]
tid = base64.decodestring(r["id"] + "\n")
pos2 = self.storage._txn_find(tid, 0)
self.storage.close()
# Overwrite the entire header.
f = open(self.path, "a+b")
f.seek(pos1 - 50)
f.write("\0" * 100)
f.close()
self.recover(self.path, self.dest)
self.recovered = FileStorage(self.dest)
self.recovered.close()
os.remove(self.path)
os.rename(self.dest, self.path)
# Overwrite part of the header.
f = open(self.path, "a+b")
f.seek(pos2 + 10)
f.write("\0" * 100)
f.close()
self.recover(self.path, self.dest)
self.recovered = FileStorage(self.dest)
self.recovered.close()
def test_suite():
return unittest.makeSuite(RecoverTest)
=== Zope3/src/zodb/storage/tests/base.py 1.15 => 1.15.14.1 ===
--- Zope3/src/zodb/storage/tests/base.py:1.15 Tue Apr 22 11:23:12 2003
+++ Zope3/src/zodb/storage/tests/base.py Sun Jun 22 10:22:29 2003
@@ -119,7 +119,7 @@
obj._p_oid = self.newObjectId()
-class StorageTestBase(unittest.TestCase):
+class StorageTestBase(unittest.TestCase, object):
# XXX It would be simpler if concrete tests didn't need to extend
# setUp() and tearDown().
=== Zope3/src/zodb/storage/tests/basic.py 1.9 => 1.9.20.1 ===
--- Zope3/src/zodb/storage/tests/basic.py:1.9 Thu Mar 20 18:01:41 2003
+++ Zope3/src/zodb/storage/tests/basic.py Sun Jun 22 10:22:29 2003
@@ -197,3 +197,20 @@
self._storage.store(oid, None, data, refs, '', t)
self._storage.tpcVote(t)
self._storage.tpcFinish(t)
+
+ def testNote(self):
+ oid = self._storage.newObjectId()
+ t = Transaction()
+ self._storage.tpcBegin(t)
+ t.note('this is a test')
+ data, refs = zodb_pickle(MinPO(5))
+ self._storage.store(oid, ZERO, data, refs, '', t)
+ self._storage.tpcVote(t)
+ self._storage.tpcFinish(t)
+
+ def testGetExtensionMethods(self):
+ m = self._storage.getExtensionMethods()
+ self.assertEqual(type(m),type({}))
+ for k,v in m.items():
+ self.assertEqual(v,None)
+ self.assert_(callable(getattr(self._storage,k)))
=== Zope3/src/zodb/storage/tests/synchronization.py 1.6 => 1.6.20.1 ===
--- Zope3/src/zodb/storage/tests/synchronization.py:1.6 Thu Mar 20 18:01:41 2003
+++ Zope3/src/zodb/storage/tests/synchronization.py Sun Jun 22 10:22:29 2003
@@ -76,7 +76,7 @@
def verifyNotCommitting(self, callable, *args):
args = (StorageTransactionError, callable) + args
- apply(self.assertRaises, args)
+ self.assertRaises(*args)
def verifyWrongTrans(self, callable, *args):
t = Transaction()
=== Zope3/src/zodb/storage/tests/undo.py 1.16 => 1.16.2.1 ===
--- Zope3/src/zodb/storage/tests/undo.py:1.16 Fri May 16 17:36:57 2003
+++ Zope3/src/zodb/storage/tests/undo.py Sun Jun 22 10:22:29 2003
@@ -156,6 +156,27 @@
eq(zodb_unpickle(data), MinPO(23))
self._iterate()
+## def testCreationUndoneGetSerial(self):
+## # XXX Do we really want to change FileStorage to make this test
+## # pass? getSerial() is nice and fast right now.
+
+## # create an object
+## oid = self._storage.newObjectId()
+## revid = self._dostore(oid, data=MinPO(23))
+## print repr(revid)
+## # undo its creation
+## info = self._storage.undoInfo()
+## tid = info[0]['id']
+## t = Transaction()
+## t.note('undo1')
+## self._storage.tpcBegin(t)
+## oids = self._storage.undo(tid, t)
+## self._storage.tpcVote(t)
+## self._storage.tpcFinish(t)
+## # Check that calling getSerial on an uncreated object raises a KeyError
+## # The current version of FileStorage fails this test
+## self.assertRaises(KeyError, self._storage.getSerial, oid)
+
def testUndoCreationBranch1(self):
eq = self.assertEqual
oid = self._storage.newObjectId()
@@ -766,3 +787,76 @@
break
else:
self.fail('transaction not found')
+
+ def testPackUndoLog(self):
+ self._initroot()
+ eq = self.assertEqual
+ raises = self.assertRaises
+ # Create a `persistent' object
+ obj = self._newobj()
+ obj.value = 1
+ oid = obj._p_oid
+ # Commit two different revisions
+ revid1 = self._dostore(oid, data=obj)
+ obj.value = 2
+ snooze()
+ packtime = time.time()
+ snooze()
+ revid2 = self._dostore(oid, revid=revid1, data=obj)
+ # Now pack the first transaction
+ self.assertEqual(3, len(self._storage.undoLog()))
+ self._storage.pack(packtime)
+ # The undo log contains only the most resent transaction
+ self.assertEqual(1, len(self._storage.undoLog()))
+
+ def dont_testPackUndoLogUndoable(self):
+ # XXX This test was copied from ZODB3, but no effort was made
+ # to convert the code to make it work in ZODB4.
+
+ # A disabled test. I wanted to test that the content of the
+ # undo log was consistent, but every storage appears to
+ # include something slightly different. If the result of this
+ # method is only used to fill a GUI then this difference
+ # doesnt matter. Perhaps re-enable this test once we agree
+ # what should be asserted.
+
+ self._initroot()
+ # Create two `persistent' object
+ obj1 = self._newobj()
+ oid1 = obj1.getoid()
+ obj1.value = 1
+ obj2 = self._newobj()
+ oid2 = obj2.getoid()
+ obj2.value = 2
+
+ # Commit the first revision of each of them
+ revid11 = self._dostoreNP(oid1, data=pickle.dumps(obj1),
+ description="1-1")
+ revid22 = self._dostoreNP(oid2, data=pickle.dumps(obj2),
+ description="2-2")
+
+ # remember the time. everything above here will be packed away
+ snooze()
+ packtime = time.time()
+ snooze()
+ # Commit two revisions of the first object
+ obj1.value = 3
+ revid13 = self._dostoreNP(oid1, revid=revid11,
+ data=pickle.dumps(obj1), description="1-3")
+ obj1.value = 4
+ revid14 = self._dostoreNP(oid1, revid=revid13,
+ data=pickle.dumps(obj1), description="1-4")
+ # Commit one revision of the second object
+ obj2.value = 5
+ revid25 = self._dostoreNP(oid2, revid=revid22,
+ data=pickle.dumps(obj2), description="2-5")
+ # Now pack
+ self.assertEqual(6,len(self._storage.undoLog()))
+ print '\ninitial undoLog was'
+ for r in self._storage.undoLog(): print r
+ self._storage.pack(packtime, referencesf)
+ # The undo log contains only two undoable transaction.
+ print '\nafter packing undoLog was'
+ for r in self._storage.undoLog(): print r
+ # what can we assert about that?
+
=== Removed File Zope3/src/zodb/storage/tests/test_config.py ===
More information about the Zodb-checkins
mailing list