Lib/bsddb/test/test_dbshelve.py

   1 """
   2 TestCases for checking dbShelve objects.
   3 """
   4
   5 import sys, os, string
   6 import tempfile, random
   7 from pprint import pprint
   8 from types import *
   9 import unittest
  10
  11 from bsddb import dbshelve, db
  12
  13 from test_all import verbose
  14
  15
  16 #----------------------------------------------------------------------
  17
  18 # We want the objects to be comparable so we can test dbshelve.values
  19 # later on.
  20 class DataClass:
  21     def __init__(self):
  22         self.value = random.random()
  23
  24     def __cmp__(self, other):
  25         return cmp(self.value, other)
  26
  27 class DBShelveTestCase(unittest.TestCase):
  28     def setUp(self):
  29         self.filename = tempfile.mktemp()
  30         self.do_open()
  31
  32     def tearDown(self):
  33         self.do_close()
  34         try:
  35             os.remove(self.filename)
  36         except os.error:
  37             pass
  38
  39     def populateDB(self, d):
  40         for x in string.letters:
  41             d['S' + x] = 10 * x           # add a string
  42             d['I' + x] = ord(x)           # add an integer
  43             d['L' + x] = [x] * 10         # add a list
  44
  45             inst = DataClass()            # add an instance
  46             inst.S = 10 * x
  47             inst.I = ord(x)
  48             inst.L = [x] * 10
  49             d['O' + x] = inst
  50
  51
  52     # overridable in derived classes to affect how the shelf is created/opened
  53     def do_open(self):
  54         self.d = dbshelve.open(self.filename)
  55
  56     # and closed...
  57     def do_close(self):
  58         self.d.close()
  59
  60
  61
  62     def test01_basics(self):
  63         if verbose:
  64             print '\n', '-=' * 30
  65             print "Running %s.test01_basics..." % self.__class__.__name__
  66
  67         self.populateDB(self.d)
  68         self.d.sync()
  69         self.do_close()
  70         self.do_open()
  71         d = self.d
  72
  73         l = len(d)
  74         k = d.keys()
  75         s = d.stat()
  76         f = d.fd()
  77
  78         if verbose:
  79             print "length:", l
  80             print "keys:", k
  81             print "stats:", s
  82
  83         assert 0 == d.has_key('bad key')
  84         assert 1 == d.has_key('IA')
  85         assert 1 == d.has_key('OA')
  86
  87         d.delete('IA')
  88         del d['OA']
  89         assert 0 == d.has_key('IA')
  90         assert 0 == d.has_key('OA')
  91         assert len(d) == l-2
  92
  93         values = []
  94         for key in d.keys():
  95             value = d[key]
  96             values.append(value)
  97             if verbose:
  98                 print "%s: %s" % (key, value)
  99             self.checkrec(key, value)
 100
 101         dbvalues = d.values()
 102         assert len(dbvalues) == len(d.keys())
 103         values.sort()
 104         dbvalues.sort()
 105         assert values == dbvalues
 106
 107         items = d.items()
 108         assert len(items) == len(values)
 109
 110         for key, value in items:
 111             self.checkrec(key, value)
 112
 113         assert d.get('bad key') == None
 114         assert d.get('bad key', None) == None
 115         assert d.get('bad key', 'a string') == 'a string'
 116         assert d.get('bad key', [1, 2, 3]) == [1, 2, 3]
 117
 118         d.set_get_returns_none(0)
 119         self.assertRaises(db.DBNotFoundError, d.get, 'bad key')
 120         d.set_get_returns_none(1)
 121
 122         d.put('new key', 'new data')
 123         assert d.get('new key') == 'new data'
 124         assert d['new key'] == 'new data'
 125
 126
 127
 128     def test02_cursors(self):
 129         if verbose:
 130             print '\n', '-=' * 30
 131             print "Running %s.test02_cursors..." % self.__class__.__name__
 132
 133         self.populateDB(self.d)
 134         d = self.d
 135
 136         count = 0
 137         c = d.cursor()
 138         rec = c.first()
 139         while rec is not None:
 140             count = count + 1
 141             if verbose:
 142                 print rec
 143             key, value = rec
 144             self.checkrec(key, value)
 145             rec = c.next()
 146         del c
 147
 148         assert count == len(d)
 149
 150         count = 0
 151         c = d.cursor()
 152         rec = c.last()
 153         while rec is not None:
 154             count = count + 1
 155             if verbose:
 156                 print rec
 157             key, value = rec
 158             self.checkrec(key, value)
 159             rec = c.prev()
 160
 161         assert count == len(d)
 162
 163         c.set('SS')
 164         key, value = c.current()
 165         self.checkrec(key, value)
 166         del c
 167
 168
 169
 170     def checkrec(self, key, value):
 171         x = key[1]
 172         if key[0] == 'S':
 173             assert type(value) == StringType
 174             assert value == 10 * x
 175
 176         elif key[0] == 'I':
 177             assert type(value) == IntType
 178             assert value == ord(x)
 179
 180         elif key[0] == 'L':
 181             assert type(value) == ListType
 182             assert value == [x] * 10
 183
 184         elif key[0] == 'O':
 185             assert type(value) == InstanceType
 186             assert value.S == 10 * x
 187             assert value.I == ord(x)
 188             assert value.L == [x] * 10
 189
 190         else:
 191             raise AssertionError, 'Unknown key type, fix the test'
 192
 193 #----------------------------------------------------------------------
 194
 195 class BasicShelveTestCase(DBShelveTestCase):
 196     def do_open(self):
 197         self.d = dbshelve.DBShelf()
 198         self.d.open(self.filename, self.dbtype, self.dbflags)
 199
 200     def do_close(self):
 201         self.d.close()
 202
 203
 204 class BTreeShelveTestCase(BasicShelveTestCase):
 205     dbtype = db.DB_BTREE
 206     dbflags = db.DB_CREATE
 207
 208
 209 class HashShelveTestCase(BasicShelveTestCase):
 210     dbtype = db.DB_HASH
 211     dbflags = db.DB_CREATE
 212
 213
 214 class ThreadBTreeShelveTestCase(BasicShelveTestCase):
 215     dbtype = db.DB_BTREE
 216     dbflags = db.DB_CREATE | db.DB_THREAD
 217
 218
 219 class ThreadHashShelveTestCase(BasicShelveTestCase):
 220     dbtype = db.DB_HASH
 221     dbflags = db.DB_CREATE | db.DB_THREAD
 222
 223
 224 #----------------------------------------------------------------------
 225
 226 class BasicEnvShelveTestCase(DBShelveTestCase):
 227     def do_open(self):
 228         self.homeDir = homeDir = os.path.join(
 229             os.path.dirname(sys.argv[0]), 'db_home')
 230         try: os.mkdir(homeDir)
 231         except os.error: pass
 232         self.env = db.DBEnv()
 233         self.env.open(homeDir, self.envflags | db.DB_INIT_MPOOL | db.DB_CREATE)
 234
 235         self.filename = os.path.split(self.filename)[1]
 236         self.d = dbshelve.DBShelf(self.env)
 237         self.d.open(self.filename, self.dbtype, self.dbflags)
 238
 239
 240     def do_close(self):
 241         self.d.close()
 242         self.env.close()
 243
 244
 245     def tearDown(self):
 246         self.do_close()
 247         import glob
 248         files = glob.glob(os.path.join(self.homeDir, '*'))
 249         for file in files:
 250             os.remove(file)
 251
 252
 253
 254 class EnvBTreeShelveTestCase(BasicEnvShelveTestCase):
 255     envflags = 0
 256     dbtype = db.DB_BTREE
 257     dbflags = db.DB_CREATE
 258
 259
 260 class EnvHashShelveTestCase(BasicEnvShelveTestCase):
 261     envflags = 0
 262     dbtype = db.DB_HASH
 263     dbflags = db.DB_CREATE
 264
 265
 266 class EnvThreadBTreeShelveTestCase(BasicEnvShelveTestCase):
 267     envflags = db.DB_THREAD
 268     dbtype = db.DB_BTREE
 269     dbflags = db.DB_CREATE | db.DB_THREAD
 270
 271
 272 class EnvThreadHashShelveTestCase(BasicEnvShelveTestCase):
 273     envflags = db.DB_THREAD
 274     dbtype = db.DB_HASH
 275     dbflags = db.DB_CREATE | db.DB_THREAD
 276
 277
 278 #----------------------------------------------------------------------
 279 # TODO:  Add test cases for a DBShelf in a RECNO DB.
 280
 281
 282 #----------------------------------------------------------------------
 283
 284 def test_suite():
 285     suite = unittest.TestSuite()
 286
 287     suite.addTest(unittest.makeSuite(DBShelveTestCase))
 288     suite.addTest(unittest.makeSuite(BTreeShelveTestCase))
 289     suite.addTest(unittest.makeSuite(HashShelveTestCase))
 290     suite.addTest(unittest.makeSuite(ThreadBTreeShelveTestCase))
 291     suite.addTest(unittest.makeSuite(ThreadHashShelveTestCase))
 292     suite.addTest(unittest.makeSuite(EnvBTreeShelveTestCase))
 293     suite.addTest(unittest.makeSuite(EnvHashShelveTestCase))
 294     suite.addTest(unittest.makeSuite(EnvThreadBTreeShelveTestCase))
 295     suite.addTest(unittest.makeSuite(EnvThreadHashShelveTestCase))
 296
 297     return suite
 298
 299
 300 if __name__ == '__main__':
 301     unittest.main(defaultTest='test_suite')