test_bsddb.py 11.4 KB
Newer Older
1
#! /usr/bin/env python
2 3
"""Test script for the bsddb C module by Roger E. Masse
   Adapted to unittest format and expanded scope by Raymond Hettinger
4
"""
5
import os, sys
6 7 8
import unittest
from test import test_support

9 10 11
# Skip test if _bsddb wasn't built.
test_support.import_module('_bsddb')

12 13 14
bsddb = test_support.import_module('bsddb', deprecated=True)
# Just so we know it's imported:
test_support.import_module('dbhash', deprecated=True)
15 16


17
class TestBSDDB(unittest.TestCase):
Gregory P. Smith's avatar
Gregory P. Smith committed
18
    openflag = 'c'
19 20

    def setUp(self):
Gregory P. Smith's avatar
Gregory P. Smith committed
21
        self.f = self.openmethod[0](self.fname, self.openflag, cachesize=32768)
22 23 24 25 26 27 28 29 30
        self.d = dict(q='Guido', w='van', e='Rossum', r='invented', t='Python', y='')
        for k, v in self.d.iteritems():
            self.f[k] = v

    def tearDown(self):
        self.f.sync()
        self.f.close()
        if self.fname is None:
            return
31
        try:
32
            os.remove(self.fname)
33 34
        except os.error:
            pass
35

36 37 38 39 40 41 42 43 44 45
    def test_getitem(self):
        for k, v in self.d.iteritems():
            self.assertEqual(self.f[k], v)

    def test_len(self):
        self.assertEqual(len(self.f), len(self.d))

    def test_change(self):
        self.f['r'] = 'discovered'
        self.assertEqual(self.f['r'], 'discovered')
46 47
        self.assertIn('r', self.f.keys())
        self.assertIn('discovered', self.f.values())
48 49 50 51 52 53 54 55 56 57 58 59

    def test_close_and_reopen(self):
        if self.fname is None:
            # if we're using an in-memory only db, we can't reopen it
            # so finish here.
            return
        self.f.close()
        self.f = self.openmethod[0](self.fname, 'w')
        for k, v in self.d.iteritems():
            self.assertEqual(self.f[k], v)

    def assertSetEquals(self, seqn1, seqn2):
60
        self.assertEqual(set(seqn1), set(seqn2))
61 62 63 64 65 66 67 68 69 70 71 72

    def test_mapping_iteration_methods(self):
        f = self.f
        d = self.d
        self.assertSetEquals(d, f)
        self.assertSetEquals(d.keys(), f.keys())
        self.assertSetEquals(d.values(), f.values())
        self.assertSetEquals(d.items(), f.items())
        self.assertSetEquals(d.iterkeys(), f.iterkeys())
        self.assertSetEquals(d.itervalues(), f.itervalues())
        self.assertSetEquals(d.iteritems(), f.iteritems())

73 74 75 76 77 78 79 80 81 82 83 84
    def test_iter_while_modifying_values(self):
        di = iter(self.d)
        while 1:
            try:
                key = di.next()
                self.d[key] = 'modified '+key
            except StopIteration:
                break

        # it should behave the same as a dict.  modifying values
        # of existing keys should not break iteration.  (adding
        # or removing keys should)
85
        loops_left = len(self.f)
86 87 88 89 90
        fi = iter(self.f)
        while 1:
            try:
                key = fi.next()
                self.f[key] = 'modified '+key
91
                loops_left -= 1
92 93
            except StopIteration:
                break
94
        self.assertEqual(loops_left, 0)
95 96 97

        self.test_mapping_iteration_methods()

98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117
    def test_iter_abort_on_changed_size(self):
        def DictIterAbort():
            di = iter(self.d)
            while 1:
                try:
                    di.next()
                    self.d['newkey'] = 'SPAM'
                except StopIteration:
                    break
        self.assertRaises(RuntimeError, DictIterAbort)

        def DbIterAbort():
            fi = iter(self.f)
            while 1:
                try:
                    fi.next()
                    self.f['newkey'] = 'SPAM'
                except StopIteration:
                    break
        self.assertRaises(RuntimeError, DbIterAbort)
118

