librpc: Remove unused RPC_NDR_KEYSVC
[Samba.git] / lib / tevent / bindings.py
blobf5e14997c3df2590ea86a7f6a435bcd67bfd53d4
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
31 class BackendListTests(TestCase):
33 def test_backend_list(self):
34 self.assertTrue(isinstance(_tevent.backend_list(), list))
37 class CreateContextTests(TestCase):
39 def test_by_name(self):
40 ctx = _tevent.Context(_tevent.backend_list()[0])
41 self.assertTrue(ctx is not None)
43 def test_no_name(self):
44 ctx = _tevent.Context()
45 self.assertTrue(ctx is not None)
48 class ContextTests(TestCase):
50 def setUp(self):
51 super(ContextTests, self).setUp()
52 self.ctx = _tevent.Context()
54 def test_signal_support(self):
55 self.assertTrue(type(self.ctx.signal_support) is bool)
57 def test_reinitialise(self):
58 self.ctx.reinitialise()
60 def test_loop_wait(self):
61 self.ctx.loop_wait()
63 def test_add_signal(self):
64 sig = self.ctx.add_signal(signal.SIGINT, 0, lambda callback: None)
65 self.assertTrue(isinstance(sig, _tevent.Signal))
67 def test_timer(self):
68 """Test a timer is can be scheduled"""
69 collecting_list = []
70 # time "0" has already passed, callback will be scheduled immediately
71 timer = self.ctx.add_timer(0, lambda t: collecting_list.append(True))
72 self.assertTrue(timer.active)
73 self.assertEqual(collecting_list, [])
74 self.ctx.loop_once()
75 self.assertFalse(timer.active)
76 self.assertEqual(collecting_list, [True])
78 def test_timer_deallocate_timer(self):
79 """Test timer is scheduled even if reference to it isn't held"""
80 collecting_list = []
82 def callback(t):
83 collecting_list.append(True)
84 timer = self.ctx.add_timer(0, lambda t: collecting_list.append(True))
85 gc.collect()
86 self.assertEqual(collecting_list, [])
87 self.ctx.loop_once()
88 self.assertEqual(collecting_list, [True])
90 def test_timer_deallocate_context(self):
91 """Test timer is unscheduled when context is freed"""
92 collecting_list = []
94 def callback(t):
95 collecting_list.append(True)
96 timer = self.ctx.add_timer(0, lambda t: collecting_list.append(True))
97 self.assertTrue(timer.active)
98 del self.ctx
99 gc.collect()
100 self.assertEqual(collecting_list, [])
101 self.assertFalse(timer.active)
103 def test_timer_offset(self):
104 """Test scheduling timer with an offset"""
105 collecting_list = []
106 self.ctx.add_timer_offset(0.2, lambda t: collecting_list.append(2))
107 self.ctx.add_timer_offset(0.1, lambda t: collecting_list.append(1))
108 self.assertEqual(collecting_list, [])
109 self.ctx.loop_once()
110 self.assertEqual(collecting_list, [1])
111 self.ctx.loop_once()
112 self.assertEqual(collecting_list, [1, 2])
115 if __name__ == '__main__':
116 TestProgram()