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