[Zodb-checkins] CVS: Packages/bsddb3Storage - test_storage_api.py:1.9

barry@digicool.com barry@digicool.com
Wed, 11 Apr 2001 18:17:32 -0400 (EDT)


Update of /cvs-repository/Packages/bsddb3Storage/test
In directory korak:/tmp/cvs-serv11917/test

Modified Files:
	test_storage_api.py 
Log Message:
Simplified and reorganized to use mixin style testing.

StorageAPI, BasicStorageAPI, FullStorageAPI, MinimalStorageAPI classes removed.

MinimalTest greatly simplified.

suite(): use unittest.makeSuite()



--- Updated File test_storage_api.py in package Packages/bsddb3Storage --
--- test_storage_api.py	2001/04/10 21:14:53	1.8
+++ test_storage_api.py	2001/04/11 22:17:21	1.9
@@ -1,432 +1,42 @@
-# Run tests against the official storage API as described in
-# http://www.zope.org/Documentation/Developer/Models/ZODB/ZODB_Architecture_Storage_Interface_Info.html
+# Unit tests for basic storage functionality
 
-import os
-import errno
-import pickle
 import unittest
-import test_create
 
-from ZODB import utils
-from ZODB.Transaction import Transaction
 from ZODB import POSException
 
-DBHOME = 'test-db'
-ZERO = '\0'*8
+import BerkeleyTestBase
+from BasicStorage import BasicStorage
+from VersionStorage import VersionStorage
+from TransactionalUndoStorage import TransactionalUndoStorage
+from TransactionalUndoVersionStorage import TransactionalUndoVersionStorage
 
 
 
-class StorageAPI(test_create.BaseFramework):
-    def setUp(self):
-        # A simpler setUp() than the base class since we don't need the DB
-        # object, the connection, or the root.
-        self._dbhome = DBHOME
-        os.mkdir(self._dbhome)
-
-        try:
-            self._storage = self.ConcreteStorage(self._dbhome)
-        except:
-            self.tearDown()
-            raise
-        self._transaction = Transaction()
-
-    def _close(self):
-        self._transaction.abort()
-        self._storage.close()
-
-    def _dostore(self, oid=None, revid=None, data=None, version=None):
-        # Defaults
-        if oid is None:
-            oid = self._storage.new_oid()
-        if revid is None:
-            revid = ZERO
-        if data is None:
-            data = pickle.dumps(7)
-        else:
-            data = pickle.dumps(data)
-        if version is None:
-            version = ''
-        # Begin the transaction
-        self._storage.tpc_begin(self._transaction)
-        # Store an object
-        newrevid = self._storage.store(oid, revid, data, version,
-                                       self._transaction)
-        # Finish the transaction
-        self._storage.tpc_vote(self._transaction)
-        self._storage.tpc_finish(self._transaction)
-        return newrevid
-
-
-
-class BasicStorageAPI(StorageAPI):
-    def checkBasics(self):
-        self._storage.tpc_begin(self._transaction)
-        # This should simply return
-        self._storage.tpc_begin(self._transaction)
-        # Aborting is easy
-        self._storage.tpc_abort(self._transaction)
-        # Test a few expected exceptions when we're doing operations giving a
-        # different Transaction object than the one we've begun on.
-        self._storage.tpc_begin(self._transaction)
-        self.assertRaises(
-            POSException.StorageTransactionError,
-            self._storage.store,
-            0, 0, 0, 0, Transaction())
-        self.assertRaises(
-            POSException.StorageTransactionError,
-            self._storage.abortVersion,
-            0, Transaction())
-        self.assertRaises(
-            POSException.StorageTransactionError,
-            self._storage.commitVersion,
-            0, 1, Transaction())
-        self.assertRaises(
-            POSException.StorageTransactionError,
-            self._storage.store,
-            0, 1, 2, 3, Transaction())
-        self._storage.tpc_abort(self._transaction)
-
-    def checkNonVersionStore(self, oid=None, revid=None, version=None):
-        revid = ZERO
-        newrevid = self._dostore(revid=revid)
-        # Finish the transaction.
-        assert newrevid <> revid
-
-    def checkLen(self):
-        # The length of the database ought to grow by one each time
-        assert len(self._storage) == 0
-        self._dostore()
-        assert len(self._storage) == 1
-        self._dostore()
-        assert len(self._storage) == 2
-
-    def checkNonVersionStoreAndLoad(self):
-        oid = self._storage.new_oid()
-        self._dostore(oid=oid, data=7)
-        data, revid = self._storage.load(oid, '')
-        value = pickle.loads(data)
-        assert value == 7
-        # Now do a bunch of updates to an object
-        for i in range(13, 22):
-            revid = self._dostore(oid, revid=revid, data=i)
-        # Now get the latest revision of the object
-        data, revid = self._storage.load(oid, '')
-        assert pickle.loads(data) == 21
-
-    def checkNonVersionModifiedInVersion(self):
-        oid = self._storage.new_oid()
-        self._dostore(oid=oid)
-        assert self._storage.modifiedInVersion(oid) == ''
-
+class MinimalTest(BerkeleyTestBase.MinimalTestBase, BasicStorage):
     def checkLoadSerial(self):
