blob: 513e5415d7f6bfa7c6b233a385d8f2338cc73983 [file] [log] [blame]
Roger E. Masse352e1861997-01-16 22:04:10 +00001#! /usr/bin/env python
Raymond Hettingerd55111f2003-09-13 05:51:09 +00002"""Test script for the bsddb C module by Roger E. Masse
3 Adapted to unittest format and expanded scope by Raymond Hettinger
Roger E. Masse352e1861997-01-16 22:04:10 +00004"""
Gregory P. Smithdc113a82003-11-02 09:10:16 +00005import os, sys
Gregory P. Smitha703a212003-11-03 01:04:41 +00006import copy
Roger E. Masse352e1861997-01-16 22:04:10 +00007import bsddb
Guido van Rossuma35e2ce2001-12-07 16:43:19 +00008import dbhash # Just so we know it's imported
Raymond Hettingerd55111f2003-09-13 05:51:09 +00009import unittest
10from test import test_support
11from sets import Set
Roger E. Masse352e1861997-01-16 22:04:10 +000012
Raymond Hettingerd55111f2003-09-13 05:51:09 +000013class TestBSDDB(unittest.TestCase):
Gregory P. Smith14c6b462006-04-12 20:16:56 +000014 openflag = 'c'
Roger E. Masse352e1861997-01-16 22:04:10 +000015
Raymond Hettingerd55111f2003-09-13 05:51:09 +000016 def setUp(self):
Gregory P. Smith64029982006-04-12 20:35:02 +000017 self.f = self.openmethod[0](self.fname, self.openflag, cachesize=32768)
Raymond Hettingerd55111f2003-09-13 05:51:09 +000018 self.d = dict(q='Guido', w='van', e='Rossum', r='invented', t='Python', y='')
19 for k, v in self.d.iteritems():
20 self.f[k] = v
Fred Drake004d5e62000-10-23 17:22:08 +000021
Raymond Hettingerd55111f2003-09-13 05:51:09 +000022 def tearDown(self):
23 self.f.sync()
24 self.f.close()
25 if self.fname is None:
26 return
Anthony Baxter83888952002-04-23 02:11:05 +000027 try:
Raymond Hettingerd55111f2003-09-13 05:51:09 +000028 os.remove(self.fname)
Anthony Baxter83888952002-04-23 02:11:05 +000029 except os.error:
30 pass
Roger E. Masse352e1861997-01-16 22:04:10 +000031
Raymond Hettingerd55111f2003-09-13 05:51:09 +000032 def test_getitem(self):
33 for k, v in self.d.iteritems():
34 self.assertEqual(self.f[k], v)
Roger E. Masse352e1861997-01-16 22:04:10 +000035
Raymond Hettingerd55111f2003-09-13 05:51:09 +000036 def test_len(self):
37 self.assertEqual(len(self.f), len(self.d))
38
39 def test_change(self):
40 self.f['r'] = 'discovered'
41 self.assertEqual(self.f['r'], 'discovered')
42 self.assert_('r' in self.f.keys())
43 self.assert_('discovered' in self.f.values())
44
45 def test_close_and_reopen(self):
46 if self.fname is None:
47 # if we're using an in-memory only db, we can't reopen it
48 # so finish here.
49 return
50 self.f.close()
51 self.f = self.openmethod[0](self.fname, 'w')
52 for k, v in self.d.iteritems():
53 self.assertEqual(self.f[k], v)
54
55 def assertSetEquals(self, seqn1, seqn2):
56 self.assertEqual(Set(seqn1), Set(seqn2))
57
58 def test_mapping_iteration_methods(self):
59 f = self.f
60 d = self.d
61 self.assertSetEquals(d, f)
62 self.assertSetEquals(d.keys(), f.keys())
63 self.assertSetEquals(d.values(), f.values())
64 self.assertSetEquals(d.items(), f.items())
65 self.assertSetEquals(d.iterkeys(), f.iterkeys())
66 self.assertSetEquals(d.itervalues(), f.itervalues())
67 self.assertSetEquals(d.iteritems(), f.iteritems())
68
Gregory P. Smitha703a212003-11-03 01:04:41 +000069 def test_iter_while_modifying_values(self):
70 if not hasattr(self.f, '__iter__'):
71 return
72
73 di = iter(self.d)
74 while 1:
75 try:
76 key = di.next()
77 self.d[key] = 'modified '+key
78 except StopIteration:
79 break
80
81 # it should behave the same as a dict. modifying values
82 # of existing keys should not break iteration. (adding
83 # or removing keys should)
84 fi = iter(self.f)
85 while 1:
86 try:
87 key = fi.next()
88 self.f[key] = 'modified '+key
89 except StopIteration:
90 break
91
92 self.test_mapping_iteration_methods()
93
94 def test_iteritems_while_modifying_values(self):
95 if not hasattr(self.f, 'iteritems'):
96 return
97
98 di = self.d.iteritems()
99 while 1:
100 try:
101 k, v = di.next()
102 self.d[k] = 'modified '+v
103 except StopIteration:
104 break
105
106 # it should behave the same as a dict. modifying values
107 # of existing keys should not break iteration. (adding
108 # or removing keys should)
109 fi = self.f.iteritems()
110 while 1:
111 try:
112 k, v = fi.next()
113 self.f[k] = 'modified '+v
114 except StopIteration:
115 break
116
117 self.test_mapping_iteration_methods()
118
Raymond Hettingerd55111f2003-09-13 05:51:09 +0000119 def test_first_next_looping(self):
120 items = [self.f.first()]
121 for i in xrange(1, len(self.f)):
122 items.append(self.f.next())
123 self.assertSetEquals(items, self.d.items())
124
125 def test_previous_last_looping(self):
126 items = [self.f.last()]
127 for i in xrange(1, len(self.f)):
128 items.append(self.f.previous())
129 self.assertSetEquals(items, self.d.items())
130
131 def test_set_location(self):
132 self.assertEqual(self.f.set_location('e'), ('e', self.d['e']))
133
134 def test_contains(self):
135 for k in self.d:
136 self.assert_(k in self.f)
137 self.assert_('not here' not in self.f)
138
139 def test_has_key(self):
140 for k in self.d:
141 self.assert_(self.f.has_key(k))
142 self.assert_(not self.f.has_key('not here'))
143
144 def test_clear(self):
145 self.f.clear()
146 self.assertEqual(len(self.f), 0)
147
Gregory P. Smithdc113a82003-11-02 09:10:16 +0000148 def test__no_deadlock_first(self, debug=0):
149 # do this so that testers can see what function we're in in
150 # verbose mode when we deadlock.
151 sys.stdout.flush()
152
153 # in pybsddb's _DBWithCursor this causes an internal DBCursor
154 # object is created. Other test_ methods in this class could
155 # inadvertently cause the deadlock but an explicit test is needed.
156 if debug: print "A"
157 k,v = self.f.first()
158 if debug: print "B", k
159 self.f[k] = "deadlock. do not pass go. do not collect $200."
160 if debug: print "C"
161 # if the bsddb implementation leaves the DBCursor open during
162 # the database write and locking+threading support is enabled
163 # the cursor's read lock will deadlock the write lock request..
164
165 # test the iterator interface (if present)
Gregory P. Smitha703a212003-11-03 01:04:41 +0000166 if hasattr(self.f, 'iteritems'):
Gregory P. Smithdc113a82003-11-02 09:10:16 +0000167 if debug: print "D"
Gregory P. Smitha703a212003-11-03 01:04:41 +0000168 i = self.f.iteritems()
169 k,v = i.next()
Gregory P. Smithdc113a82003-11-02 09:10:16 +0000170 if debug: print "E"
171 self.f[k] = "please don't deadlock"
172 if debug: print "F"
173 while 1:
174 try:
Gregory P. Smitha703a212003-11-03 01:04:41 +0000175 k,v = i.next()
Gregory P. Smithdc113a82003-11-02 09:10:16 +0000176 except StopIteration:
177 break
178 if debug: print "F2"
179
180 i = iter(self.f)
181 if debug: print "G"
182 while i:
183 try:
184 if debug: print "H"
185 k = i.next()
186 if debug: print "I"
187 self.f[k] = "deadlocks-r-us"
188 if debug: print "J"
189 except StopIteration:
190 i = None
191 if debug: print "K"
192
193 # test the legacy cursor interface mixed with writes
194 self.assert_(self.f.first()[0] in self.d)
195 k = self.f.next()[0]
196 self.assert_(k in self.d)
197 self.f[k] = "be gone with ye deadlocks"
198 self.assert_(self.f[k], "be gone with ye deadlocks")
199
Gregory P. Smitha703a212003-11-03 01:04:41 +0000200 def test_for_cursor_memleak(self):
201 if not hasattr(self.f, 'iteritems'):
202 return
203
204 # do the bsddb._DBWithCursor _iter_mixin internals leak cursors?
205 nc1 = len(self.f._cursor_refs)
206 # create iterator
207 i = self.f.iteritems()
208 nc2 = len(self.f._cursor_refs)
209 # use the iterator (should run to the first yeild, creating the cursor)
210 k, v = i.next()
211 nc3 = len(self.f._cursor_refs)
212 # destroy the iterator; this should cause the weakref callback
213 # to remove the cursor object from self.f._cursor_refs
214 del i
215 nc4 = len(self.f._cursor_refs)
216
217 self.assertEqual(nc1, nc2)
218 self.assertEqual(nc1, nc4)
219 self.assert_(nc3 == nc1+1)
220
Raymond Hettingerd55111f2003-09-13 05:51:09 +0000221 def test_popitem(self):
222 k, v = self.f.popitem()
223 self.assert_(k in self.d)
224 self.assert_(v in self.d.values())
225 self.assert_(k not in self.f)
226 self.assertEqual(len(self.d)-1, len(self.f))
227
228 def test_pop(self):
229 k = 'w'
230 v = self.f.pop(k)
231 self.assertEqual(v, self.d[k])
232 self.assert_(k not in self.f)
233 self.assert_(v not in self.f.values())
234 self.assertEqual(len(self.d)-1, len(self.f))
235
236 def test_get(self):
237 self.assertEqual(self.f.get('NotHere'), None)
238 self.assertEqual(self.f.get('NotHere', 'Default'), 'Default')
239 self.assertEqual(self.f.get('q', 'Default'), self.d['q'])
240
241 def test_setdefault(self):
242 self.assertEqual(self.f.setdefault('new', 'dog'), 'dog')
243 self.assertEqual(self.f.setdefault('r', 'cat'), self.d['r'])
244
245 def test_update(self):
246 new = dict(y='life', u='of', i='brian')
247 self.f.update(new)
248 self.d.update(new)
249 for k, v in self.d.iteritems():
250 self.assertEqual(self.f[k], v)
251
252 def test_keyordering(self):
253 if self.openmethod[0] is not bsddb.btopen:
254 return
255 keys = self.d.keys()
256 keys.sort()
257 self.assertEqual(self.f.first()[0], keys[0])
258 self.assertEqual(self.f.next()[0], keys[1])
259 self.assertEqual(self.f.last()[0], keys[-1])
260 self.assertEqual(self.f.previous()[0], keys[-2])
261 self.assertEqual(list(self.f), keys)
262
263class TestBTree(TestBSDDB):
264 fname = test_support.TESTFN
265 openmethod = [bsddb.btopen]
266
267class TestBTree_InMemory(TestBSDDB):
268 fname = None
269 openmethod = [bsddb.btopen]
270
Gregory P. Smith14c6b462006-04-12 20:16:56 +0000271class TestBTree_InMemory_Truncate(TestBSDDB):
272 fname = None
273 openflag = 'n'
274 openmethod = [bsddb.btopen]
275
Raymond Hettingerd55111f2003-09-13 05:51:09 +0000276class TestHashTable(TestBSDDB):
277 fname = test_support.TESTFN
278 openmethod = [bsddb.hashopen]
279
280class TestHashTable_InMemory(TestBSDDB):
281 fname = None
282 openmethod = [bsddb.hashopen]
283
284## # (bsddb.rnopen,'Record Numbers'), 'put' for RECNO for bsddb 1.85
285## # appears broken... at least on
286## # Solaris Intel - rmasse 1/97
287
288def test_main(verbose=None):
289 test_support.run_unittest(
290 TestBTree,
291 TestHashTable,
292 TestBTree_InMemory,
293 TestHashTable_InMemory,
Gregory P. Smith14c6b462006-04-12 20:16:56 +0000294 TestBTree_InMemory_Truncate,
Raymond Hettingerd55111f2003-09-13 05:51:09 +0000295 )
296
297if __name__ == "__main__":
298 test_main(verbose=True)