0001 """ 0002 TestCases for checking dbShelve objects. 0003 """ 0004 0005 import sys, os, string 0006 import tempfile, random 0007 from pprint import pprint 0008 from types import * 0009 import unittest 0010 0011 try: 0012 # For Pythons w/distutils pybsddb 0013 from bsddb3 import db, dbshelve 0014 except ImportError: 0015 # For Python 2.3 0016 from bsddb import db, dbshelve 0017 0018 from test_all import verbose 0019 0020 0021 #---------------------------------------------------------------------- 0022 0023 # We want the objects to be comparable so we can test dbshelve.values 0024 # later on. 0025 class DataClass: 0026 def __init__(self): 0027 self.value = random.random() 0028 0029 def __cmp__(self, other): 0030 return cmp(self.value, other) 0031 0032 class DBShelveTestCase(unittest.TestCase): 0033 def setUp(self): 0034 self.filename = tempfile.mktemp() 0035 self.do_open() 0036 0037 def tearDown(self): 0038 self.do_close() 0039 try: 0040 os.remove(self.filename) 0041 except os.error: 0042 pass 0043 0044 def populateDB(self, d): 0045 for x in string.letters: 0046 d['S' + x] = 10 * x # add a string 0047 d['I' + x] = ord(x) # add an integer 0048 d['L' + x] = [x] * 10 # add a list 0049 0050 inst = DataClass() # add an instance 0051 inst.S = 10 * x 0052 inst.I = ord(x) 0053 inst.L = [x] * 10 0054 d['O' + x] = inst 0055 0056 0057 # overridable in derived classes to affect how the shelf is created/opened 0058 def do_open(self): 0059 self.d = dbshelve.open(self.filename) 0060 0061 # and closed... 0062 def do_close(self): 0063 self.d.close() 0064 0065 0066 0067 def test01_basics(self): 0068 if verbose: 0069 print '\n', '-=' * 30 0070 print "Running %s.test01_basics..." % self.__class__.__name__ 0071 0072 self.populateDB(self.d) 0073 self.d.sync() 0074 self.do_close() 0075 self.do_open() 0076 d = self.d 0077 0078 l = len(d) 0079 k = d.keys() 0080 s = d.stat() 0081 f = d.fd() 0082 0083 if verbose: 0084 print "length:", l 0085 print "keys:", k 0086 print "stats:", s 0087 0088 assert 0 == d.has_key('bad key') 0089 assert 1 == d.has_key('IA') 0090 assert 1 == d.has_key('OA') 0091 0092 d.delete('IA') 0093 del d['OA'] 0094 assert 0 == d.has_key('IA') 0095 assert 0 == d.has_key('OA') 0096 assert len(d) == l-2 0097 0098 values = [] 0099 for key in d.keys(): 0100 value = d[key] 0101 values.append(value) 0102 if verbose: 0103 print "%s: %s" % (key, value) 0104 self.checkrec(key, value) 0105 0106 dbvalues = d.values() 0107 assert len(dbvalues) == len(d.keys()) 0108 values.sort() 0109 dbvalues.sort() 0110 assert values == dbvalues 0111 0112 items = d.items() 0113 assert len(items) == len(values) 0114 0115 for key, value in items: 0116 self.checkrec(key, value) 0117 0118 assert d.get('bad key') == None 0119 assert d.get('bad key', None) == None 0120 assert d.get('bad key', 'a string') == 'a string' 0121 assert d.get('bad key', [1, 2, 3]) == [1, 2, 3] 0122 0123 d.set_get_returns_none(0) 0124 self.assertRaises(db.DBNotFoundError, d.get, 'bad key') 0125 d.set_get_returns_none(1) 0126 0127 d.put('new key', 'new data') 0128 assert d.get('new key') == 'new data' 0129 assert d['new key'] == 'new data' 0130 0131 0132 0133 def test02_cursors(self): 0134 if verbose: 0135 print '\n', '-=' * 30 0136 print "Running %s.test02_cursors..." % self.__class__.__name__ 0137 0138 self.populateDB(self.d) 0139 d = self.d 0140 0141 count = 0 0142 c = d.cursor() 0143 rec = c.first() 0144 while rec is not None: 0145 count = count + 1 0146 if verbose: 0147 print rec 0148 key, value = rec 0149 self.checkrec(key, value) 0150 rec = c.next() 0151 del c 0152 0153 assert count == len(d) 0154 0155 count = 0 0156 c = d.cursor() 0157 rec = c.last() 0158 while rec is not None: 0159 count = count + 1 0160 if verbose: 0161 print rec 0162 key, value = rec 0163 self.checkrec(key, value) 0164 rec = c.prev() 0165 0166 assert count == len(d) 0167 0168 c.set('SS') 0169 key, value = c.current() 0170 self.checkrec(key, value) 0171 del c 0172 0173 0174 0175 def checkrec(self, key, value): 0176 x = key[1] 0177 if key[0] == 'S': 0178 assert type(value) == StringType 0179 assert value == 10 * x 0180 0181 elif key[0] == 'I': 0182 assert type(value) == IntType 0183 assert value == ord(x) 0184 0185 elif key[0] == 'L': 0186 assert type(value) == ListType 0187 assert value == [x] * 10 0188 0189 elif key[0] == 'O': 0190 assert type(value) == InstanceType 0191 assert value.S == 10 * x 0192 assert value.I == ord(x) 0193 assert value.L == [x] * 10 0194 0195 else: 0196 raise AssertionError, 'Unknown key type, fix the test' 0197 0198 #---------------------------------------------------------------------- 0199 0200 class BasicShelveTestCase(DBShelveTestCase): 0201 def do_open(self): 0202 self.d = dbshelve.DBShelf() 0203 self.d.open(self.filename, self.dbtype, self.dbflags) 0204 0205 def do_close(self): 0206 self.d.close() 0207 0208 0209 class BTreeShelveTestCase(BasicShelveTestCase): 0210 dbtype = db.DB_BTREE 0211 dbflags = db.DB_CREATE 0212 0213 0214 class HashShelveTestCase(BasicShelveTestCase): 0215 dbtype = db.DB_HASH 0216 dbflags = db.DB_CREATE 0217 0218 0219 class ThreadBTreeShelveTestCase(BasicShelveTestCase): 0220 dbtype = db.DB_BTREE 0221 dbflags = db.DB_CREATE | db.DB_THREAD 0222 0223 0224 class ThreadHashShelveTestCase(BasicShelveTestCase): 0225 dbtype = db.DB_HASH 0226 dbflags = db.DB_CREATE | db.DB_THREAD 0227 0228 0229 #---------------------------------------------------------------------- 0230 0231 class BasicEnvShelveTestCase(DBShelveTestCase): 0232 def do_open(self): 0233 self.homeDir = homeDir = os.path.join( 0234 os.path.dirname(sys.argv[0]), 'db_home') 0235 try: os.mkdir(homeDir) 0236 except os.error: pass 0237 self.env = db.DBEnv() 0238 self.env.open(homeDir, self.envflags | db.DB_INIT_MPOOL | db.DB_CREATE) 0239 0240 self.filename = os.path.split(self.filename)[1] 0241 self.d = dbshelve.DBShelf(self.env) 0242 self.d.open(self.filename, self.dbtype, self.dbflags) 0243 0244 0245 def do_close(self): 0246 self.d.close() 0247 self.env.close() 0248 0249 0250 def tearDown(self): 0251 self.do_close() 0252 import glob 0253 files = glob.glob(os.path.join(self.homeDir, '*')) 0254 for file in files: 0255 os.remove(file) 0256 0257 0258 0259 class EnvBTreeShelveTestCase(BasicEnvShelveTestCase): 0260 envflags = 0 0261 dbtype = db.DB_BTREE 0262 dbflags = db.DB_CREATE 0263 0264 0265 class EnvHashShelveTestCase(BasicEnvShelveTestCase): 0266 envflags = 0 0267 dbtype = db.DB_HASH 0268 dbflags = db.DB_CREATE 0269 0270 0271 class EnvThreadBTreeShelveTestCase(BasicEnvShelveTestCase): 0272 envflags = db.DB_THREAD 0273 dbtype = db.DB_BTREE 0274 dbflags = db.DB_CREATE | db.DB_THREAD 0275 0276 0277 class EnvThreadHashShelveTestCase(BasicEnvShelveTestCase): 0278 envflags = db.DB_THREAD 0279 dbtype = db.DB_HASH 0280 dbflags = db.DB_CREATE | db.DB_THREAD 0281 0282 0283 #---------------------------------------------------------------------- 0284 # TODO: Add test cases for a DBShelf in a RECNO DB. 0285 0286 0287 #---------------------------------------------------------------------- 0288 0289 def test_suite(): 0290 suite = unittest.TestSuite() 0291 0292 suite.addTest(unittest.makeSuite(DBShelveTestCase)) 0293 suite.addTest(unittest.makeSuite(BTreeShelveTestCase)) 0294 suite.addTest(unittest.makeSuite(HashShelveTestCase)) 0295 suite.addTest(unittest.makeSuite(ThreadBTreeShelveTestCase)) 0296 suite.addTest(unittest.makeSuite(ThreadHashShelveTestCase)) 0297 suite.addTest(unittest.makeSuite(EnvBTreeShelveTestCase)) 0298 suite.addTest(unittest.makeSuite(EnvHashShelveTestCase)) 0299 suite.addTest(unittest.makeSuite(EnvThreadBTreeShelveTestCase)) 0300 suite.addTest(unittest.makeSuite(EnvThreadHashShelveTestCase)) 0301 0302 return suite 0303 0304 0305 if __name__ == '__main__': 0306 unittest.main(defaultTest='test_suite') 0307
Generated by PyXR 0.9.4