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

[barry@digicool.com]

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

Added Files:
	test_full.py 
Log Message:
Test some simple stuff on the Full database.  This currently only
tests the same basic stuff that test_minimal.py tests.



--- Added File test_full.py in package Packages/bsddb3Storage ---
# Test the Full storage at the ZODB level

import os
import errno
import time
import unittest

DBHOME = 'test-db'



class CommitAndRead(unittest.TestCase):
    # Never tear down the test framework since we want the database support
    # files to persist.  MasterSetup will take care of cleaning things up when
    # we're done.
    def setUp(self):
        import Full
        from ZODB import DB

        self._dbhome = DBHOME
        try:
            os.mkdir(self._dbhome)
        except OSError, e:
            if e.errno <> errno.EEXIST: raise

        try:
            self._storage = Full.Full(self._dbhome)
            self._db = DB(self._storage)
            self._conn = self._db.open()
            self._root = self._conn.root()
        except:
            self.tearDown()
            raise

    def tearDown(self):
        for file in os.listdir(DBHOME):
            os.unlink(os.path.join(DBHOME, file))
        os.removedirs(DBHOME)
        
    def checkCommit(self):
        """Full: On an empty database, commit some data"""
        from BTree import BTree

        assert not self._root
        names = self._root['names'] = BTree()
        names['Warsaw'] = 'Barry'
        names['Hylton'] = 'Jeremy'
        get_transaction().commit()

    def checkReadAfterCommit(self):
        """Full: On a database with data, try to read the data"""
        self.checkCommit()
        names = self._root['names']
        assert names['Warsaw'] == 'Barry'
        assert names['Hylton'] == 'Jeremy'
        assert names.get('Drake') is None

    def checkAbortAfterRead(self):
        """Full: On a database with data, abort a transaction"""
        self.checkReadAfterCommit()
        names = self._root['names']
        names['Drake'] = 'Fred'
        get_transaction().abort()

    def checkReadAfterAbort(self):
        """Full: On a database, make sure the aborted data is missing"""
        self.checkAbortAfterRead()
        names = self._root['names']
        assert names.get('Drake') is None

    def checkChangingCommits(self):
        """Full: Commit some ever changing data"""
        self.checkReadAfterAbort()
        now = time.time()
        # Make sure the last timestamp was more than 3 seconds ago
        timestamp = self._root.get('timestamp')
        if timestamp is None:
            timestamp = self._root['timestamp'] = 0
            get_transaction().commit()
        assert now > timestamp + 3
        self._root['timestamp'] = now
        time.sleep(3)



def suite():
    suite = unittest.TestSuite()
    suite.addTest(CommitAndRead('checkCommit'))
    suite.addTest(CommitAndRead('checkReadAfterCommit'))
    suite.addTest(CommitAndRead('checkAbortAfterRead'))
    suite.addTest(CommitAndRead('checkReadAfterCommit'))
    for i in range(5):
        suite.addTest(CommitAndRead('checkChangingCommits'))
    return suite


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