From 8daa83a594a2e98f39d764422bfbdbc62c9efd44 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Fri, 19 Apr 2024 19:20:00 +0200 Subject: Adding upstream version 2:4.20.0+dfsg. Signed-off-by: Daniel Baumann --- lib/tevent/bindings.py | 116 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 116 insertions(+) create mode 100644 lib/tevent/bindings.py (limited to 'lib/tevent/bindings.py') diff --git a/lib/tevent/bindings.py b/lib/tevent/bindings.py new file mode 100644 index 0000000..f5e1499 --- /dev/null +++ b/lib/tevent/bindings.py @@ -0,0 +1,116 @@ +#!/usr/bin/python +# +# Python integration for tevent - tests +# +# Copyright (C) Jelmer Vernooij 2010 +# +# ** NOTE! The following LGPL license applies to the tevent +# ** library. This does NOT imply that all of Samba is released +# ** under the LGPL +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 3 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, see . + +import signal +from unittest import TestCase, TestProgram +import gc + +import _tevent + + +class BackendListTests(TestCase): + + def test_backend_list(self): + self.assertTrue(isinstance(_tevent.backend_list(), list)) + + +class CreateContextTests(TestCase): + + def test_by_name(self): + ctx = _tevent.Context(_tevent.backend_list()[0]) + self.assertTrue(ctx is not None) + + def test_no_name(self): + ctx = _tevent.Context() + self.assertTrue(ctx is not None) + + +class ContextTests(TestCase): + + def setUp(self): + super(ContextTests, self).setUp() + self.ctx = _tevent.Context() + + def test_signal_support(self): + self.assertTrue(type(self.ctx.signal_support) is bool) + + def test_reinitialise(self): + self.ctx.reinitialise() + + def test_loop_wait(self): + self.ctx.loop_wait() + + def test_add_signal(self): + sig = self.ctx.add_signal(signal.SIGINT, 0, lambda callback: None) + self.assertTrue(isinstance(sig, _tevent.Signal)) + + def test_timer(self): + """Test a timer is can be scheduled""" + collecting_list = [] + # time "0" has already passed, callback will be scheduled immediately + timer = self.ctx.add_timer(0, lambda t: collecting_list.append(True)) + self.assertTrue(timer.active) + self.assertEqual(collecting_list, []) + self.ctx.loop_once() + self.assertFalse(timer.active) + self.assertEqual(collecting_list, [True]) + + def test_timer_deallocate_timer(self): + """Test timer is scheduled even if reference to it isn't held""" + collecting_list = [] + + def callback(t): + collecting_list.append(True) + timer = self.ctx.add_timer(0, lambda t: collecting_list.append(True)) + gc.collect() + self.assertEqual(collecting_list, []) + self.ctx.loop_once() + self.assertEqual(collecting_list, [True]) + + def test_timer_deallocate_context(self): + """Test timer is unscheduled when context is freed""" + collecting_list = [] + + def callback(t): + collecting_list.append(True) + timer = self.ctx.add_timer(0, lambda t: collecting_list.append(True)) + self.assertTrue(timer.active) + del self.ctx + gc.collect() + self.assertEqual(collecting_list, []) + self.assertFalse(timer.active) + + def test_timer_offset(self): + """Test scheduling timer with an offset""" + collecting_list = [] + self.ctx.add_timer_offset(0.2, lambda t: collecting_list.append(2)) + self.ctx.add_timer_offset(0.1, lambda t: collecting_list.append(1)) + self.assertEqual(collecting_list, []) + self.ctx.loop_once() + self.assertEqual(collecting_list, [1]) + self.ctx.loop_once() + self.assertEqual(collecting_list, [1, 2]) + + +if __name__ == '__main__': + TestProgram() -- cgit v1.2.3