| """ |
| TestCases for checking dbShelve objects. |
| """ |
| |
| import os, string, sys |
| 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): |
| if (sys.version_info < (2, 7)) or ((sys.version_info >= (3, 0)) and |
| (sys.version_info < (3, 2))) : |
| def assertIn(self, a, b, msg=None) : |
| return self.assertTrue(a in b, msg=msg) |
| |
| |
| def setUp(self): |
| 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): |
| 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 |
| 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 = d.keys() |
| s = d.stat() |
| f = d.fd() |
| |
| if verbose: |
| print "length:", l |
| print "keys:", k |
| print "stats:", s |
| |
| self.assertEqual(0, d.has_key(self.mk('bad key'))) |
| self.assertEqual(1, d.has_key(self.mk('IA'))) |
| self.assertEqual(1, d.has_key(self.mk('OA'))) |
| |
| d.delete(self.mk('IA')) |
| del d[self.mk('OA')] |
| self.assertEqual(0, d.has_key(self.mk('IA'))) |
| self.assertEqual(0, d.has_key(self.mk('OA'))) |
| self.assertEqual(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() |
| self.assertEqual(len(dbvalues), len(d.keys())) |
| if sys.version_info < (2, 6) : |
| 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 = 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 test04_iterable(self) : |
| self.populateDB(self.d) |
| d = self.d |
| keys = d.keys() |
| keyset = set(keys) |
| self.assertEqual(len(keyset), len(keys)) |
| |
| for key in d : |
| self.assertIn(key, keyset) |
| keyset.remove(key) |
| self.assertEqual(len(keyset), 0) |
| |
| def checkrec(self, key, value): |
| # override this in a subclass if the key type is different |
| |
| 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': |
| 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.assertTrue(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): |
| 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') |
