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