diff options
author | Victor Stinner <victor.stinner@gmail.com> | 2015-03-31 10:08:09 (GMT) |
---|---|---|
committer | Victor Stinner <victor.stinner@gmail.com> | 2015-03-31 10:08:09 (GMT) |
commit | b31017331994abcc6af3ce2febfc33593c3d7fec (patch) | |
tree | ab3097d35ec2ba5c001a81781f8726e3603ea73c /Lib/test/test_selectors.py | |
parent | 45ca48b03d08cdc9959e0577bc84b845931112f6 (diff) | |
download | cpython-b31017331994abcc6af3ce2febfc33593c3d7fec.zip cpython-b31017331994abcc6af3ce2febfc33593c3d7fec.tar.gz cpython-b31017331994abcc6af3ce2febfc33593c3d7fec.tar.bz2 |
Issue #23485: Enhance and update selectors doc and test_selectors
Selector.select() is now retried with the recomputed timeout when interrupted
by a signal.
Write an unit test with a signal handler raising an exception, and a unit with
a signal handler which does not raise an exception (it does nothing).
Diffstat (limited to 'Lib/test/test_selectors.py')
-rw-r--r-- | Lib/test/test_selectors.py | 37 |
1 files changed, 34 insertions, 3 deletions
diff --git a/Lib/test/test_selectors.py b/Lib/test/test_selectors.py index 9521481..454c17b 100644 --- a/Lib/test/test_selectors.py +++ b/Lib/test/test_selectors.py @@ -357,7 +357,35 @@ class BaseSelectorTestCase(unittest.TestCase): @unittest.skipUnless(hasattr(signal, "alarm"), "signal.alarm() required for this test") - def test_select_interrupt(self): + def test_select_interrupt_exc(self): + s = self.SELECTOR() + self.addCleanup(s.close) + + rd, wr = self.make_socketpair() + + class InterruptSelect(Exception): + pass + + def handler(*args): + raise InterruptSelect + + orig_alrm_handler = signal.signal(signal.SIGALRM, handler) + self.addCleanup(signal.signal, signal.SIGALRM, orig_alrm_handler) + self.addCleanup(signal.alarm, 0) + + signal.alarm(1) + + s.register(rd, selectors.EVENT_READ) + t = time() + # select() is interrupted by a signal which raises an exception + with self.assertRaises(InterruptSelect): + s.select(30) + # select() was interrupted before the timeout of 30 seconds + self.assertLess(time() - t, 5.0) + + @unittest.skipUnless(hasattr(signal, "alarm"), + "signal.alarm() required for this test") + def test_select_interrupt_noraise(self): s = self.SELECTOR() self.addCleanup(s.close) @@ -371,8 +399,11 @@ class BaseSelectorTestCase(unittest.TestCase): s.register(rd, selectors.EVENT_READ) t = time() - self.assertFalse(s.select(2)) - self.assertLess(time() - t, 2.5) + # select() is interrupted by a signal, but the signal handler doesn't + # raise an exception, so select() should by retries with a recomputed + # timeout + self.assertFalse(s.select(1.5)) + self.assertGreaterEqual(time() - t, 1.0) class ScalableSelectorMixIn: |