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.
6 # Copyright (C) 2007-2008 Jelmer Vernooij <jelmer@samba.org>
7 # Published under the GNU LGPLv3 or later
12 from unittest import TestCase
17 class OpenTdbTests(TestCase):
19 def test_nonexistent_read(self):
20 self.assertRaises(IOError, tdb.Tdb, "/some/nonexistent/file", 0,
21 tdb.DEFAULT, os.O_RDWR)
23 class CloseTdbTests(TestCase):
25 def test_double_close(self):
26 self.tdb = tdb.Tdb(tempfile.mkstemp()[1], 0, tdb.DEFAULT,
28 self.assertNotEqual(None, self.tdb)
30 # ensure that double close does not crash python
34 # Check that further operations do not crash python
35 self.assertRaises(RuntimeError, lambda: self.tdb.transaction_start())
37 self.assertRaises(RuntimeError, lambda: self.tdb["bar"])
40 class InternalTdbTests(TestCase):
45 # repr used to crash on internal db
46 self.assertEqual(repr(self.tdb), "Tdb(<internal>)")
49 class CommonTdbTests(TestCase):
50 """Tests common to both the text & bytes interfaces"""
55 super(CommonTdbTests, self).setUp()
56 self.tdb = tdb.Tdb(tempfile.mkstemp()[1], 0, tdb.DEFAULT,
58 self.assertNotEqual(None, self.tdb)
60 self.tdb = self.tdb.text
62 def test_lockall(self):
65 def test_max_dead(self):
66 self.tdb.max_dead = 20
68 def test_unlockall(self):
72 def test_lockall_read(self):
73 self.tdb.read_lock_all()
74 self.tdb.read_unlock_all()
76 def test_reopen(self):
79 def test_hash_size(self):
82 def test_map_size(self):
85 def test_freelist_size(self):
86 self.tdb.freelist_size
91 def test_add_flags(self):
92 self.tdb.add_flags(tdb.NOMMAP)
93 self.tdb.remove_flags(tdb.NOMMAP)
96 class TextCommonTdbTests(CommonTdbTests):
101 class SimpleTdbTests(TestCase):
104 super(SimpleTdbTests, self).setUp()
105 self.tdb = tdb.Tdb(tempfile.mkstemp()[1], 0, tdb.DEFAULT,
106 os.O_CREAT|os.O_RDWR)
107 self.assertNotEqual(None, self.tdb)
110 self.assertTrue(repr(self.tdb).startswith("Tdb('"))
112 def test_store(self):
113 self.tdb.store(b"bar", b"bla")
114 self.assertEqual(b"bla", self.tdb.get(b"bar"))
116 def test_storev(self):
117 self.tdb.storev(b"bar", [b"first", b"second", b"third"])
118 self.assertEqual(b"firstsecondthird", self.tdb.get(b"bar"))
120 def test_getitem(self):
121 self.tdb[b"bar"] = b"foo"
123 self.assertEqual(b"foo", self.tdb[b"bar"])
125 def test_delete(self):
126 self.tdb[b"bar"] = b"foo"
128 self.assertRaises(KeyError, lambda: self.tdb[b"bar"])
130 def test_contains(self):
131 self.tdb[b"bla"] = b"bloe"
132 self.assertTrue(b"bla" in self.tdb)
133 self.assertFalse(b"qwertyuiop" in self.tdb)
134 if sys.version_info < (3, 0):
135 self.assertTrue(self.tdb.has_key(b"bla"))
136 self.assertFalse(self.tdb.has_key(b"qwertyuiop"))
138 def test_keyerror(self):
139 self.assertRaises(KeyError, lambda: self.tdb[b"bla"])
141 def test_iterator(self):
142 self.tdb[b"bla"] = b"1"
143 self.tdb[b"brainslug"] = b"2"
146 self.assertEqual([b"bla", b"brainslug"], l)
148 def test_transaction_cancel(self):
149 self.tdb[b"bloe"] = b"2"
150 self.tdb.transaction_start()
151 self.tdb[b"bloe"] = b"1"
152 self.tdb.transaction_cancel()
153 self.assertEqual(b"2", self.tdb[b"bloe"])
155 def test_transaction_commit(self):
156 self.tdb[b"bloe"] = b"2"
157 self.tdb.transaction_start()
158 self.tdb[b"bloe"] = b"1"
159 self.tdb.transaction_commit()
160 self.assertEqual(b"1", self.tdb[b"bloe"])
162 def test_transaction_prepare_commit(self):
163 self.tdb[b"bloe"] = b"2"
164 self.tdb.transaction_start()
165 self.tdb[b"bloe"] = b"1"
166 self.tdb.transaction_prepare_commit()
167 self.tdb.transaction_commit()
168 self.assertEqual(b"1", self.tdb[b"bloe"])
170 def test_iterkeys(self):
171 self.tdb[b"bloe"] = b"2"
172 self.tdb[b"bla"] = b"25"
173 if sys.version_info >= (3, 0):
176 i = self.tdb.iterkeys()
177 self.assertEqual(set([b"bloe", b"bla"]), set([next(i), next(i)]))
179 def test_clear(self):
180 self.tdb[b"bloe"] = b"2"
181 self.tdb[b"bla"] = b"25"
182 self.assertEqual(2, len(list(self.tdb)))
184 self.assertEqual(0, len(list(self.tdb)))
186 def test_repack(self):
187 self.tdb[b"foo"] = b"abc"
188 self.tdb[b"bar"] = b"def"
192 def test_seqnum(self):
193 self.tdb.enable_seqnum()
194 seq1 = self.tdb.seqnum
195 self.tdb.increment_seqnum_nonblock()
196 seq2 = self.tdb.seqnum
197 self.assertEqual(seq2-seq1, 1)
200 self.assertEqual(0, len(list(self.tdb)))
201 self.tdb[b"entry"] = b"value"
202 self.assertEqual(1, len(list(self.tdb)))
205 class TdbTextTests(TestCase):
208 super(TdbTextTests, self).setUp()
209 self.tdb = tdb.Tdb(tempfile.mkstemp()[1], 0, tdb.DEFAULT,
210 os.O_CREAT|os.O_RDWR)
211 self.assertNotEqual(None, self.tdb)
214 self.assertTrue(repr(self.tdb).startswith("Tdb('"))
216 def test_store(self):
217 self.tdb.text.store("bar", "bla")
218 self.assertEqual("bla", self.tdb.text.get("bar"))
220 def test_getitem(self):
221 self.tdb.text["bar"] = "foo"
223 self.assertEqual("foo", self.tdb.text["bar"])
225 def test_delete(self):
226 self.tdb.text["bar"] = "foo"
227 del self.tdb.text["bar"]
228 self.assertRaises(KeyError, lambda: self.tdb.text["bar"])
230 def test_contains(self):
231 self.tdb.text["bla"] = "bloe"
232 self.assertTrue("bla" in self.tdb.text)
233 self.assertFalse("qwertyuiop" in self.tdb.text)
234 if sys.version_info < (3, 0):
235 self.assertTrue(self.tdb.text.has_key("bla"))
236 self.assertFalse(self.tdb.text.has_key("qwertyuiop"))
238 def test_keyerror(self):
239 self.assertRaises(KeyError, lambda: self.tdb.text["bla"])
241 def test_iterator(self):
242 self.tdb.text["bla"] = "1"
243 self.tdb.text["brainslug"] = "2"
244 l = list(self.tdb.text)
246 self.assertEqual(["bla", "brainslug"], l)
248 def test_transaction_cancel(self):
249 self.tdb.text["bloe"] = "2"
250 self.tdb.transaction_start()
251 self.tdb.text["bloe"] = "1"
252 self.tdb.transaction_cancel()
253 self.assertEqual("2", self.tdb.text["bloe"])
255 def test_transaction_commit(self):
256 self.tdb.text["bloe"] = "2"
257 self.tdb.transaction_start()
258 self.tdb.text["bloe"] = "1"
259 self.tdb.transaction_commit()
260 self.assertEqual("1", self.tdb.text["bloe"])
262 def test_transaction_prepare_commit(self):
263 self.tdb.text["bloe"] = "2"
264 self.tdb.transaction_start()
265 self.tdb.text["bloe"] = "1"
266 self.tdb.transaction_prepare_commit()
267 self.tdb.transaction_commit()
268 self.assertEqual("1", self.tdb.text["bloe"])
270 def test_iterkeys(self):
271 self.tdb.text["bloe"] = "2"
272 self.tdb.text["bla"] = "25"
273 if sys.version_info >= (3, 0):
274 i = self.tdb.text.keys()
276 i = self.tdb.text.iterkeys()
277 self.assertEqual(set(["bloe", "bla"]), set([next(i), next(i)]))
279 def test_clear(self):
280 self.tdb.text["bloe"] = "2"
281 self.tdb.text["bla"] = "25"
282 self.assertEqual(2, len(list(self.tdb)))
284 self.assertEqual(0, len(list(self.tdb)))
286 def test_repack(self):
287 self.tdb.text["foo"] = "abc"
288 self.tdb.text["bar"] = "def"
289 del self.tdb.text["foo"]
293 self.assertEqual(0, len(list(self.tdb.text)))
294 self.tdb.text["entry"] = "value"
295 self.assertEqual(1, len(list(self.tdb.text)))
297 def test_text_and_binary(self):
298 text = u'\xfa\u0148\xef\xe7\xf8\xf0\xea'
299 bytestr = text.encode('utf-8')
300 self.tdb[b"entry"] = bytestr
301 self.tdb.text[u"entry2"] = text
302 self.assertEqual(self.tdb.text["entry"], text)
303 self.assertEqual(self.tdb[b"entry2"], bytestr)
304 assert self.tdb.text.raw == self.tdb
307 class VersionTests(TestCase):
309 def test_present(self):
310 self.assertTrue(isinstance(tdb.__version__, str))
313 if __name__ == '__main__':
315 unittest.TestProgram()