-        oid = self._storage.new_oid()
-        revid = ZERO
-        revisions = {}
-        for i in range(31, 38):
-            revid = self._dostore(oid, revid=revid, data=i)
-            revisions[revid] = i
-        # Now make sure all the revisions have the correct value
-        for revid, value in revisions.items():
-            data = self._storage.loadSerial(oid, revid)
-            assert pickle.loads(data) == value
-
-    def checkVersionedStoreAndLoad(self):
-        # Store a couple of non-version revisions of the object
-        oid = self._storage.new_oid()
-        revid = self._dostore(oid, data=11)
-        revid = self._dostore(oid, revid=revid, data=12)
-        # And now store some new revisions in a version
-        version = 'test-version'
-        revid = self._dostore(oid, revid=revid, data=13, version=version)
-        revid = self._dostore(oid, revid=revid, data=14, version=version)
-        revid = self._dostore(oid, revid=revid, data=15, version=version)
-        # Now read back the object in both the non-version and version and
-        # make sure the values jive.
-        data, revid = self._storage.load(oid, '')
-        assert pickle.loads(data) == 12
-        data, revid = self._storage.load(oid, version)
-        assert pickle.loads(data) == 15
-
-    def checkVersionedLoadErrors(self):
-        oid = self._storage.new_oid()
-        version = 'test-version'
-        revid = self._dostore(oid, data=11)
-        revid = self._dostore(oid, revid=revid, data=12, version=version)
-        # Try to load a bogus oid
-        self.assertRaises(KeyError,
-                          self._storage.load,
-                          self._storage.new_oid(), '')
-        # Try to load a bogus version string
-        self.assertRaises(POSException.VersionError,
-                          self._storage.load,
-                          oid, 'bogus')
-
-    def checkConflicts(self):
-        oid = self._storage.new_oid()
-        revid1 = self._dostore(oid, data=11)
-        revid2 = self._dostore(oid, revid=revid1, data=12)
-        self.assertRaises(POSException.ConflictError,
-                          self._dostore,
-                          oid, revid=revid1, data=13)
-
-    def checkVersionLock(self):
-        oid = self._storage.new_oid()
-        revid = self._dostore(oid, data=11)
-        version = 'test-version'
-        revid = self._dostore(oid, revid=revid, data=12, version=version)
-        self.assertRaises(POSException.VersionLockError,
-                          self._dostore,
-                          oid, revid=revid, data=14,
-                          version='another-version')
-
-    def checkVersionEmpty(self):
-        # Before we store anything, these versions ought to be empty
-        version = 'test-version'
-        assert self._storage.versionEmpty('')
-        assert self._storage.versionEmpty(version)
-        # Now store some objects
-        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, version=version)
-        revid = self._dostore(oid, revid=revid, data=14, version=version)
-        # The blank version should not be empty
-        assert not self._storage.versionEmpty('')
-        # Neither should 'test-version'
-        assert not self._storage.versionEmpty(version)
-        # But this non-existant version should be empty
-        assert self._storage.versionEmpty('bogus')
-
-    def checkVersions(self):
-        # Store some objects in the non-version
-        oid1 = self._storage.new_oid()
-        oid2 = self._storage.new_oid()
-        oid3 = self._storage.new_oid()
-        revid1 = self._dostore(oid1, data=11)
-        revid2 = self._dostore(oid2, data=12)
-        revid3 = self._dostore(oid3, data=13)
-        # Now create some new versions
-        revid1 = self._dostore(oid1, revid=revid1, data=14, version='one')
-        revid2 = self._dostore(oid2, revid=revid2, data=15, version='two')
-        revid3 = self._dostore(oid3, revid=revid3, data=16, version='three')
-        # Ask for the versions
-        versions = self._storage.versions()
-        assert 'one' in versions
-        assert 'two' in versions
-        assert 'three' in versions
-        # Now flex the `max' argument
-        versions = self._storage.versions(1)
-        assert len(versions) == 1
-        assert 'one' in versions or 'two' in versions or 'three' in versions
-
-    def _setup_version(self, version='test-version'):
-        # Store some revisions in the non-version
-        oid = self._storage.new_oid()
-        revid = self._dostore(oid, data=49)
-        revid = self._dostore(oid, revid=revid, data=50)
-        nvrevid = revid = self._dostore(oid, revid=revid, data=51)
-        # Now do some stores in a version
-        revid = self._dostore(oid, revid=revid, data=52, version=version)
-        revid = self._dostore(oid, revid=revid, data=53, version=version)
-        revid = self._dostore(oid, revid=revid, data=54, version=version)
-        return oid, version
-
-    def checkAbortVersion(self):
-        oid, version = self._setup_version()
-        # Now abort the version -- must be done in a transaction
-        self._storage.tpc_begin(self._transaction)
-        oids = self._storage.abortVersion(version, self._transaction)
-        self._storage.tpc_vote(self._transaction)
-        self._storage.tpc_finish(self._transaction)
-        assert len(oids) == 1
-        assert oids[0] == oid
-        data, revid = self._storage.load(oid, '')
-        assert pickle.loads(data) == 51
-
-    def checkAbortVersionErrors(self):
-        oid, version = self._setup_version()
-        # Now abort a bogus version
-        self._storage.tpc_begin(self._transaction)
-        self.assertRaises(KeyError,
-                          self._storage.abortVersion,
-                          'bogus', self._transaction)
-        # And try to abort the empty version
-        self.assertRaises(KeyError,
-                          self._storage.abortVersion,
-                          '', self._transaction)
-        # But now we really try to abort the version
-        oids = self._storage.abortVersion(version, self._transaction)
-        self._storage.tpc_vote(self._transaction)
-        self._storage.tpc_finish(self._transaction)
-        assert len(oids) == 1
-        assert oids[0] == oid
-        data, revid = self._storage.load(oid, '')
-        assert pickle.loads(data) == 51
-
-    def checkModifyAfterAbortVersion(self):
-        oid, version = self._setup_version()
-        # Now abort the version
-        self._storage.tpc_begin(self._transaction)
-        oids = self._storage.abortVersion(version, self._transaction)
-        self._storage.tpc_vote(self._transaction)
-        self._storage.tpc_finish(self._transaction)
-        # Load the object's current state (which gets us the revid)
-        data, revid = self._storage.load(oid, '')
-        # And modify it a few times
-        revid = self._dostore(oid, revid=revid, data=52)
-        revid = self._dostore(oid, revid=revid, data=53)
-        revid = self._dostore(oid, revid=revid, data=54)
-        data, newrevid = self._storage.load(oid, '')
-        assert newrevid == revid
-        assert pickle.loads(data) == 54
-
-    def checkCommitToNonVersion(self):
-        oid, version = self._setup_version()
-        data, revid = self._storage.load(oid, version)
-        assert pickle.loads(data) == 54
-        data, revid = self._storage.load(oid, '')
-        assert pickle.loads(data) == 51
-        # Try committing this version to the empty version
-        self._storage.tpc_begin(self._transaction)
-        oids = self._storage.commitVersion(version, '', self._transaction)
-        self._storage.tpc_vote(self._transaction)
-        self._storage.tpc_finish(self._transaction)
-        data, revid = self._storage.load(oid, '')
-        assert pickle.loads(data) == 54
-
-    def checkCommitToOtherVersion(self):
-        oid1, version1 = self._setup_version('one')
-        data, revid1 = self._storage.load(oid1, version1)
-        assert pickle.loads(data) == 54
-        oid2, version2 = self._setup_version('two')
-        data, revid2 = self._storage.load(oid2, version2)
-        assert pickle.loads(data) == 54
-        # Let's make sure we can't get object1 in version2
-        self.assertRaises(POSException.VersionError,
-                          self._storage.load, oid1, version2)
-        # Okay, now let's commit object1 to version2
-        self._storage.tpc_begin(self._transaction)
-        oids = self._storage.commitVersion(version1, version2,
-                                           self._transaction)
-        self._storage.tpc_vote(self._transaction)
-        self._storage.tpc_finish(self._transaction)
-        assert len(oids) == 1
-        assert oids[0] == oid1
-        data, revid = self._storage.load(oid1, version2)
-        assert pickle.loads(data) == 54
-        data, revid = self._storage.load(oid2, version2)
-        assert pickle.loads(data) == 54
-        self.assertRaises(POSException.VersionError,
-                          self._storage.load, oid1, version1)
-
-    def checkAbortOneVersionCommitTheOther(self):
-        oid1, version1 = self._setup_version('one')
-        data, revid1 = self._storage.load(oid1, version1)
-        assert pickle.loads(data) == 54
-        oid2, version2 = self._setup_version('two')
-        data, revid2 = self._storage.load(oid2, version2)
-        assert pickle.loads(data) == 54
-        # Let's make sure we can't get object1 in version2
-        self.assertRaises(POSException.VersionError,
-                          self._storage.load, oid1, version2)
-        # First, let's abort version1
-        self._storage.tpc_begin(self._transaction)
-        oids = self._storage.abortVersion(version1, self._transaction)
-        self._storage.tpc_vote(self._transaction)
-        self._storage.tpc_finish(self._transaction)
-        assert len(oids) == 1
-        assert oids[0] == oid1
-        data, revid = self._storage.load(oid1, '')
-        assert pickle.loads(data) == 51
-        self.assertRaises(POSException.VersionError,
-                          self._storage.load, oid1, version1)
-        self.assertRaises(POSException.VersionError,
-                          self._storage.load, oid1, version2)
-        data, revid = self._storage.load(oid2, '')
-        assert pickle.loads(data) == 51
-        data, revid = self._storage.load(oid2, version2)
-        assert pickle.loads(data) == 54
-        # Okay, now let's commit version2 back to the trunk
-        self._storage.tpc_begin(self._transaction)
-        oids = self._storage.commitVersion(version2, '', self._transaction)
-        self._storage.tpc_vote(self._transaction)
-        self._storage.tpc_finish(self._transaction)
-        assert len(oids) == 1
-        assert oids[0] == oid2
-        # These objects should not be found in version 2
-        self.assertRaises(POSException.VersionError,
-                          self._storage.load, oid1, version2)
-        self.assertRaises(POSException.VersionError,
-                          self._storage.load, oid2, version2)
-        # But the trunk should be up to date now
-        data, revid = self._storage.load(oid2, '')
-        assert pickle.loads(data) == 54
-
-
-class FullStorageAPI(BasicStorageAPI):
-    import Full
-    ConcreteStorage = Full.Full
-
-
-class MinimalStorageAPI(BasicStorageAPI):
-    import Minimal
-    ConcreteStorage = Minimal.Minimal
-
-    def checkLoadSerial(self):
         # This storage doesn't support versions, so we should get an exception
         self.assertRaises(POSException.Unsupported,
-                          BasicStorageAPI.checkLoadSerial,
+                          BasicStorage.checkLoadSerial,
                           self)
 
     def checkVersionedStoreAndLoad(self):
         # This storage doesn't support versions, so we should get an exception
