[Zodb-checkins] CVS: Packages/bsddb3Storage - test_storage_api.py:1.3
barry@digicool.com
barry@digicool.com
Wed, 4 Apr 2001 15:29:29 -0400 (EDT)
Update of /cvs-repository/Packages/bsddb3Storage/test
In directory korak:/tmp/cvs-serv24251
Modified Files:
test_storage_api.py
Log Message:
Factor out _dostore() which stores object revisions, since this is a
common operation. ;) Also, _dostore() automatically pickles the data
argument, since you don't really want anything else.
Lots of new tests:
checkNonVersionModifiedInVersion()
checkLoadSerial()
checkVersionedStoreAndLoad()
checkVersionedLoadErrors()
checkConflicts()
checkVersionLock()
checkVersionEmpty()
checkVersions()
checkAbortVersion()
Skip for the Minimal storage:
checkVersionedLoadErrors()
checkVersionLock()
checkVersionEmpty()
checkVersions()
checkAbortVersion()
Override for the Minimal storage (basically check for Unsupported exceptions):
checkLoadSerial()
checkVersionedStoreAndLoad()
--- Updated File test_storage_api.py in package Packages/bsddb3Storage --
--- test_storage_api.py 2001/04/03 23:22:32 1.2
+++ test_storage_api.py 2001/04/04 19:29:28 1.3
@@ -9,7 +9,7 @@
from ZODB import utils
from ZODB.Transaction import Transaction
-from ZODB.POSException import StorageTransactionError
+from ZODB import POSException
DBHOME = 'test-db'
ZERO = '\0'*8
@@ -40,57 +40,199 @@
# different Transaction object than the one we've begun on.
self._storage.tpc_begin(self._transaction)
self.assertRaises(
- StorageTransactionError,
+ POSException.StorageTransactionError,
self._storage.store,
0, 0, 0, 0, Transaction())
self.assertRaises(
- StorageTransactionError,
+ POSException.StorageTransactionError,
self._storage.abortVersion,
0, Transaction())
self.assertRaises(
- StorageTransactionError,
+ POSException.StorageTransactionError,
self._storage.commitVersion,
0, 1, Transaction())
self.assertRaises(
- StorageTransactionError,
+ POSException.StorageTransactionError,
self._storage.store,
0, 1, 2, 3, Transaction())
- def checkNonVersionStore(self, oid=None, revid=None, version=None):
- # Some objects to store
+ 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
- data = pickle.dumps(7)
+ if data is None:
+ data = pickle.dumps(7)
+ else:
+ data = pickle.dumps(data)
if version is None:
version = ''
- # Start the transaction, store an object, and be sure the revision id
- # is different than what we passed.
+ # Begin the transaction
self._storage.tpc_begin(self._transaction)
+ # Store an object
newrevid = self._storage.store(oid, revid, data, version,
self._transaction)
- assert newrevid <> revid
- # Finish the transaction.
+ # Finish the transaction
self._storage.tpc_vote(self._transaction)
self._storage.tpc_finish(self._transaction)
+ return newrevid
+
+ 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.checkNonVersionStore()
+ self._dostore()
assert len(self._storage) == 1
- self.checkNonVersionStore()
+ self._dostore()
assert len(self._storage) == 2
def checkNonVersionStoreAndLoad(self):
oid = self._storage.new_oid()
- self.checkNonVersionStore(oid)
+ self._dostore(oid=oid, data=7)
data, revid = self._storage.load(oid, '')
- assert revid == ZERO
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) == ''
+
+ 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 checkAbortVersion(self):
+ # 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
+ version = 'test-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)
+ # 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 nvrevid == revid
+ assert pickle.loads(data) == 51
+
class FullStorageAPI(StorageAPI):
@@ -102,6 +244,18 @@
import Minimal
ConcreteStorage = Minimal.Minimal
+ def checkLoadSerial(self):
+ # This storage doesn't support versions, so we should get an exception
+ self.assertRaises(POSException.Unsupported,
+ StorageAPI.checkLoadSerial,
+ self)
+
+ def checkVersionedStoreAndLoad(self):
+ # This storage doesn't support versions, so we should get an exception
+ self.assertRaises(POSException.Unsupported,
+ StorageAPI.checkVersionedStoreAndLoad,
+ self)
+
def suite():
@@ -111,11 +265,29 @@
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()
# 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'))
return suite