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