1import unittest
2import os, glob
3
4from test_all import db, test_support, get_new_environment_path, \
5        get_new_database_path
6
7#----------------------------------------------------------------------
8
9class 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
19class 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
65class 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
89class 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
118class 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
140class 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
151def 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
162if __name__ == '__main__':
163    unittest.main(defaultTest='test_suite')
164