PyXR

c:\python24\lib \ test \ test_bsddb.py



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
SourceForge.net Logo