Commit b7d086a9 authored by Barry Warsaw's avatar Barry Warsaw

Refactor some non-FileStorage specific tests into RecoveryStorage.

parent d411e289
......@@ -11,7 +11,7 @@ from ZODB.tests import StorageTestBase, BasicStorage, \
TransactionalUndoVersionStorage, PackableStorage, \
Synchronization, ConflictResolution, HistoryStorage, \
IteratorStorage, Corruption, RevisionStorage, PersistentStorage, \
MTStorage, ReadOnlyStorage
MTStorage, ReadOnlyStorage, RecoveryStorage
class FileStorageTests(
StorageTestBase.StorageTestBase,
......@@ -63,7 +63,7 @@ class FileStorageTests(
class FileStorageRecoveryTest(
StorageTestBase.StorageTestBase,
IteratorStorage.IteratorDeepCompare,
RecoveryStorage.RecoveryStorage,
):
def setUp(self):
......@@ -76,55 +76,6 @@ class FileStorageRecoveryTest(
StorageTestBase.removefs("Source.fs")
StorageTestBase.removefs("Dest.fs")
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 checkRecoverAbortVersion(self):
oid = self._storage.new_oid()
revid = self._dostore(oid, data=21, version="one")
revid = self._dostore(oid, revid=revid, data=23, version='one')
revid = self._dostore(oid, revid=revid, data=34, version='one')
# Now abort the version and the creation
t = Transaction()
self._storage.tpc_begin(t)
oids = self._storage.abortVersion('one', t)
self._storage.tpc_vote(t)
self._storage.tpc_finish(t)
self.assertEqual(oids, [oid])
self._dst.copyTransactionsFrom(self._storage)
self.compare(self._storage, self._dst)
# Also make sure the the last transaction has a data record
# with None for its data attribute, because we've undone the
# object.
for s in self._storage, self._dst:
iter = s.iterator()
for trans in iter:
pass # iterate until we get the last one
data = trans[0]
self.assertRaises(IndexError, lambda i:trans[i], 1)
self.assertEqual(data.oid, oid)
self.assertEqual(data.data, None)
def test_suite():
suite = unittest.makeSuite(FileStorageTests, 'check')
......
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