diff options
Diffstat (limited to 'Lib/bsddb/test/test_dbshelve.py')
-rw-r--r-- | Lib/bsddb/test/test_dbshelve.py | 385 |
1 files changed, 0 insertions, 385 deletions
diff --git a/Lib/bsddb/test/test_dbshelve.py b/Lib/bsddb/test/test_dbshelve.py deleted file mode 100644 index 5149ac8661..0000000000 --- a/Lib/bsddb/test/test_dbshelve.py +++ /dev/null @@ -1,385 +0,0 @@ -""" -TestCases for checking dbShelve objects. -""" - -import os, string -import random -import unittest - - -from .test_all import db, dbshelve, test_support, verbose, \ - get_new_environment_path, get_new_database_path - - - -#---------------------------------------------------------------------- - -# 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 __repr__(self) : # For Python 3.0 comparison - return "DataClass %f" %self.value - - def __cmp__(self, other): # For Python 2.x comparison - return cmp(self.value, other) - - -class DBShelveTestCase(unittest.TestCase): - def setUp(self): - import sys - if sys.version_info[0] >= 3 : - from .test_all import do_proxy_db_py3k - self._flag_proxy_db_py3k = do_proxy_db_py3k(False) - self.filename = get_new_database_path() - self.do_open() - - def tearDown(self): - import sys - if sys.version_info[0] >= 3 : - from .test_all import do_proxy_db_py3k - do_proxy_db_py3k(self._flag_proxy_db_py3k) - self.do_close() - test_support.unlink(self.filename) - - def mk(self, key): - """Turn key into an appropriate key type for this db""" - # override in child class for RECNO - import sys - if sys.version_info[0] < 3 : - return key - else : - return bytes(key, "iso8859-1") # 8 bits - - def populateDB(self, d): - for x in string.letters: - d[self.mk('S' + x)] = 10 * x # add a string - d[self.mk('I' + x)] = ord(x) # add an integer - d[self.mk('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[self.mk('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 = list(d.keys()) - s = d.stat() - f = d.fd() - - if verbose: - print("length:", l) - print("keys:", k) - print("stats:", s) - - self.assertEqual(0, self.mk('bad key') in d) - self.assertEqual(1, self.mk('IA') in d) - self.assertEqual(1, self.mk('OA') in d) - - d.delete(self.mk('IA')) - del d[self.mk('OA')] - self.assertEqual(0, self.mk('IA') in d) - self.assertEqual(0, self.mk('OA') in d) - self.assertEqual(len(d), l-2) - - values = [] - for key in list(d.keys()): - value = d[key] - values.append(value) - if verbose: - print("%s: %s" % (key, value)) - self.checkrec(key, value) - - dbvalues = list(d.values()) - self.assertEqual(len(dbvalues), len(list(d.keys()))) - import sys - if sys.version_info[0] < 3 : - values.sort() - dbvalues.sort() - self.assertEqual(values, dbvalues) - else : # XXX: Convert all to strings. Please, improve - values.sort(key=lambda x : str(x)) - dbvalues.sort(key=lambda x : str(x)) - self.assertEqual(repr(values), repr(dbvalues)) - - items = list(d.items()) - self.assertEqual(len(items), len(values)) - - for key, value in items: - self.checkrec(key, value) - - self.assertEqual(d.get(self.mk('bad key')), None) - self.assertEqual(d.get(self.mk('bad key'), None), None) - self.assertEqual(d.get(self.mk('bad key'), 'a string'), 'a string') - self.assertEqual(d.get(self.mk('bad key'), [1, 2, 3]), [1, 2, 3]) - - d.set_get_returns_none(0) - self.assertRaises(db.DBNotFoundError, d.get, self.mk('bad key')) - d.set_get_returns_none(1) - - d.put(self.mk('new key'), 'new data') - self.assertEqual(d.get(self.mk('new key')), 'new data') - self.assertEqual(d[self.mk('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) - # Hack to avoid conversion by 2to3 tool - rec = getattr(c, "next")() - del c - - self.assertEqual(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() - - self.assertEqual(count, len(d)) - - c.set(self.mk('SS')) - key, value = c.current() - self.checkrec(key, value) - del c - - - def test03_append(self): - # NOTE: this is overridden in RECNO subclass, don't change its name. - if verbose: - print('\n', '-=' * 30) - print("Running %s.test03_append..." % self.__class__.__name__) - - self.assertRaises(dbshelve.DBShelveError, - self.d.append, 'unit test was here') - - - def checkrec(self, key, value): - # override this in a subclass if the key type is different - - import sys - if sys.version_info[0] >= 3 : - if isinstance(key, bytes) : - key = key.decode("iso8859-1") # 8 bits - - x = key[1] - if key[0] == 'S': - self.assertEqual(type(value), str) - self.assertEqual(value, 10 * x) - - elif key[0] == 'I': - self.assertEqual(type(value), int) - self.assertEqual(value, ord(x)) - - elif key[0] == 'L': - self.assertEqual(type(value), list) - self.assertEqual(value, [x] * 10) - - elif key[0] == 'O': - import sys - if sys.version_info[0] < 3 : - from types import InstanceType - self.assertEqual(type(value), InstanceType) - else : - self.assertEqual(type(value), DataClass) - - self.assertEqual(value.S, 10 * x) - self.assertEqual(value.I, ord(x)) - self.assertEqual(value.L, [x] * 10) - - else: - self.assert_(0, '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_HASH - dbflags = db.DB_CREATE - - -class ThreadBTreeShelveTestCase(BasicShelveTestCase): - dbtype = db.DB_BTREE - dbflags = db.DB_CREATE | db.DB_THREAD - - -class ThreadHashShelveTestCase(BasicShelveTestCase): - dbtype = db.DB_HASH - dbflags = db.DB_CREATE | db.DB_THREAD - - -#---------------------------------------------------------------------- - -class BasicEnvShelveTestCase(DBShelveTestCase): - def do_open(self): - self.env = db.DBEnv() - self.env.open(self.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 setUp(self) : - self.homeDir = get_new_environment_path() - DBShelveTestCase.setUp(self) - - def tearDown(self): - import sys - if sys.version_info[0] >= 3 : - from .test_all import do_proxy_db_py3k - do_proxy_db_py3k(self._flag_proxy_db_py3k) - self.do_close() - test_support.rmtree(self.homeDir) - - -class EnvBTreeShelveTestCase(BasicEnvShelveTestCase): - envflags = 0 - dbtype = db.DB_BTREE - dbflags = db.DB_CREATE - - -class EnvHashShelveTestCase(BasicEnvShelveTestCase): - envflags = 0 - dbtype = db.DB_HASH - 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_HASH - dbflags = db.DB_CREATE | db.DB_THREAD - - -#---------------------------------------------------------------------- -# test cases for a DBShelf in a RECNO DB. - -class RecNoShelveTestCase(BasicShelveTestCase): - dbtype = db.DB_RECNO - dbflags = db.DB_CREATE - - def setUp(self): - BasicShelveTestCase.setUp(self) - - # pool to assign integer key values out of - self.key_pool = list(range(1, 5000)) - self.key_map = {} # map string keys to the number we gave them - self.intkey_map = {} # reverse map of above - - def mk(self, key): - if key not in self.key_map: - self.key_map[key] = self.key_pool.pop(0) - self.intkey_map[self.key_map[key]] = key - return self.key_map[key] - - def checkrec(self, intkey, value): - key = self.intkey_map[intkey] - BasicShelveTestCase.checkrec(self, key, value) - - def test03_append(self): - if verbose: - print('\n', '-=' * 30) - print("Running %s.test03_append..." % self.__class__.__name__) - - self.d[1] = 'spam' - self.d[5] = 'eggs' - self.assertEqual(6, self.d.append('spam')) - self.assertEqual(7, self.d.append('baked beans')) - self.assertEqual('spam', self.d.get(6)) - self.assertEqual('spam', self.d.get(1)) - self.assertEqual('baked beans', self.d.get(7)) - self.assertEqual('eggs', self.d.get(5)) - - -#---------------------------------------------------------------------- - -def test_suite(): - suite = unittest.TestSuite() - - suite.addTest(unittest.makeSuite(DBShelveTestCase)) - suite.addTest(unittest.makeSuite(BTreeShelveTestCase)) - suite.addTest(unittest.makeSuite(HashShelveTestCase)) - suite.addTest(unittest.makeSuite(ThreadBTreeShelveTestCase)) - suite.addTest(unittest.makeSuite(ThreadHashShelveTestCase)) - suite.addTest(unittest.makeSuite(EnvBTreeShelveTestCase)) - suite.addTest(unittest.makeSuite(EnvHashShelveTestCase)) - suite.addTest(unittest.makeSuite(EnvThreadBTreeShelveTestCase)) - suite.addTest(unittest.makeSuite(EnvThreadHashShelveTestCase)) - suite.addTest(unittest.makeSuite(RecNoShelveTestCase)) - - return suite - - -if __name__ == '__main__': - unittest.main(defaultTest='test_suite') |