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

[barry@digicool.com]

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

Added Files:
	test_storage_api.py 
Log Message:
The start of a suite of unit tests against the storage API, using both
the Minimal and Full storages.



--- Added File test_storage_api.py in package Packages/bsddb3Storage ---
# Run tests against the official storage API as described in
# http://www.zope.org/Documentation/Developer/Models/ZODB/ZODB_Architecture_Storage_Interface_Info.html

import os
import errno
import pickle
import unittest
import test_create

from ZODB import utils
from ZODB.Transaction import Transaction
from ZODB.POSException import StorageTransactionError

DBHOME = 'test-db'
ZERO = '\0'*8



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 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(
            StorageTransactionError,
            self._storage.store,
            0, 0, 0, 0, Transaction())
        self.assertRaises(
            StorageTransactionError,
            self._storage.abortVersion,
            0, Transaction())
        self.assertRaises(
            StorageTransactionError,
            self._storage.commitVersion,
            0, 1, Transaction())
        self.assertRaises(
            StorageTransactionError,
            self._storage.store,
            0, 1, 2, 3, Transaction())

    def checkNonVersionStore(self):
        # Some objects to store
        oid = self._storage.new_oid()
        revid = ZERO
        data = pickle.dumps(7)
        version = ''
        # Start the transaction, store an object, and be sure the revision id
        # is different than what we passed.
        self._storage.tpc_begin(self._transaction)
        newrevid = self._storage.store(oid, revid, data, version,
                                       self._transaction)
        assert newrevid <> revid
        # Finish the transaction.
        self._storage.tpc_vote(self._transaction)
        self._storage.tpc_finish(self._transaction)

    def checkLen(self):
        # The length of the database ought to grow by one each time
        assert len(self._storage) == 0
        self.checkNonVersionStore()
        assert len(self._storage) == 1
        self.checkNonVersionStore()
        assert len(self._storage) == 2



class FullStorageAPI(StorageAPI):
    import Full
    ConcreteStorage = Full.Full


class MinimalStorageAPI(StorageAPI):
    import Minimal
    ConcreteStorage = Minimal.Minimal



def suite():
    suite = unittest.TestSuite()
    # Minimal storage tests
    suite.addTest(MinimalStorageAPI('checkBasics'))
    suite.addTest(MinimalStorageAPI('checkNonVersionStore'))
    suite.addTest(MinimalStorageAPI('checkLen'))
    # Full storage tests
    suite.addTest(FullStorageAPI('checkBasics'))
    suite.addTest(FullStorageAPI('checkNonVersionStore'))
    suite.addTest(FullStorageAPI('checkLen'))
    return suite



if __name__ == '__main__':
    unittest.main(defaultTest='suite')