0001 #! /usr/bin/env python 0002 """Test script for the bsddb C module by Roger E. Masse 0003 Adapted to unittest format and expanded scope by Raymond Hettinger 0004 """ 0005 import os, sys 0006 import copy 0007 import bsddb 0008 import dbhash # Just so we know it's imported 0009 import unittest 0010 from test import test_support 0011 from sets import Set 0012 0013 class TestBSDDB(unittest.TestCase): 0014 0015 def setUp(self): 0016 self.f = self.openmethod[0](self.fname, 'c') 0017 self.d = dict(q='Guido', w='van', e='Rossum', r='invented', t='Python', y='') 0018 for k, v in self.d.iteritems(): 0019 self.f[k] = v 0020 0021 def tearDown(self): 0022 self.f.sync() 0023 self.f.close() 0024 if self.fname is None: 0025 return 0026 try: 0027 os.remove(self.fname) 0028 except os.error: 0029 pass 0030 0031 def test_getitem(self): 0032 for k, v in self.d.iteritems(): 0033 self.assertEqual(self.f[k], v) 0034 0035 def test_len(self): 0036 self.assertEqual(len(self.f), len(self.d)) 0037 0038 def test_change(self): 0039 self.f['r'] = 'discovered' 0040 self.assertEqual(self.f['r'], 'discovered') 0041 self.assert_('r' in self.f.keys()) 0042 self.assert_('discovered' in self.f.values()) 0043 0044 def test_close_and_reopen(self): 0045 if self.fname is None: 0046 # if we're using an in-memory only db, we can't reopen it 0047 # so finish here. 0048 return 0049 self.f.close() 0050 self.f = self.openmethod[0](self.fname, 'w') 0051 for k, v in self.d.iteritems(): 0052 self.assertEqual(self.f[k], v) 0053 0054 def assertSetEquals(self, seqn1, seqn2): 0055 self.assertEqual(Set(seqn1), Set(seqn2)) 0056 0057 def test_mapping_iteration_methods(self): 0058 f = self.f 0059 d = self.d 0060 self.assertSetEquals(d, f) 0061 self.assertSetEquals(d.keys(), f.keys()) 0062 self.assertSetEquals(d.values(), f.values()) 0063 self.assertSetEquals(d.items(), f.items()) 0064 self.assertSetEquals(d.iterkeys(), f.iterkeys()) 0065 self.assertSetEquals(d.itervalues(), f.itervalues()) 0066 self.assertSetEquals(d.iteritems(), f.iteritems()) 0067 0068 def test_iter_while_modifying_values(self): 0069 if not hasattr(self.f, '__iter__'): 0070 return 0071 0072 di = iter(self.d) 0073 while 1: 0074 try: 0075 key = di.next() 0076 self.d[key] = 'modified '+key 0077 except StopIteration: 0078 break 0079 0080 # it should behave the same as a dict. modifying values 0081 # of existing keys should not break iteration. (adding 0082 # or removing keys should) 0083 fi = iter(self.f) 0084 while 1: 0085 try: 0086 key = fi.next() 0087 self.f[key] = 'modified '+key 0088 except StopIteration: 0089 break 0090 0091 self.test_mapping_iteration_methods() 0092 0093 def test_iteritems_while_modifying_values(self): 0094 if not hasattr(self.f, 'iteritems'): 0095 return 0096 0097 di = self.d.iteritems() 0098 while 1: 0099 try: 0100 k, v = di.next() 0101 self.d[k] = 'modified '+v 0102 except StopIteration: 0103 break 0104 0105 # it should behave the same as a dict. modifying values 0106 # of existing keys should not break iteration. (adding 0107 # or removing keys should) 0108 fi = self.f.iteritems() 0109 while 1: 0110 try: 0111 k, v = fi.next() 0112 self.f[k] = 'modified '+v 0113 except StopIteration: 0114 break 0115 0116 self.test_mapping_iteration_methods() 0117 0118 def test_first_next_looping(self): 0119 items = [self.f.first()] 0120 for i in xrange(1, len(self.f)): 0121 items.append(self.f.next()) 0122 self.assertSetEquals(items, self.d.items()) 0123 0124 def test_previous_last_looping(self): 0125 items = [self.f.last()] 0126 for i in xrange(1, len(self.f)): 0127 items.append(self.f.previous()) 0128 self.assertSetEquals(items, self.d.items()) 0129 0130 def test_set_location(self): 0131 self.assertEqual(self.f.set_location('e'), ('e', self.d['e'])) 0132 0133 def test_contains(self): 0134 for k in self.d: 0135 self.assert_(k in self.f) 0136 self.assert_('not here' not in self.f) 0137 0138 def test_has_key(self): 0139 for k in self.d: 0140 self.assert_(self.f.has_key(k)) 0141 self.assert_(not self.f.has_key('not here')) 0142 0143 def test_clear(self): 0144 self.f.clear() 0145 self.assertEqual(len(self.f), 0) 0146 0147 def test__no_deadlock_first(self, debug=0): 0148 # do this so that testers can see what function we're in in 0149 # verbose mode when we deadlock. 0150 sys.stdout.flush() 0151 0152 # in pybsddb's _DBWithCursor this causes an internal DBCursor 0153 # object is created. Other test_ methods in this class could 0154 # inadvertently cause the deadlock but an explicit test is needed. 0155 if debug: print "A" 0156 k,v = self.f.first() 0157 if debug: print "B", k 0158 self.f[k] = "deadlock. do not pass go. do not collect $200." 0159 if debug: print "C" 0160 # if the bsddb implementation leaves the DBCursor open during 0161 # the database write and locking+threading support is enabled 0162 # the cursor's read lock will deadlock the write lock request.. 0163 0164 # test the iterator interface (if present) 0165 if hasattr(self.f, 'iteritems'): 0166 if debug: print "D" 0167 i = self.f.iteritems() 0168 k,v = i.next() 0169 if debug: print "E" 0170 self.f[k] = "please don't deadlock" 0171 if debug: print "F" 0172 while 1: 0173 try: 0174 k,v = i.next() 0175 except StopIteration: 0176 break 0177 if debug: print "F2" 0178 0179 i = iter(self.f) 0180 if debug: print "G" 0181 while i: 0182 try: 0183 if debug: print "H" 0184 k = i.next() 0185 if debug: print "I" 0186 self.f[k] = "deadlocks-r-us" 0187 if debug: print "J" 0188 except StopIteration: 0189 i = None 0190 if debug: print "K" 0191 0192 # test the legacy cursor interface mixed with writes 0193 self.assert_(self.f.first()[0] in self.d) 0194 k = self.f.next()[0] 0195 self.assert_(k in self.d) 0196 self.f[k] = "be gone with ye deadlocks" 0197 self.assert_(self.f[k], "be gone with ye deadlocks") 0198 0199 def test_for_cursor_memleak(self): 0200 if not hasattr(self.f, 'iteritems'): 0201 return 0202 0203 # do the bsddb._DBWithCursor _iter_mixin internals leak cursors? 0204 nc1 = len(self.f._cursor_refs) 0205 # create iterator 0206 i = self.f.iteritems() 0207 nc2 = len(self.f._cursor_refs) 0208 # use the iterator (should run to the first yeild, creating the cursor) 0209 k, v = i.next() 0210 nc3 = len(self.f._cursor_refs) 0211 # destroy the iterator; this should cause the weakref callback 0212 # to remove the cursor object from self.f._cursor_refs 0213 del i 0214 nc4 = len(self.f._cursor_refs) 0215 0216 self.assertEqual(nc1, nc2) 0217 self.assertEqual(nc1, nc4) 0218 self.assert_(nc3 == nc1+1) 0219 0220 def test_popitem(self): 0221 k, v = self.f.popitem() 0222 self.assert_(k in self.d) 0223 self.assert_(v in self.d.values()) 0224 self.assert_(k not in self.f) 0225 self.assertEqual(len(self.d)-1, len(self.f)) 0226 0227 def test_pop(self): 0228 k = 'w' 0229 v = self.f.pop(k) 0230 self.assertEqual(v, self.d[k]) 0231 self.assert_(k not in self.f) 0232 self.assert_(v not in self.f.values()) 0233 self.assertEqual(len(self.d)-1, len(self.f)) 0234 0235 def test_get(self): 0236 self.assertEqual(self.f.get('NotHere'), None) 0237 self.assertEqual(self.f.get('NotHere', 'Default'), 'Default') 0238 self.assertEqual(self.f.get('q', 'Default'), self.d['q']) 0239 0240 def test_setdefault(self): 0241 self.assertEqual(self.f.setdefault('new', 'dog'), 'dog') 0242 self.assertEqual(self.f.setdefault('r', 'cat'), self.d['r']) 0243 0244 def test_update(self): 0245 new = dict(y='life', u='of', i='brian') 0246 self.f.update(new) 0247 self.d.update(new) 0248 for k, v in self.d.iteritems(): 0249 self.assertEqual(self.f[k], v) 0250 0251 def test_keyordering(self): 0252 if self.openmethod[0] is not bsddb.btopen: 0253 return 0254 keys = self.d.keys() 0255 keys.sort() 0256 self.assertEqual(self.f.first()[0], keys[0]) 0257 self.assertEqual(self.f.next()[0], keys[1]) 0258 self.assertEqual(self.f.last()[0], keys[-1]) 0259 self.assertEqual(self.f.previous()[0], keys[-2]) 0260 self.assertEqual(list(self.f), keys) 0261 0262 class TestBTree(TestBSDDB): 0263 fname = test_support.TESTFN 0264 openmethod = [bsddb.btopen] 0265 0266 class TestBTree_InMemory(TestBSDDB): 0267 fname = None 0268 openmethod = [bsddb.btopen] 0269 0270 class TestHashTable(TestBSDDB): 0271 fname = test_support.TESTFN 0272 openmethod = [bsddb.hashopen] 0273 0274 class TestHashTable_InMemory(TestBSDDB): 0275 fname = None 0276 openmethod = [bsddb.hashopen] 0277 0278 ## # (bsddb.rnopen,'Record Numbers'), 'put' for RECNO for bsddb 1.85 0279 ## # appears broken... at least on 0280 ## # Solaris Intel - rmasse 1/97 0281 0282 def test_main(verbose=None): 0283 test_support.run_unittest( 0284 TestBTree, 0285 TestHashTable, 0286 TestBTree_InMemory, 0287 TestHashTable_InMemory, 0288 ) 0289 0290 if __name__ == "__main__": 0291 test_main(verbose=True) 0292
Generated by PyXR 0.9.4