[Zodb-checkins] CVS: Zope/lib/python/ZODB/tests - testConfig.py:1.3.2.1 IteratorStorage.py:1.12.4.2 testFileStorage.py:1.19.4.3 testStorageConfig.py:1.4.2.3
Chris McDonough
chrism@zope.com
Sat, 4 Jan 2003 13:16:29 -0500
Update of /cvs-repository/Zope/lib/python/ZODB/tests
In directory cvs.zope.org:/tmp/cvs-serv3465/tests
Modified Files:
Tag: chrism-install-branch
IteratorStorage.py testFileStorage.py testStorageConfig.py
Added Files:
Tag: chrism-install-branch
testConfig.py
Log Message:
Merge with HEAD.
=== Added File Zope/lib/python/ZODB/tests/testConfig.py ===
##############################################################################
#
# Copyright (c) 2001, 2002 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.
#
##############################################################################
import os
import errno
import shutil
import tempfile
import unittest
import ZODB.config
import ZODB.tests
from ZODB.POSException import ReadOnlyError
from ZEO.ClientStorage import ClientDisconnected
class ConfigTestBase(unittest.TestCase):
def _opendb(self, s):
return ZODB.config.databaseFromString(s)
def _test(self, s):
db = self._opendb(s)
# Do something with the database to make sure it works
cn = db.open()
rt = cn.root()
rt["test"] = 1
get_transaction().commit()
db.close()
class ZODBConfigTest(ConfigTestBase):
def test_map_config1(self):
self._test("<mappingstorage/>")
def test_map_config2(self):
self._test(
"""<mappingstorage/>
cache_size 1000
""")
def test_file_config1(self):
path = tempfile.mktemp()
self._test(
"""<filestorage>
path %s
</filestorage>
""" % path)
os.unlink(path)
def test_file_config2(self):
path = tempfile.mktemp()
cfg = """
<filestorage>
path %s
create false
read_only true
</filestorage>
""" % path
self.assertRaises(ReadOnlyError, self._test, cfg)
def test_zeo_config(self):
cfg = """
<zeoclient>
server /no/path/var/test/foo
wait false
</zeoclient>
"""
self.assertRaises(ClientDisconnected, self._test, cfg)
class BDBConfigTest(ConfigTestBase):
def setUp(self):
self._path = tempfile.mktemp()
try:
os.mkdir(self._path)
except OSError, e:
if e.errno <> errno.EEXIST:
raise
def tearDown(self):
shutil.rmtree(self._path)
def test_bdbfull_simple(self):
cfg = """
<fullstorage>
name %s
</fullstorage>
""" % self._path
self._test(cfg)
def test_bdbminimal_simple(self):
cfg = """
<minimalstorage>
name %s
</minimalstorage>
""" % self._path
self._test(cfg)
def test_suite():
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(ZODBConfigTest))
# Only run the Berkeley tests if they are available
import BDBStorage
if BDBStorage.is_available:
suite.addTest(unittest.makeSuite(BDBConfigTest))
return suite
if __name__ == '__main__':
unittest.main(defaultTest='test_suite')
=== Zope/lib/python/ZODB/tests/IteratorStorage.py 1.12.4.1 => 1.12.4.2 ===
--- Zope/lib/python/ZODB/tests/IteratorStorage.py:1.12.4.1 Fri Jan 3 01:37:26 2003
+++ Zope/lib/python/ZODB/tests/IteratorStorage.py Sat Jan 4 13:16:25 2003
@@ -38,6 +38,7 @@
eq(zodb_unpickle(rec.data), MinPO(val))
val = val + 1
eq(val, val0 + len(revids))
+ txniter.close()
class IteratorStorage(IteratorCompare):
@@ -191,3 +192,5 @@
# they were the same length
self.assertRaises(IndexError, iter1.next)
self.assertRaises(IndexError, iter2.next)
+ iter1.close()
+ iter2.close()
=== Zope/lib/python/ZODB/tests/testFileStorage.py 1.19.4.2 => 1.19.4.3 ===
--- Zope/lib/python/ZODB/tests/testFileStorage.py:1.19.4.2 Fri Jan 3 01:37:26 2003
+++ Zope/lib/python/ZODB/tests/testFileStorage.py Sat Jan 4 13:16:25 2003
@@ -72,6 +72,101 @@
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()
+ self._storage.close()
+ os.remove('FileStorageTests.fs.index')
+ self.open()
+ self.assertEqual(self._storage._saved, 1)
+
+
+ # This would make the unit tests too slow
+ # check_save_after_load_that_worked_hard(self)
+
+ def check_periodic_save_index(self):
+
+ # Check the basic algorithm
+ oldsaved = self._storage._saved
+ self._storage._records_before_save = 10
+ for i in range(4):
+ self._dostore()
+ self.assertEqual(self._storage._saved, oldsaved)
+ self._dostore()
+ self.assertEqual(self._storage._saved, oldsaved+1)
+
+ # Now make sure the parameter changes as we get bigger
+ for i in range(20):
+ self._dostore()
+
+ self.failUnless(self._storage._records_before_save > 20)
+
class FileStorageRecoveryTest(
StorageTestBase.StorageTestBase,
RecoveryStorage.RecoveryStorage,
@@ -93,63 +188,6 @@
StorageTestBase.removefs('Dest.fs')
return ZODB.FileStorage.FileStorage('Dest.fs')
- def checkRecoverUndoInVersion(self):
- oid = self._storage.new_oid()
- version = "aVersion"
- revid_a = self._dostore(oid, data=MinPO(91))
- revid_b = self._dostore(oid, revid=revid_a, version=version,
- data=MinPO(92))
- revid_c = self._dostore(oid, revid=revid_b, version=version,
- data=MinPO(93))
- self._undo(self._storage.undoInfo()[0]['id'], oid)
- self._commitVersion(version, '')
- self._undo(self._storage.undoInfo()[0]['id'], oid)
-
- # now copy the records to a new storage
- self._dst.copyTransactionsFrom(self._storage)
- self.compare(self._storage, self._dst)
-
- # The last two transactions were applied directly rather than
- # copied. So we can't use compare() to verify that they new
- # transactions are applied correctly. (The new transactions
- # will have different timestamps for each storage.)
-
- self._abortVersion(version)
- self.assert_(self._storage.versionEmpty(version))
- self._undo(self._storage.undoInfo()[0]['id'], oid)
- self.assert_(not self._storage.versionEmpty(version))
-
- # check the data is what we expect it to be
- data, revid = self._storage.load(oid, version)
- self.assertEqual(zodb_unpickle(data), MinPO(92))
- data, revid = self._storage.load(oid, '')
- self.assertEqual(zodb_unpickle(data), MinPO(91))
-
- # and swap the storages
- tmp = self._storage
- self._storage = self._dst
- self._abortVersion(version)
- self.assert_(self._storage.versionEmpty(version))
- self._undo(self._storage.undoInfo()[0]['id'], oid)
- self.assert_(not self._storage.versionEmpty(version))
-
- # check the data is what we expect it to be
- data, revid = self._storage.load(oid, version)
- self.assertEqual(zodb_unpickle(data), MinPO(92))
- data, revid = self._storage.load(oid, '')
- self.assertEqual(zodb_unpickle(data), MinPO(91))
-
- # swap them back
- self._storage = tmp
-
- # Now remove _dst and copy all the transactions a second time.
- # This time we will be able to confirm via compare().
- self._dst.close()
- StorageTestBase.removefs("Dest.fs")
- self._dst = ZODB.FileStorage.FileStorage('Dest.fs')
- self._dst.copyTransactionsFrom(self._storage)
- self.compare(self._storage, self._dst)
-
def test_suite():
suite = unittest.makeSuite(FileStorageTests, 'check')
=== Zope/lib/python/ZODB/tests/testStorageConfig.py 1.4.2.2 => 1.4.2.3 ===
--- Zope/lib/python/ZODB/tests/testStorageConfig.py:1.4.2.2 Fri Jan 3 01:37:26 2003
+++ Zope/lib/python/ZODB/tests/testStorageConfig.py Sat Jan 4 13:16:25 2003
@@ -17,12 +17,11 @@
import unittest
from StringIO import StringIO
-import ZConfig
+import ZConfig.Context
from ZODB import StorageConfig
-class StorageTestCase(unittest.TestCase):
-
+class StorageTestBase(unittest.TestCase):
def setUp(self):
unittest.TestCase.setUp(self)
self.tmpfn = tempfile.mktemp()
@@ -46,6 +45,13 @@
except os.error:
pass
+ def loadConfigText(self, text):
+ context = ZConfig.Context.Context()
+ io = StringIO(text)
+ return context.loadFile(io)
+
+
+class StorageTestCase(StorageTestBase):
def testFileStorage(self):
from ZODB.FileStorage import FileStorage
sample = """
@@ -55,8 +61,7 @@
create yes
</Storage>
""" % self.tmpfn
- io = StringIO(sample)
- rootconf = ZConfig.loadfile(io)
+ rootconf = self.loadConfigText(sample)
storageconf = rootconf.getSection("Storage")
cls, args = StorageConfig.getStorageInfo(storageconf)
self.assertEqual(cls, FileStorage)
@@ -73,8 +78,7 @@
wait no
</Storage>
"""
- io = StringIO(sample)
- rootconf = ZConfig.loadfile(io)
+ rootconf = self.loadConfigText(sample)
storageconf = rootconf.getSection("Storage")
cls, args = StorageConfig.getStorageInfo(storageconf)
self.assertEqual(cls, ClientStorage)
@@ -89,8 +93,7 @@
type DemoStorage
</Storage>
"""
- io = StringIO(sample)
- rootconf = ZConfig.loadfile(io)
+ rootconf = self.loadConfigText(sample)
storageconf = rootconf.getSection("Storage")
cls, args = StorageConfig.getStorageInfo(storageconf)
self.assertEqual(cls, DemoStorage)
@@ -106,8 +109,7 @@
type ZODB.DemoStorage.DemoStorage
</Storage>
"""
- io = StringIO(sample)
- rootconf = ZConfig.loadfile(io)
+ rootconf = self.loadConfigText(sample)
storageconf = rootconf.getSection("Storage")
cls, args = StorageConfig.getStorageInfo(storageconf)
self.assertEqual(cls, DemoStorage)
@@ -115,6 +117,8 @@
self.storage = StorageConfig.createStorage(storageconf)
self.assert_(isinstance(self.storage, DemoStorage))
+
+class BDBStorageTests(StorageTestBase):
def testFullStorage(self):
try:
from BDBStorage.BDBFullStorage import BDBFullStorage
@@ -128,8 +132,7 @@
</Storage>
""" % self.tmpfn
os.mkdir(self.tmpfn)
- io = StringIO(sample)
- rootconf = ZConfig.loadfile(io)
+ rootconf = self.loadConfigText(sample)
storageconf = rootconf.getSection("Storage")
cls, args = StorageConfig.getStorageInfo(storageconf)
self.assertEqual(cls, BDBFullStorage)
@@ -155,8 +158,7 @@
</Storage>
""" % self.tmpfn
os.mkdir(self.tmpfn)
- io = StringIO(sample)
- rootconf = ZConfig.loadfile(io)
+ rootconf = self.loadConfigText(sample)
storageconf = rootconf.getSection("Storage")
cls, args = StorageConfig.getStorageInfo(storageconf)
self.assertEqual(cls, BDBMinimalStorage)
@@ -169,8 +171,15 @@
# XXX _config isn't public
self.assert_(self.storage._config.cachesize, 1000)
+
def test_suite():
- return unittest.makeSuite(StorageTestCase)
+ suite = unittest.TestSuite()
+ suite.addTest(unittest.makeSuite(StorageTestCase))
+ import BDBStorage
+ if BDBStorage.is_available:
+ suite.addTest(unittest.makeSuite(BDBStorageTests))
+ return suite
+
if __name__ == '__main__':
unittest.main(defaultTest='test_suite')