119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140
    def test_iteritems_abort_on_changed_size(self):
        def DictIteritemsAbort():
            di = self.d.iteritems()
            while 1:
                try:
                    di.next()
                    self.d['newkey'] = 'SPAM'
                except StopIteration:
                    break
        self.assertRaises(RuntimeError, DictIteritemsAbort)

        def DbIteritemsAbort():
            fi = self.f.iteritems()
            while 1:
                try:
                    key, value = fi.next()
                    del self.f[key]
                except StopIteration:
                    break
        self.assertRaises(RuntimeError, DbIteritemsAbort)

    def test_iteritems_while_modifying_values(self):
141 142 143 144 145 146 147 148 149 150 151
        di = self.d.iteritems()
        while 1:
            try:
                k, v = di.next()
                self.d[k] = 'modified '+v
            except StopIteration:
                break

        # it should behave the same as a dict.  modifying values
        # of existing keys should not break iteration.  (adding
        # or removing keys should)
152
        loops_left = len(self.f)
153 154 155 156 157
        fi = self.f.iteritems()
        while 1:
            try:
                k, v = fi.next()
                self.f[k] = 'modified '+v
158
                loops_left -= 1
159 160
            except StopIteration:
                break
161
        self.assertEqual(loops_left, 0)
162 163 164

        self.test_mapping_iteration_methods()

165 166 167 168 169 170 171 172 173 174 175 176
    def test_first_next_looping(self):
        items = [self.f.first()]
        for i in xrange(1, len(self.f)):
            items.append(self.f.next())
        self.assertSetEquals(items, self.d.items())

    def test_previous_last_looping(self):
        items = [self.f.last()]
        for i in xrange(1, len(self.f)):
            items.append(self.f.previous())
        self.assertSetEquals(items, self.d.items())

Gregory P. Smith's avatar
Gregory P. Smith committed
177 178
    def test_first_while_deleting(self):
        # Test for bug 1725856
179
        self.assertTrue(len(self.d) >= 2, "test requires >=2 items")
Gregory P. Smith's avatar
Gregory P. Smith committed
180 181 182 183 184 185 186
        for _ in self.d:
            key = self.f.first()[0]
            del self.f[key]
        self.assertEqual([], self.f.items(), "expected empty db after test")

    def test_last_while_deleting(self):
        # Test for bug 1725856's evil twin
187
        self.assertTrue(len(self.d) >= 2, "test requires >=2 items")
Gregory P. Smith's avatar
Gregory P. Smith committed
188 189 190 191 192
        for _ in self.d:
            key = self.f.last()[0]
            del self.f[key]
        self.assertEqual([], self.f.items(), "expected empty db after test")

193 194 195 196 197
    def test_set_location(self):
        self.assertEqual(self.f.set_location('e'), ('e', self.d['e']))

    def test_contains(self):
        for k in self.d:
198 199
            self.assertIn(k, self.f)
        self.assertNotIn('not here', self.f)
200 201 202

    def test_has_key(self):
        for k in self.d:
203 204
            self.assertTrue(self.f.has_key(k))
        self.assertTrue(not self.f.has_key('not here'))
205 206 207 208 209

    def test_clear(self):
        self.f.clear()
        self.assertEqual(len(self.f), 0)

210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226
    def test__no_deadlock_first(self, debug=0):
        # do this so that testers can see what function we're in in
        # verbose mode when we deadlock.
        sys.stdout.flush()

        # in pybsddb's _DBWithCursor this causes an internal DBCursor
        # object is created.  Other test_ methods in this class could
        # inadvertently cause the deadlock but an explicit test is needed.
        if debug: print "A"
        k,v = self.f.first()
        if debug: print "B", k
        self.f[k] = "deadlock.  do not pass go.  do not collect $200."
        if debug: print "C"
        # if the bsddb implementation leaves the DBCursor open during
        # the database write and locking+threading support is enabled
        # the cursor's read lock will deadlock the write lock request..

227 228
        # test the iterator interface
        if True:
229
            if debug: print "D"
230 231
            i = self.f.iteritems()
            k,v = i.next()
232 233 234 235 236
            if debug: print "E"
            self.f[k] = "please don't deadlock"
            if debug: print "F"
            while 1:
                try:
