[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')