1 | #!/usr/bin/env python
|
---|
2 | # Some simple tests for the Python bindings for TDB
|
---|
3 | # Note that this tests the interface of the Python bindings
|
---|
4 | # It does not test tdb itself.
|
---|
5 | #
|
---|
6 | # Copyright (C) 2007-2008 Jelmer Vernooij <jelmer@samba.org>
|
---|
7 | # Published under the GNU LGPLv3 or later
|
---|
8 |
|
---|
9 | import tdb
|
---|
10 | from unittest import TestCase
|
---|
11 | import os, tempfile
|
---|
12 |
|
---|
13 |
|
---|
14 | class OpenTdbTests(TestCase):
|
---|
15 |
|
---|
16 | def test_nonexistant_read(self):
|
---|
17 | self.assertRaises(IOError, tdb.Tdb, "/some/nonexistant/file", 0,
|
---|
18 | tdb.DEFAULT, os.O_RDWR)
|
---|
19 |
|
---|
20 | class CloseTdbTests(TestCase):
|
---|
21 |
|
---|
22 | def test_double_close(self):
|
---|
23 | self.tdb = tdb.Tdb(tempfile.mkstemp()[1], 0, tdb.DEFAULT,
|
---|
24 | os.O_CREAT|os.O_RDWR)
|
---|
25 | self.assertNotEqual(None, self.tdb)
|
---|
26 |
|
---|
27 | # ensure that double close does not crash python
|
---|
28 | self.tdb.close()
|
---|
29 | self.tdb.close()
|
---|
30 |
|
---|
31 |
|
---|
32 | class InternalTdbTests(TestCase):
|
---|
33 |
|
---|
34 | def test_repr(self):
|
---|
35 | self.tdb = tdb.Tdb()
|
---|
36 |
|
---|
37 | # repr used to crash on internal db
|
---|
38 | self.assertEquals(repr(self.tdb), "Tdb(<internal>)")
|
---|
39 |
|
---|
40 |
|
---|
41 | class SimpleTdbTests(TestCase):
|
---|
42 |
|
---|
43 | def setUp(self):
|
---|
44 | super(SimpleTdbTests, self).setUp()
|
---|
45 | self.tdb = tdb.Tdb(tempfile.mkstemp()[1], 0, tdb.DEFAULT,
|
---|
46 | os.O_CREAT|os.O_RDWR)
|
---|
47 | self.assertNotEqual(None, self.tdb)
|
---|
48 |
|
---|
49 | def tearDown(self):
|
---|
50 | del self.tdb
|
---|
51 |
|
---|
52 | def test_repr(self):
|
---|
53 | self.assertTrue(repr(self.tdb).startswith("Tdb('"))
|
---|
54 |
|
---|
55 | def test_lockall(self):
|
---|
56 | self.tdb.lock_all()
|
---|
57 |
|
---|
58 | def test_max_dead(self):
|
---|
59 | self.tdb.max_dead = 20
|
---|
60 |
|
---|
61 | def test_unlockall(self):
|
---|
62 | self.tdb.lock_all()
|
---|
63 | self.tdb.unlock_all()
|
---|
64 |
|
---|
65 | def test_lockall_read(self):
|
---|
66 | self.tdb.read_lock_all()
|
---|
67 | self.tdb.read_unlock_all()
|
---|
68 |
|
---|
69 | def test_reopen(self):
|
---|
70 | self.tdb.reopen()
|
---|
71 |
|
---|
72 | def test_store(self):
|
---|
73 | self.tdb.store("bar", "bla")
|
---|
74 | self.assertEquals("bla", self.tdb.get("bar"))
|
---|
75 |
|
---|
76 | def test_getitem(self):
|
---|
77 | self.tdb["bar"] = "foo"
|
---|
78 | self.tdb.reopen()
|
---|
79 | self.assertEquals("foo", self.tdb["bar"])
|
---|
80 |
|
---|
81 | def test_delete(self):
|
---|
82 | self.tdb["bar"] = "foo"
|
---|
83 | del self.tdb["bar"]
|
---|
84 | self.assertRaises(KeyError, lambda: self.tdb["bar"])
|
---|
85 |
|
---|
86 | def test_contains(self):
|
---|
87 | self.tdb["bla"] = "bloe"
|
---|
88 | self.assertTrue("bla" in self.tdb)
|
---|
89 |
|
---|
90 | def test_keyerror(self):
|
---|
91 | self.assertRaises(KeyError, lambda: self.tdb["bla"])
|
---|
92 |
|
---|
93 | def test_hash_size(self):
|
---|
94 | self.tdb.hash_size
|
---|
95 |
|
---|
96 | def test_map_size(self):
|
---|
97 | self.tdb.map_size
|
---|
98 |
|
---|
99 | def test_freelist_size(self):
|
---|
100 | self.tdb.freelist_size
|
---|
101 |
|
---|
102 | def test_name(self):
|
---|
103 | self.tdb.filename
|
---|
104 |
|
---|
105 | def test_iterator(self):
|
---|
106 | self.tdb["bla"] = "1"
|
---|
107 | self.tdb["brainslug"] = "2"
|
---|
108 | self.assertEquals(["bla", "brainslug"], list(self.tdb))
|
---|
109 |
|
---|
110 | def test_transaction_cancel(self):
|
---|
111 | self.tdb["bloe"] = "2"
|
---|
112 | self.tdb.transaction_start()
|
---|
113 | self.tdb["bloe"] = "1"
|
---|
114 | self.tdb.transaction_cancel()
|
---|
115 | self.assertEquals("2", self.tdb["bloe"])
|
---|
116 |
|
---|
117 | def test_transaction_commit(self):
|
---|
118 | self.tdb["bloe"] = "2"
|
---|
119 | self.tdb.transaction_start()
|
---|
120 | self.tdb["bloe"] = "1"
|
---|
121 | self.tdb.transaction_commit()
|
---|
122 | self.assertEquals("1", self.tdb["bloe"])
|
---|
123 |
|
---|
124 | def test_transaction_prepare_commit(self):
|
---|
125 | self.tdb["bloe"] = "2"
|
---|
126 | self.tdb.transaction_start()
|
---|
127 | self.tdb["bloe"] = "1"
|
---|
128 | self.tdb.transaction_prepare_commit()
|
---|
129 | self.tdb.transaction_commit()
|
---|
130 | self.assertEquals("1", self.tdb["bloe"])
|
---|
131 |
|
---|
132 | def test_iterkeys(self):
|
---|
133 | self.tdb["bloe"] = "2"
|
---|
134 | self.tdb["bla"] = "25"
|
---|
135 | i = self.tdb.iterkeys()
|
---|
136 | self.assertEquals(set(["bloe", "bla"]), set([i.next(), i.next()]))
|
---|
137 |
|
---|
138 | def test_clear(self):
|
---|
139 | self.tdb["bloe"] = "2"
|
---|
140 | self.tdb["bla"] = "25"
|
---|
141 | self.assertEquals(2, len(list(self.tdb)))
|
---|
142 | self.tdb.clear()
|
---|
143 | self.assertEquals(0, len(list(self.tdb)))
|
---|
144 |
|
---|
145 | def test_repack(self):
|
---|
146 | self.tdb["foo"] = "abc"
|
---|
147 | self.tdb["bar"] = "def"
|
---|
148 | del self.tdb["foo"]
|
---|
149 | self.tdb.repack()
|
---|
150 |
|
---|
151 | def test_seqnum(self):
|
---|
152 | self.tdb.enable_seqnum()
|
---|
153 | seq1 = self.tdb.seqnum
|
---|
154 | self.tdb.increment_seqnum_nonblock()
|
---|
155 | seq2 = self.tdb.seqnum
|
---|
156 | self.assertEquals(seq2-seq1, 1)
|
---|
157 |
|
---|
158 | def test_len(self):
|
---|
159 | self.assertEquals(0, len(list(self.tdb)))
|
---|
160 | self.tdb["entry"] = "value"
|
---|
161 | self.assertEquals(1, len(list(self.tdb)))
|
---|
162 |
|
---|
163 | def test_add_flags(self):
|
---|
164 | self.tdb.add_flags(tdb.NOMMAP)
|
---|
165 | self.tdb.remove_flags(tdb.NOMMAP)
|
---|
166 |
|
---|
167 |
|
---|
168 | class VersionTests(TestCase):
|
---|
169 |
|
---|
170 | def test_present(self):
|
---|
171 | self.assertTrue(isinstance(tdb.__version__, str))
|
---|
172 |
|
---|
173 |
|
---|
174 | if __name__ == '__main__':
|
---|
175 | import unittest
|
---|
176 | unittest.TestProgram()
|
---|