diff options
Diffstat (limited to 'Lib/bsddb/test/test_dbshelve.py')
| -rw-r--r-- | Lib/bsddb/test/test_dbshelve.py | 305 |
1 files changed, 305 insertions, 0 deletions
diff --git a/Lib/bsddb/test/test_dbshelve.py b/Lib/bsddb/test/test_dbshelve.py new file mode 100644 index 0000000..bed7212 --- /dev/null +++ b/Lib/bsddb/test/test_dbshelve.py @@ -0,0 +1,305 @@ +""" +TestCases for checking dbShelve objects. +""" + +import sys, os, string +import tempfile, random +from pprint import pprint +from types import * +import unittest + +from bsddb import dbshelve, db + +from test.test_support import verbose + + +#---------------------------------------------------------------------- + +# We want the objects to be comparable so we can test dbshelve.values +# later on. +class DataClass: + def __init__(self): + self.value = random.random() + + def __cmp__(self, other): + return cmp(self.value, other) + +class DBShelveTestCase(unittest.TestCase): + def setUp(self): + self.filename = tempfile.mktemp() + self.do_open() + + def tearDown(self): + self.do_close() + try: + os.remove(self.filename) + except os.error: + pass + + def populateDB(self, d): + for x in string.letters: + d['S' + x] = 10 * x # add a string + d['I' + x] = ord(x) # add an integer + d['L' + x] = [x] * 10 # add a list + + inst = DataClass() # add an instance + inst.S = 10 * x + inst.I = ord(x) + inst.L = [x] * 10 + d['O' + x] = inst + + + # overridable in derived classes to affect how the shelf is created/opened + def do_open(self): + self.d = dbshelve.open(self.filename) + + # and closed... + def do_close(self): + self.d.close() + + + + def test01_basics(self): + if verbose: + print '\n', '-=' * 30 + print "Running %s.test01_basics..." % self.__class__.__name__ + + self.populateDB(self.d) + self.d.sync() + self.do_close() + self.do_open() + d = self.d + + l = len(d) + k = d.keys() + s = d.stat() + f = d.fd() + + if verbose: + print "length:", l + print "keys:", k + print "stats:", s + + assert 0 == d.has_key('bad key') + assert 1 == d.has_key('IA') + assert 1 == d.has_key('OA') + + d.delete('IA') + del d['OA'] + assert 0 == d.has_key('IA') + assert 0 == d.has_key('OA') + assert len(d) == l-2 + + values = [] + for key in d.keys(): + value = d[key] + values.append(value) + if verbose: + print "%s: %s" % (key, value) + self.checkrec(key, value) + + dbvalues = d.values() + assert len(dbvalues) == len(d.keys()) + values.sort() + dbvalues.sort() + assert values == dbvalues + + items = d.items() + assert len(items) == len(values) + + for key, value in items: + self.checkrec(key, value) + + assert d.get('bad key') == None + assert d.get('bad key', None) == None + assert d.get('bad key', 'a string') == 'a string' + assert d.get('bad key', [1, 2, 3]) == [1, 2, 3] + + d.set_get_returns_none(0) + self.assertRaises(db.DBNotFoundError, d.get, 'bad key') + d.set_get_returns_none(1) + + d.put('new key', 'new data') + assert d.get('new key') == 'new data' + assert d['new key'] == 'new data' + + + + def test02_cursors(self): + if verbose: + print '\n', '-=' * 30 + print "Running %s.test02_cursors..." % self.__class__.__name__ + + self.populateDB(self.d) + d = self.d + + count = 0 + c = d.cursor() + rec = c.first() + while rec is not None: + count = count + 1 + if verbose: + print rec + key, value = rec + self.checkrec(key, value) + rec = c.next() + + assert count == len(d) + + count = 0 + c = d.cursor() + rec = c.last() + while rec is not None: + count = count + 1 + if verbose: + print rec + key, value = rec + self.checkrec(key, value) + rec = c.prev() + + assert count == len(d) + + c.set('SS') + key, value = c.current() + self.checkrec(key, value) + + c.close() + + + + + def checkrec(self, key, value): + x = key[1] + if key[0] == 'S': + assert type(value) == StringType + assert value == 10 * x + + elif key[0] == 'I': + assert type(value) == IntType + assert value == ord(x) + + elif key[0] == 'L': + assert type(value) == ListType + assert value == [x] * 10 + + elif key[0] == 'O': + assert type(value) == InstanceType + assert value.S == 10 * x + assert value.I == ord(x) + assert value.L == [x] * 10 + + else: + raise AssertionError, 'Unknown key type, fix the test' + +#---------------------------------------------------------------------- + +class BasicShelveTestCase(DBShelveTestCase): + def do_open(self): + self.d = dbshelve.DBShelf() + self.d.open(self.filename, self.dbtype, self.dbflags) + + def do_close(self): + self.d.close() + + + + +class BTreeShelveTestCase(BasicShelveTestCase): + dbtype = db.DB_BTREE + dbflags = db.DB_CREATE + + +class HashShelveTestCase(BasicShelveTestCase): + dbtype = db.DB_BTREE + dbflags = db.DB_CREATE + + +class ThreadBTreeShelveTestCase(BasicShelveTestCase): + dbtype = db.DB_BTREE + dbflags = db.DB_CREATE | db.DB_THREAD + + +class ThreadHashShelveTestCase(BasicShelveTestCase): + dbtype = db.DB_BTREE + dbflags = db.DB_CREATE | db.DB_THREAD + + +#---------------------------------------------------------------------- + +class BasicEnvShelveTestCase(DBShelveTestCase): + def do_open(self): + self.homeDir = homeDir = os.path.join(os.path.dirname(sys.argv[0]), 'db_home') + try: os.mkdir(homeDir) + except os.error: pass + self.env = db.DBEnv() + self.env.open(homeDir, self.envflags | db.DB_INIT_MPOOL | db.DB_CREATE) + + self.filename = os.path.split(self.filename)[1] + self.d = dbshelve.DBShelf(self.env) + self.d.open(self.filename, self.dbtype, self.dbflags) + + + def do_close(self): + self.d.close() + self.env.close() + + + def tearDown(self): + self.do_close() + import glob + files = glob.glob(os.path.join(self.homeDir, '*')) + for file in files: + os.remove(file) + + + +class EnvBTreeShelveTestCase(BasicEnvShelveTestCase): + envflags = 0 + dbtype = db.DB_BTREE + dbflags = db.DB_CREATE + + +class EnvHashShelveTestCase(BasicEnvShelveTestCase): + envflags = 0 + dbtype = db.DB_BTREE + dbflags = db.DB_CREATE + + +class EnvThreadBTreeShelveTestCase(BasicEnvShelveTestCase): + envflags = db.DB_THREAD + dbtype = db.DB_BTREE + dbflags = db.DB_CREATE | db.DB_THREAD + + +class EnvThreadHashShelveTestCase(BasicEnvShelveTestCase): + envflags = db.DB_THREAD + dbtype = db.DB_BTREE + dbflags = db.DB_CREATE | db.DB_THREAD + + +#---------------------------------------------------------------------- +# TODO: Add test cases for a DBShelf in a RECNO DB. + + +#---------------------------------------------------------------------- + +def suite(): + theSuite = unittest.TestSuite() + + theSuite.addTest(unittest.makeSuite(DBShelveTestCase)) + theSuite.addTest(unittest.makeSuite(BTreeShelveTestCase)) + theSuite.addTest(unittest.makeSuite(HashShelveTestCase)) + theSuite.addTest(unittest.makeSuite(ThreadBTreeShelveTestCase)) + theSuite.addTest(unittest.makeSuite(ThreadHashShelveTestCase)) + theSuite.addTest(unittest.makeSuite(EnvBTreeShelveTestCase)) + theSuite.addTest(unittest.makeSuite(EnvHashShelveTestCase)) + theSuite.addTest(unittest.makeSuite(EnvThreadBTreeShelveTestCase)) + theSuite.addTest(unittest.makeSuite(EnvThreadHashShelveTestCase)) + + return theSuite + + +if __name__ == '__main__': + unittest.main( defaultTest='suite' ) + + |