ctdb-scripts: Don't shutdown CTDB when memory monitoring fails
[Samba.git] / lib / tevent / bindings.py
blob55aafbb64a60686989b7918f2e597ffeffcb05a8
1 #!/usr/bin/python
3 # Python integration for tevent - tests
5 # Copyright (C) Jelmer Vernooij 2010
7 # ** NOTE! The following LGPL license applies to the tevent
8 # ** library. This does NOT imply that all of Samba is released
9 # ** under the LGPL
11 # This library is free software; you can redistribute it and/or
12 # modify it under the terms of the GNU Lesser General Public
13 # License as published by the Free Software Foundation; either
14 # version 3 of the License, or (at your option) any later version.
16 # This library is distributed in the hope that it will be useful,
17 # but WITHOUT ANY WARRANTY; without even the implied warranty of
18 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
19 # Lesser General Public License for more details.
21 # You should have received a copy of the GNU Lesser General Public
22 # License along with this library; if not, see <http://www.gnu.org/licenses/>.
24 import signal
25 from unittest import TestCase, TestProgram
26 import gc
28 import _tevent
30 class BackendListTests(TestCase):
32 def test_backend_list(self):
33 self.assertTrue(isinstance(_tevent.backend_list(), list))
36 class CreateContextTests(TestCase):
38 def test_by_name(self):
39 ctx = _tevent.Context(_tevent.backend_list()[0])
40 self.assertTrue(ctx is not None)
42 def test_no_name(self):
43 ctx = _tevent.Context()
44 self.assertTrue(ctx is not None)
47 class ContextTests(TestCase):
49 def setUp(self):
50 super(ContextTests, self).setUp()
51 self.ctx = _tevent.Context()
53 def test_signal_support(self):
54 self.assertTrue(type(self.ctx.signal_support) is bool)
56 def test_reinitialise(self):
57 self.ctx.reinitialise()
59 def test_loop_wait(self):
60 self.ctx.loop_wait()
62 def test_add_signal(self):
63 sig = self.ctx.add_signal(signal.SIGINT, 0, lambda callback: None)
64 self.assertTrue(isinstance(sig, _tevent.Signal))
66 def test_timer(self):
67 """Test a timer is can be scheduled"""
68 collecting_list = []
69 # time "0" has already passed, callback will be scheduled immediately
70 timer = self.ctx.add_timer(0, lambda t: collecting_list.append(True))
71 self.assertTrue(timer.active)
72 self.assertEqual(collecting_list, [])
73 self.ctx.loop_once()
74 self.assertFalse(timer.active)
75 self.assertEqual(collecting_list, [True])
77 def test_timer_deallocate_timer(self):
78 """Test timer is scheduled even if reference to it isn't held"""
79 collecting_list = []
80 def callback(t):
81 collecting_list.append(True)
82 timer = self.ctx.add_timer(0, lambda t: collecting_list.append(True))
83 gc.collect()
84 self.assertEqual(collecting_list, [])
85 self.ctx.loop_once()
86 self.assertEqual(collecting_list, [True])
88 def test_timer_deallocate_context(self):
89 """Test timer is unscheduled when context is freed"""
90 collecting_list = []
91 def callback(t):
92 collecting_list.append(True)
93 timer = self.ctx.add_timer(0, lambda t: collecting_list.append(True))
94 self.assertTrue(timer.active)
95 del self.ctx
96 gc.collect()
97 self.assertEqual(collecting_list, [])
98 self.assertFalse(timer.active)
100 def test_timer_offset(self):
101 """Test scheduling timer with an offset"""
102 collecting_list = []
103 self.ctx.add_timer_offset(0.2, lambda t: collecting_list.append(2))
104 self.ctx.add_timer_offset(0.1, lambda t: collecting_list.append(1))
105 self.assertEqual(collecting_list, [])
106 self.ctx.loop_once()
107 self.assertEqual(collecting_list, [1])
108 self.ctx.loop_once()
109 self.assertEqual(collecting_list, [1, 2])
111 if __name__ == '__main__':
112 TestProgram()