Lib/bsddb/test/test_db.py

   1 import unittest
   2 import os, glob
   3
   4 from test_all import db, test_support, get_new_environment_path, \
   5         get_new_database_path
   6
   7 #----------------------------------------------------------------------
   8
   9 class DB(unittest.TestCase):
  10     import sys
  11     if sys.version_info < (2, 4) :
  12         def assertTrue(self, expr, msg=None):
  13             self.failUnless(expr,msg=msg)
  14
  15     def setUp(self):
  16         self.path = get_new_database_path()
  17         self.db = db.DB()
  18
  19     def tearDown(self):
  20         self.db.close()
  21         del self.db
  22         test_support.rmtree(self.path)
  23
  24 class DB_general(DB) :
  25     if db.version() >= (4, 2) :
  26         def test_bt_minkey(self) :
  27             for i in [17, 108, 1030] :
  28                 self.db.set_bt_minkey(i)
  29                 self.assertEqual(i, self.db.get_bt_minkey())
  30
  31         def test_lorder(self) :
  32             self.db.set_lorder(1234)
  33             self.assertEqual(1234, self.db.get_lorder())
  34             self.db.set_lorder(4321)
  35             self.assertEqual(4321, self.db.get_lorder())
  36             self.assertRaises(db.DBInvalidArgError, self.db.set_lorder, 9182)
  37
  38     if db.version() >= (4, 6) :
  39         def test_priority(self) :
  40             flags = [db.DB_PRIORITY_VERY_LOW, db.DB_PRIORITY_LOW,
  41                     db.DB_PRIORITY_DEFAULT, db.DB_PRIORITY_HIGH,
  42                     db.DB_PRIORITY_VERY_HIGH]
  43             for flag in flags :
  44                 self.db.set_priority(flag)
  45                 self.assertEqual(flag, self.db.get_priority())
  46
  47 class DB_hash(DB) :
  48     if db.version() >= (4, 2) :
  49         def test_h_ffactor(self) :
  50             for ffactor in [4, 16, 256] :
  51                 self.db.set_h_ffactor(ffactor)
  52                 self.assertEqual(ffactor, self.db.get_h_ffactor())
  53
  54         def test_h_nelem(self) :
  55             for nelem in [1, 2, 4] :
  56                 nelem = nelem*1024*1024  # Millions
  57                 self.db.set_h_nelem(nelem)
  58                 self.assertEqual(nelem, self.db.get_h_nelem())
  59
  60         def test_pagesize(self) :
  61             for i in xrange(9, 17) :  # From 512 to 65536
  62                 i = 1<<i
  63                 self.db.set_pagesize(i)
  64                 self.assertEqual(i, self.db.get_pagesize())
  65
  66             # The valid values goes from 512 to 65536
  67             # Test 131072 bytes...
  68             self.assertRaises(db.DBInvalidArgError, self.db.set_pagesize, 1<<17)
  69             # Test 256 bytes...
  70             self.assertRaises(db.DBInvalidArgError, self.db.set_pagesize, 1<<8)
  71
  72 class DB_txn(DB) :
  73     def setUp(self) :
  74         self.homeDir = get_new_environment_path()
  75         self.env = db.DBEnv()
  76         self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL |
  77                 db.DB_INIT_LOG | db.DB_INIT_TXN)
  78         self.db = db.DB(self.env)
  79
  80     def tearDown(self) :
  81         self.db.close()
  82         del self.db
  83         self.env.close()
  84         del self.env
  85         test_support.rmtree(self.homeDir)
  86
  87     if db.version() >= (4, 2) :
  88         def test_flags(self) :
  89             self.db.set_flags(db.DB_CHKSUM)
  90             self.assertEqual(db.DB_CHKSUM, self.db.get_flags())
  91             self.db.set_flags(db.DB_TXN_NOT_DURABLE)
  92             self.assertEqual(db.DB_TXN_NOT_DURABLE | db.DB_CHKSUM,
  93                     self.db.get_flags())
  94
  95 class DB_recno(DB) :
  96     if db.version() >= (4, 2) :
  97         def test_re_pad(self) :
  98             for i in [' ', '*'] :  # Check chars
  99                 self.db.set_re_pad(i)
 100                 self.assertEqual(ord(i), self.db.get_re_pad())
 101             for i in [97, 65] :  # Check integers
 102                 self.db.set_re_pad(i)
 103                 self.assertEqual(i, self.db.get_re_pad())
 104
 105         def test_re_delim(self) :
 106             for i in [' ', '*'] :  # Check chars
 107                 self.db.set_re_delim(i)
 108                 self.assertEqual(ord(i), self.db.get_re_delim())
 109             for i in [97, 65] :  # Check integers
 110                 self.db.set_re_delim(i)
 111                 self.assertEqual(i, self.db.get_re_delim())
 112
 113         def test_re_source(self) :
 114             for i in ["test", "test2", "test3"] :
 115                 self.db.set_re_source(i)
 116                 self.assertEqual(i, self.db.get_re_source())
 117
 118 class DB_queue(DB) :
 119     if db.version() >= (4, 2) :
 120         def test_re_len(self) :
 121             for i in [33, 65, 300, 2000] :
 122                 self.db.set_re_len(i)
 123                 self.assertEqual(i, self.db.get_re_len())
 124
 125         def test_q_extentsize(self) :
 126             for i in [1, 60, 100] :
 127                 self.db.set_q_extentsize(i)
 128                 self.assertEqual(i, self.db.get_q_extentsize())
 129
 130 def test_suite():
 131     suite = unittest.TestSuite()
 132
 133     suite.addTest(unittest.makeSuite(DB_general))
 134     suite.addTest(unittest.makeSuite(DB_txn))
 135     suite.addTest(unittest.makeSuite(DB_hash))
 136     suite.addTest(unittest.makeSuite(DB_recno))
 137     suite.addTest(unittest.makeSuite(DB_queue))
 138
 139     return suite
 140
 141 if __name__ == '__main__':
 142     unittest.main(defaultTest='test_suite')