[Zodb-checkins] CVS: Zope/lib/python/ZODB/tests - IteratorStorage.py:1.12.6.2 StorageTestBase.py:1.17.6.3 testFileStorage.py:1.19.6.4 testTransaction.py:1.11.6.2
Barry Warsaw
barry@wooz.org
Thu, 30 Jan 2003 18:25:25 -0500
Update of /cvs-repository/Zope/lib/python/ZODB/tests
In directory cvs.zope.org:/tmp/cvs-serv23601/lib/python/ZODB/tests
Modified Files:
Tag: Zope-2_6-branch
IteratorStorage.py StorageTestBase.py testFileStorage.py
testTransaction.py
Log Message:
Port changes on the ZODB 3.1.1 release branch to Zope 2.6 branch for
the next beta.
=== Zope/lib/python/ZODB/tests/IteratorStorage.py 1.12.6.1 => 1.12.6.2 ===
--- Zope/lib/python/ZODB/tests/IteratorStorage.py:1.12.6.1 Wed Dec 18 16:59:18 2002
+++ Zope/lib/python/ZODB/tests/IteratorStorage.py Thu Jan 30 18:25:18 2003
@@ -5,7 +5,7 @@
"""
from ZODB.tests.MinPO import MinPO
-from ZODB.tests.StorageTestBase import zodb_unpickle
+from ZODB.tests.StorageTestBase import zodb_pickle, zodb_unpickle
from ZODB.utils import U64, p64
from ZODB.Transaction import Transaction
@@ -102,6 +102,27 @@
# None in the data attribute.
self.assertEqual(rec.oid, oid)
self.assertEqual(rec.data, None)
+
+ def checkIterationIntraTransaction(self):
+ # XXX try this test with logging enabled. If you see something like
+ #
+ # ZODB FS FS21 warn: FileStorageTests.fs truncated, possibly due to
+ # damaged records at 4
+ #
+ # Then the code in FileIterator.next() hasn't yet been fixed.
+ oid = self._storage.new_oid()
+ t = Transaction()
+ data = zodb_pickle(MinPO(0))
+ try:
+ self._storage.tpc_begin(t)
+ self._storage.store(oid, '\0'*8, data, '', t)
+ self._storage.tpc_vote(t)
+ # Don't do tpc_finish yet
+ it = self._storage.iterator()
+ for x in it:
+ pass
+ finally:
+ self._storage.tpc_finish(t)
class ExtendedIteratorStorage(IteratorCompare):
=== Zope/lib/python/ZODB/tests/StorageTestBase.py 1.17.6.2 => 1.17.6.3 ===
--- Zope/lib/python/ZODB/tests/StorageTestBase.py:1.17.6.2 Wed Dec 11 11:30:45 2002
+++ Zope/lib/python/ZODB/tests/StorageTestBase.py Thu Jan 30 18:25:19 2003
@@ -26,6 +26,7 @@
"""Create a pickle in the format expected by ZODB."""
f = StringIO()
p = Pickler(f, 1)
+ p.persistent_id = lambda obj: getattr(obj, '_p_oid', None)
klass = obj.__class__
assert not hasattr(obj, '__getinitargs__'), "not ready for constructors"
args = None
@@ -215,3 +216,4 @@
self._storage.tpc_vote(t)
self._storage.tpc_finish(t)
return oids
+
=== Zope/lib/python/ZODB/tests/testFileStorage.py 1.19.6.3 => 1.19.6.4 ===
--- Zope/lib/python/ZODB/tests/testFileStorage.py:1.19.6.3 Wed Dec 18 16:59:18 2002
+++ Zope/lib/python/ZODB/tests/testFileStorage.py Thu Jan 30 18:25:19 2003
@@ -67,7 +67,72 @@
else:
self.fail("expect long user field to raise error")
+ def check_use_fsIndex(self):
+ from ZODB.fsIndex import fsIndex
+
+ self.assertEqual(self._storage._index.__class__, fsIndex)
+
# XXX We could really use some tests for sanity checking
+
+ def check_conversion_to_fsIndex_not_if_readonly(self):
+
+ self.tearDown()
+
+ class OldFileStorage(ZODB.FileStorage.FileStorage):
+ def _newIndexes(self):
+ return {}, {}, {}, {}
+
+
+ from ZODB.fsIndex import fsIndex
+
+ # Hack FileStorage to create dictionary indexes
+ self._storage = OldFileStorage('FileStorageTests.fs')
+
+ self.assertEqual(type(self._storage._index), type({}))
+ for i in range(10):
+ self._dostore()
+
+ # Should save the index
+ self._storage.close()
+
+ self._storage = ZODB.FileStorage.FileStorage(
+ 'FileStorageTests.fs', read_only=1)
+ self.assertEqual(type(self._storage._index), type({}))
+
+ def check_conversion_to_fsIndex(self):
+
+ self.tearDown()
+
+ class OldFileStorage(ZODB.FileStorage.FileStorage):
+ def _newIndexes(self):
+ return {}, {}, {}, {}
+
+
+ from ZODB.fsIndex import fsIndex
+
+ # Hack FileStorage to create dictionary indexes
+ self._storage = OldFileStorage('FileStorageTests.fs')
+
+ self.assertEqual(type(self._storage._index), type({}))
+ for i in range(10):
+ self._dostore()
+
+ oldindex = self._storage._index.copy()
+
+ # Should save the index
+ self._storage.close()
+
+ self._storage = ZODB.FileStorage.FileStorage('FileStorageTests.fs')
+ self.assertEqual(self._storage._index.__class__, fsIndex)
+ self.failUnless(self._storage._used_index)
+
+ index = {}
+ for k, v in self._storage._index.items():
+ index[k] = v
+
+ self.assertEqual(index, oldindex)
+
+
def check_save_after_load_with_no_index(self):
for i in range(10):
self._dostore()
@@ -237,6 +302,7 @@
siter.close()
self._dst.tpc_vote(final)
self._dst.tpc_finish(final)
+
def test_suite():
suite = unittest.makeSuite(FileStorageTests, 'check')
=== Zope/lib/python/ZODB/tests/testTransaction.py 1.11.6.1 => 1.11.6.2 ===
--- Zope/lib/python/ZODB/tests/testTransaction.py:1.11.6.1 Tue Nov 12 16:13:58 2002
+++ Zope/lib/python/ZODB/tests/testTransaction.py Thu Jan 30 18:25:19 2003
@@ -477,37 +477,6 @@
else:
self.assertEqual(self.sub3._p_jar.cabort_sub, 1)
- # last test, check the hosing mechanism
-
- def testHoserStoppage(self):
- # It's hard to test the "hosed" state of the database, where
- # hosed means that a failure occurred in the second phase of
- # the two phase commit. It's hard because the database can
- # recover from such an error if it occurs during the very first
- # tpc_finish() call of the second phase.
-
- for obj in self.sub1, self.sub2:
- j = HoserJar(errors='tpc_finish')
- j.reset()
- obj._p_jar = j
- obj.modify(nojar=1)
-
- try:
- get_transaction().commit()
- except TestTxnException:
- pass
-
- self.assert_(Transaction.hosed)
-
- self.sub2.modify()
-
- try:
- get_transaction().commit()
- except Transaction.POSException.TransactionError:
- pass
- else:
- raise "Hosed Application didn't stop commits"
-
class DataObject: