Lib/test/test_bsddb.py

   1 #! /usr/bin/env python
   2 """Test script for the bsddb C module by Roger E. Masse
   3    Adapted to unittest format and expanded scope by Raymond Hettinger
   4 """
   5 import os, sys
   6 import unittest
   7 from test import test_support
   8
   9 # Skip test if _bsddb wasn't built.
  10 test_support.import_module('_bsddb')
  11
  12 bsddb = test_support.import_module('bsddb', deprecated=True)
  13 # Just so we know it's imported:
  14 test_support.import_module('dbhash', deprecated=True)
  15
  16
  17 class TestBSDDB(unittest.TestCase):
  18     openflag = 'c'
  19
  20     def setUp(self):
  21         self.f = self.openmethod[0](self.fname, self.openflag, cachesize=32768)
  22         self.d = dict(q='Guido', w='van', e='Rossum', r='invented', t='Python', y='')
  23         for k, v in self.d.iteritems():
  24             self.f[k] = v
  25
  26     def tearDown(self):
  27         self.f.sync()
  28         self.f.close()
  29         if self.fname is None:
  30             return
  31         try:
  32             os.remove(self.fname)
  33         except os.error:
  34             pass
  35
  36     def test_getitem(self):
  37         for k, v in self.d.iteritems():
  38             self.assertEqual(self.f[k], v)
  39
  40     def test_len(self):
  41         self.assertEqual(len(self.f), len(self.d))
  42
  43     def test_change(self):
  44         self.f['r'] = 'discovered'
  45         self.assertEqual(self.f['r'], 'discovered')
  46         self.assertIn('r', self.f.keys())
  47         self.assertIn('discovered', self.f.values())
  48
  49     def test_close_and_reopen(self):
  50         if self.fname is None:
  51             # if we're using an in-memory only db, we can't reopen it
  52             # so finish here.
  53             return
  54         self.f.close()
  55         self.f = self.openmethod[0](self.fname, 'w')
  56         for k, v in self.d.iteritems():
  57             self.assertEqual(self.f[k], v)
  58
  59     def assertSetEquals(self, seqn1, seqn2):
  60         self.assertEqual(set(seqn1), set(seqn2))
  61
  62     def test_mapping_iteration_methods(self):
  63         f = self.f
  64         d = self.d
  65         self.assertSetEquals(d, f)
  66         self.assertSetEquals(d.keys(), f.keys())
  67         self.assertSetEquals(d.values(), f.values())
  68         self.assertSetEquals(d.items(), f.items())
  69         self.assertSetEquals(d.iterkeys(), f.iterkeys())
  70         self.assertSetEquals(d.itervalues(), f.itervalues())
  71         self.assertSetEquals(d.iteritems(), f.iteritems())
  72
  73     def test_iter_while_modifying_values(self):
  74         di = iter(self.d)
  75         while 1:
  76             try:
  77                 key = di.next()
  78                 self.d[key] = 'modified '+key
  79             except StopIteration:
  80                 break
  81
  82         # it should behave the same as a dict.  modifying values
  83         # of existing keys should not break iteration.  (adding
  84         # or removing keys should)
  85         loops_left = len(self.f)
  86         fi = iter(self.f)
  87         while 1:
  88             try:
  89                 key = fi.next()
  90                 self.f[key] = 'modified '+key
  91                 loops_left -= 1
  92             except StopIteration:
  93                 break
  94         self.assertEqual(loops_left, 0)
  95
  96         self.test_mapping_iteration_methods()
  97
  98     def test_iter_abort_on_changed_size(self):
  99         def DictIterAbort():
 100             di = iter(self.d)
 101             while 1:
 102                 try:
 103                     di.next()
 104                     self.d['newkey'] = 'SPAM'
 105                 except StopIteration:
 106                     break
 107         self.assertRaises(RuntimeError, DictIterAbort)
 108
 109         def DbIterAbort():
 110             fi = iter(self.f)
 111             while 1:
 112                 try:
 113                     fi.next()
 114                     self.f['newkey'] = 'SPAM'
 115                 except StopIteration:
 116                     break
 117         self.assertRaises(RuntimeError, DbIterAbort)
 118
 119     def test_iteritems_abort_on_changed_size(self):
 120         def DictIteritemsAbort():
 121             di = self.d.iteritems()
 122             while 1:
 123                 try:
 124                     di.next()
 125                     self.d['newkey'] = 'SPAM'
 126                 except StopIteration:
 127                     break
 128         self.assertRaises(RuntimeError, DictIteritemsAbort)
 129
 130         def DbIteritemsAbort():
 131             fi = self.f.iteritems()
 132             while 1:
 133                 try:
 134                     key, value = fi.next()
 135                     del self.f[key]
 136                 except StopIteration:
 137                     break
 138         self.assertRaises(RuntimeError, DbIteritemsAbort)
 139
 140     def test_iteritems_while_modifying_values(self):
 141         di = self.d.iteritems()
 142         while 1:
 143             try:
 144                 k, v = di.next()
 145                 self.d[k] = 'modified '+v
 146             except StopIteration:
 147                 break
 148
 149         # it should behave the same as a dict.  modifying values
 150         # of existing keys should not break iteration.  (adding
 151         # or removing keys should)
 152         loops_left = len(self.f)
 153         fi = self.f.iteritems()
 154         while 1:
 155             try:
 156                 k, v = fi.next()
 157                 self.f[k] = 'modified '+v
 158                 loops_left -= 1
 159             except StopIteration:
 160                 break
 161         self.assertEqual(loops_left, 0)
 162
 163         self.test_mapping_iteration_methods()
 164
 165     def test_first_next_looping(self):
 166         items = [self.f.first()]
 167         for i in xrange(1, len(self.f)):
 168             items.append(self.f.next())
 169         self.assertSetEquals(items, self.d.items())
 170
 171     def test_previous_last_looping(self):
 172         items = [self.f.last()]
 173         for i in xrange(1, len(self.f)):
 174             items.append(self.f.previous())
 175         self.assertSetEquals(items, self.d.items())
 176
 177     def test_first_while_deleting(self):
 178         # Test for bug 1725856
 179         self.assertTrue(len(self.d) >= 2, "test requires >=2 items")
 180         for _ in self.d:
 181             key = self.f.first()[0]
 182             del self.f[key]
 183         self.assertEqual([], self.f.items(), "expected empty db after test")
 184
 185     def test_last_while_deleting(self):
 186         # Test for bug 1725856's evil twin
 187         self.assertTrue(len(self.d) >= 2, "test requires >=2 items")
 188         for _ in self.d:
 189             key = self.f.last()[0]
 190             del self.f[key]
 191         self.assertEqual([], self.f.items(), "expected empty db after test")
 192
 193     def test_set_location(self):
 194         self.assertEqual(self.f.set_location('e'), ('e', self.d['e']))
 195
 196     def test_contains(self):
 197         for k in self.d:
 198             self.assertIn(k, self.f)
 199         self.assertNotIn('not here', self.f)
 200
 201     def test_has_key(self):
 202         for k in self.d:
 203             self.assertTrue(self.f.has_key(k))
 204         self.assertTrue(not self.f.has_key('not here'))
 205
 206     def test_clear(self):
 207         self.f.clear()
 208         self.assertEqual(len(self.f), 0)
 209
 210     def test__no_deadlock_first(self, debug=0):
 211         # do this so that testers can see what function we're in in
 212         # verbose mode when we deadlock.
 213         sys.stdout.flush()
 214
 215         # in pybsddb's _DBWithCursor this causes an internal DBCursor
 216         # object is created.  Other test_ methods in this class could
 217         # inadvertently cause the deadlock but an explicit test is needed.
 218         if debug: print "A"
 219         k,v = self.f.first()
 220         if debug: print "B", k
 221         self.f[k] = "deadlock.  do not pass go.  do not collect $200."
 222         if debug: print "C"
 223         # if the bsddb implementation leaves the DBCursor open during
 224         # the database write and locking+threading support is enabled
 225         # the cursor's read lock will deadlock the write lock request..
 226
 227         # test the iterator interface
 228         if True:
 229             if debug: print "D"
 230             i = self.f.iteritems()
 231             k,v = i.next()
 232             if debug: print "E"
 233             self.f[k] = "please don't deadlock"
 234             if debug: print "F"
 235             while 1:
 236                 try:
 237                     k,v = i.next()
 238                 except StopIteration:
 239                     break
 240             if debug: print "F2"
 241
 242             i = iter(self.f)
 243             if debug: print "G"
 244             while i:
 245                 try:
 246                     if debug: print "H"
 247                     k = i.next()
 248                     if debug: print "I"
 249                     self.f[k] = "deadlocks-r-us"
 250                     if debug: print "J"
 251                 except StopIteration:
 252                     i = None
 253             if debug: print "K"
 254
 255         # test the legacy cursor interface mixed with writes
 256         self.assertIn(self.f.first()[0], self.d)
 257         k = self.f.next()[0]
 258         self.assertIn(k, self.d)
 259         self.f[k] = "be gone with ye deadlocks"
 260         self.assertTrue(self.f[k], "be gone with ye deadlocks")
 261
 262     def test_for_cursor_memleak(self):
 263         # do the bsddb._DBWithCursor iterator internals leak cursors?
 264         nc1 = len(self.f._cursor_refs)
 265         # create iterator
 266         i = self.f.iteritems()
 267         nc2 = len(self.f._cursor_refs)
 268         # use the iterator (should run to the first yield, creating the cursor)
 269         k, v = i.next()
 270         nc3 = len(self.f._cursor_refs)
 271         # destroy the iterator; this should cause the weakref callback
 272         # to remove the cursor object from self.f._cursor_refs
 273         del i
 274         nc4 = len(self.f._cursor_refs)
 275
 276         self.assertEqual(nc1, nc2)
 277         self.assertEqual(nc1, nc4)
 278         self.assertTrue(nc3 == nc1+1)
 279
 280     def test_popitem(self):
 281         k, v = self.f.popitem()
 282         self.assertIn(k, self.d)
 283         self.assertIn(v, self.d.values())
 284         self.assertNotIn(k, self.f)
 285         self.assertEqual(len(self.d)-1, len(self.f))
 286
 287     def test_pop(self):
 288         k = 'w'
 289         v = self.f.pop(k)
 290         self.assertEqual(v, self.d[k])
 291         self.assertNotIn(k, self.f)
 292         self.assertNotIn(v, self.f.values())
 293         self.assertEqual(len(self.d)-1, len(self.f))
 294
 295     def test_get(self):
 296         self.assertEqual(self.f.get('NotHere'), None)
 297         self.assertEqual(self.f.get('NotHere', 'Default'), 'Default')
 298         self.assertEqual(self.f.get('q', 'Default'), self.d['q'])
 299
 300     def test_setdefault(self):
 301         self.assertEqual(self.f.setdefault('new', 'dog'), 'dog')
 302         self.assertEqual(self.f.setdefault('r', 'cat'), self.d['r'])
 303
 304     def test_update(self):
 305         new = dict(y='life', u='of', i='brian')
 306         self.f.update(new)
 307         self.d.update(new)
 308         for k, v in self.d.iteritems():
 309             self.assertEqual(self.f[k], v)
 310
 311     def test_keyordering(self):
 312         if self.openmethod[0] is not bsddb.btopen:
 313             return
 314         keys = self.d.keys()
 315         keys.sort()
 316         self.assertEqual(self.f.first()[0], keys[0])
 317         self.assertEqual(self.f.next()[0], keys[1])
 318         self.assertEqual(self.f.last()[0], keys[-1])
 319         self.assertEqual(self.f.previous()[0], keys[-2])
 320         self.assertEqual(list(self.f), keys)
 321
 322 class TestBTree(TestBSDDB):
 323     fname = test_support.TESTFN
 324     openmethod = [bsddb.btopen]
 325
 326 class TestBTree_InMemory(TestBSDDB):
 327     fname = None
 328     openmethod = [bsddb.btopen]
 329
 330 class TestBTree_InMemory_Truncate(TestBSDDB):
 331     fname = None
 332     openflag = 'n'
 333     openmethod = [bsddb.btopen]
 334
 335 class TestHashTable(TestBSDDB):
 336     fname = test_support.TESTFN
 337     openmethod = [bsddb.hashopen]
 338
 339 class TestHashTable_InMemory(TestBSDDB):
 340     fname = None
 341     openmethod = [bsddb.hashopen]
 342
 343 ##         # (bsddb.rnopen,'Record Numbers'), 'put' for RECNO for bsddb 1.85
 344 ##         #                                   appears broken... at least on
 345 ##         #                                   Solaris Intel - rmasse 1/97
 346
 347 def test_main(verbose=None):
 348     test_support.run_unittest(
 349         TestBTree,
 350         TestHashTable,
 351         TestBTree_InMemory,
 352         TestHashTable_InMemory,
 353         TestBTree_InMemory_Truncate,
 354     )
 355
 356 if __name__ == "__main__":
 357     test_main(verbose=True)