+        oid = self._storage.new_oid()
         self.assertRaises(POSException.Unsupported,
-                          BasicStorageAPI.checkVersionedStoreAndLoad,
-                          self)
+                          self._dostore,
+                          oid, data=11, version='a version')
+
+
+class FullTest(BerkeleyTestBase.FullTestBase, BasicStorage, VersionStorage,
+               TransactionalUndoStorage, TransactionalUndoVersionStorage):
+    pass
 
 
 
 def suite():
     suite = unittest.TestSuite()
-    # Minimal storage tests
-    suite.addTest(MinimalStorageAPI('checkBasics'))
-    suite.addTest(MinimalStorageAPI('checkNonVersionStore'))
-    suite.addTest(MinimalStorageAPI('checkLen'))
-    suite.addTest(MinimalStorageAPI('checkNonVersionStoreAndLoad'))
-    suite.addTest(MinimalStorageAPI('checkNonVersionModifiedInVersion'))
-    suite.addTest(MinimalStorageAPI('checkLoadSerial'))
-    suite.addTest(MinimalStorageAPI('checkVersionedStoreAndLoad'))
-    # Skipping: MinimalStorageAPI.checkVersionedLoadErrors()
-    suite.addTest(MinimalStorageAPI('checkConflicts'))
-    # Skipping: MinimalStorageAPI.checkVersionLock()
-    # Skipping: MinimalStorageAPI.checkVersionEmpty()
-    # Skipping: MinimalStorageAPI.checkVersions()
-    # Skipping: MinimalStorageAPI.checkAbortVersion()
-    # Skipping: MinimalStorageAPI.checkAbortVersionErrors()
-    # Skipping: MinimalStorageAPI.checkModifyAfterAbortVersion()
-    # Skipping: MinimalStorageAPI.checkCommitToNonVersion()
-    # Skipping: MinimalStorageAPI.checkCommitToOtherVersion()
-    # Skipping: MinimalStorageAPI.checkAbortOneVersionCommitTheOther()
-    #
-    # Full storage tests
-    suite.addTest(FullStorageAPI('checkBasics'))
-    suite.addTest(FullStorageAPI('checkNonVersionStore'))
-    suite.addTest(FullStorageAPI('checkLen'))
-    suite.addTest(FullStorageAPI('checkNonVersionStoreAndLoad'))
-    suite.addTest(FullStorageAPI('checkNonVersionModifiedInVersion'))
-    suite.addTest(FullStorageAPI('checkLoadSerial'))
-    suite.addTest(FullStorageAPI('checkVersionedStoreAndLoad'))
-    suite.addTest(FullStorageAPI('checkVersionedLoadErrors'))
-    suite.addTest(FullStorageAPI('checkConflicts'))
-    suite.addTest(FullStorageAPI('checkVersionLock'))
-    suite.addTest(FullStorageAPI('checkVersionEmpty'))
-    suite.addTest(FullStorageAPI('checkVersions'))
-    suite.addTest(FullStorageAPI('checkAbortVersion'))
-    suite.addTest(FullStorageAPI('checkAbortVersionErrors'))
-    suite.addTest(FullStorageAPI('checkModifyAfterAbortVersion'))
-    suite.addTest(FullStorageAPI('checkCommitToNonVersion'))
-    suite.addTest(FullStorageAPI('checkCommitToOtherVersion'))
-    suite.addTest(FullStorageAPI('checkAbortOneVersionCommitTheOther'))
+    suite.addTest(unittest.makeSuite(MinimalTest, 'check'))
+    suite.addTest(unittest.makeSuite(FullTest, 'check'))
     return suite