237
                    k,v = i.next()
238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255
                except StopIteration:
                    break
            if debug: print "F2"

            i = iter(self.f)
            if debug: print "G"
            while i:
                try:
                    if debug: print "H"
                    k = i.next()
                    if debug: print "I"
                    self.f[k] = "deadlocks-r-us"
                    if debug: print "J"
                except StopIteration:
                    i = None
            if debug: print "K"

        # 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 265 266 267
        nc1 = len(self.f._cursor_refs)
        # create iterator
        i = self.f.iteritems()
        nc2 = len(self.f._cursor_refs)
Neal Norwitz's avatar
Neal Norwitz committed
268
        # use the iterator (should run to the first yield, creating the cursor)
269 270 271 272 273 274 275 276 277
        k, v = i.next()
        nc3 = len(self.f._cursor_refs)
        # destroy the iterator; this should cause the weakref callback
        # to remove the cursor object from self.f._cursor_refs
        del i
        nc4 = len(self.f._cursor_refs)

        self.assertEqual(nc1, nc2)
        self.assertEqual(nc1, nc4)
278
        self.assertTrue(nc3 == nc1+1)
279

280 281
    def test_popitem(self):
        k, v = self.f.popitem()
282 283 284
        self.assertIn(k, self.d)
        self.assertIn(v, self.d.values())
        self.assertNotIn(k, self.f)
285 286 287 288 289 290
        self.assertEqual(len(self.d)-1, len(self.f))

    def test_pop(self):
        k = 'w'
        v = self.f.pop(k)
        self.assertEqual(v, self.d[k])
291 292
        self.assertNotIn(k, self.f)
        self.assertNotIn(v, self.f.values())
293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329
        self.assertEqual(len(self.d)-1, len(self.f))

    def test_get(self):
        self.assertEqual(self.f.get('NotHere'), None)
        self.assertEqual(self.f.get('NotHere', 'Default'), 'Default')
        self.assertEqual(self.f.get('q', 'Default'), self.d['q'])

    def test_setdefault(self):
        self.assertEqual(self.f.setdefault('new', 'dog'), 'dog')
        self.assertEqual(self.f.setdefault('r', 'cat'), self.d['r'])

    def test_update(self):
        new = dict(y='life', u='of', i='brian')
        self.f.update(new)
        self.d.update(new)
        for k, v in self.d.iteritems():
            self.assertEqual(self.f[k], v)

    def test_keyordering(self):
        if self.openmethod[0] is not bsddb.btopen:
            return
        keys = self.d.keys()
        keys.sort()
        self.assertEqual(self.f.first()[0], keys[0])
        self.assertEqual(self.f.next()[0], keys[1])
        self.assertEqual(self.f.last()[0], keys[-1])
        self.assertEqual(self.f.previous()[0], keys[-2])
        self.assertEqual(list(self.f), keys)

class TestBTree(TestBSDDB):
    fname = test_support.TESTFN
    openmethod = [bsddb.btopen]

class TestBTree_InMemory(TestBSDDB):
    fname = None
    openmethod = [bsddb.btopen]

Gregory P. Smith's avatar
Gregory P. Smith committed
330 331 332 333 334
class TestBTree_InMemory_Truncate(TestBSDDB):
    fname = None
    openflag = 'n'
    openmethod = [bsddb.btopen]

335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352
class TestHashTable(TestBSDDB):
    fname = test_support.TESTFN
    openmethod = [bsddb.hashopen]

class TestHashTable_InMemory(TestBSDDB):
    fname = None
    openmethod = [bsddb.hashopen]

##         # (bsddb.rnopen,'Record Numbers'), 'put' for RECNO for bsddb 1.85
##         #                                   appears broken... at least on
##         #                                   Solaris Intel - rmasse 1/97

def test_main(verbose=None):
    test_support.run_unittest(
        TestBTree,
        TestHashTable,
        TestBTree_InMemory,
        TestHashTable_InMemory,
Gregory P. Smith's avatar
Gregory P. Smith committed
353
        TestBTree_InMemory_Truncate,
354 355 356 357
    )

if __name__ == "__main__":
    test_main(verbose=True)