[Zodb-checkins] CVS: StandaloneZODB/ZODB/tests - testFileStorage.py:1.14.4.1

[Barry Warsaw]

Update of /cvs-repository/StandaloneZODB/ZODB/tests
In directory cvs.zope.org:/tmp/cvs-serv13045

Modified Files:
      Tag: Recovery
	testFileStorage.py 
Log Message:
New class FileStorageRecoveryTest, which adds two simple tests of
copyTransactionsFrom().  This indirectly tests the new recovery()
method.  (But we should probably add some explicit tests of that
method).


=== StandaloneZODB/ZODB/tests/testFileStorage.py 1.14 => 1.14.4.1 ===
 import sys, os, unittest
+import errno
+from ZODB.Transaction import Transaction
 
 from ZODB.tests import StorageTestBase, BasicStorage, \
      TransactionalUndoStorage, VersionStorage, \
@@ -45,10 +47,56 @@
             if os.path.exists(path):
                 os.remove(path)
 
+class FileStorageRecoveryTest(
+    StorageTestBase.StorageTestBase,
+    IteratorStorage.IteratorDeepCompare,
+    ):
+
+    def setUp(self):
+        StorageTestBase.StorageTestBase.setUp(self)
+        self._storage = ZODB.FileStorage.FileStorage('Source.fs')
+        self._dst = ZODB.FileStorage.FileStorage('Dest.fs')
+
+    def tearDown(self):
+        StorageTestBase.StorageTestBase.tearDown(self)
+        self._dst.close()
+        for ext in '', '.old', '.tmp', '.lock', '.index':
+            for fs in 'Source', 'Dest':
+                path = fs + '.fs' + ext
+                try:
+                    os.remove(path)
+                except OSError, e:
+                    if e.errno <> errno.ENOENT: raise
+
+    def checkSimpleRecovery(self):
+        oid = self._storage.new_oid()
+        revid = self._dostore(oid, data=11)
+        revid = self._dostore(oid, revid=revid, data=12)
+        revid = self._dostore(oid, revid=revid, data=13)
+        self._dst.copyTransactionsFrom(self._storage)
+        self.compare(self._storage, self._dst)
+
+    def checkRecoveryAcrossVersions(self):
+        oid = self._storage.new_oid()
+        revid = self._dostore(oid, data=21)
+        revid = self._dostore(oid, revid=revid, data=22)
+        revid = self._dostore(oid, revid=revid, data=23, version='one')
+        revid = self._dostore(oid, revid=revid, data=34, version='one')
+        # Now commit the version
+        t = Transaction()
+        self._storage.tpc_begin(t)
+        self._storage.commitVersion('one', '', t)
+        self._storage.tpc_vote(t)
+        self._storage.tpc_finish(t)
+        self._dst.copyTransactionsFrom(self._storage)
+        self.compare(self._storage, self._dst)
+
 def test_suite():
     suite = unittest.makeSuite(FileStorageTests, 'check')
     suite2 = unittest.makeSuite(Corruption.FileStorageCorruptTests, 'check')
+    suite3 = unittest.makeSuite(FileStorageRecoveryTest, 'check')
     suite.addTest(suite2)
+    suite.addTest(suite3)
     return suite
 
 def main():