mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-25 21:11:09 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Fix issue2669: bsddb simple/legacy interface iteration silently fails

Browse source 
when database changes size during iteration.

It now behaves like a dictionary, the next attempt to get a value from
the iterator after the database has changed size will raise a RuntimeError.
This commit is contained in:
Gregory P. Smith 2008-05-25 08:28:29 +00:00
parent e08e3d0686
commit 9e6468be1d
2 changed files with 122 additions and 69 deletions
Download patch file Download diff file Expand all files Collapse all files

63
Lib/test/test_bsddb.py
View file

@ -66,9 +66,6 @@ class TestBSDDB(unittest.TestCase):
self.assertSetEquals(d.iteritems(), f.iteritems())
def test_iter_while_modifying_values(self):
if not hasattr(self.f, '__iter__'):
return
di = iter(self.d)
while 1:
try:
@ -80,20 +77,62 @@ class TestBSDDB(unittest.TestCase):
# it should behave the same as a dict. modifying values
# of existing keys should not break iteration. (adding
# or removing keys should)
loops_left = len(self.f)
fi = iter(self.f)
while 1:
try:
key = fi.next()
self.f[key] = 'modified '+key
loops_left -= 1
except StopIteration:
break
self.assertEqual(loops_left, 0)
self.test_mapping_iteration_methods()
def test_iteritems_while_modifying_values(self):
if not hasattr(self.f, 'iteritems'):
return
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)
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):
di = self.d.iteritems()
while 1:
try:
@ -105,13 +144,16 @@ class TestBSDDB(unittest.TestCase):
# it should behave the same as a dict. modifying values
# of existing keys should not break iteration. (adding
# or removing keys should)
loops_left = len(self.f)
fi = self.f.iteritems()
while 1:
try:
k, v = fi.next()
self.f[k] = 'modified '+v
loops_left -= 1
except StopIteration:
break
self.assertEqual(loops_left, 0)
self.test_mapping_iteration_methods()
@ -177,8 +219,8 @@ class TestBSDDB(unittest.TestCase):
# the database write and locking+threading support is enabled
# the cursor's read lock will deadlock the write lock request..
# test the iterator interface (if present)
if hasattr(self.f, 'iteritems'):
# test the iterator interface
if True:
if debug: print "D"
i = self.f.iteritems()
k,v = i.next()
@ -213,10 +255,7 @@ class TestBSDDB(unittest.TestCase):
self.assert_(self.f[k], "be gone with ye deadlocks")
def test_for_cursor_memleak(self):
if not hasattr(self.f, 'iteritems'):
return
# do the bsddb._DBWithCursor _iter_mixin internals leak cursors?
# do the bsddb._DBWithCursor iterator internals leak cursors?
nc1 = len(self.f._cursor_refs)
# create iterator
i = self.f.iteritems()