summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_selectors.py
diff options
context:
space:
mode:
authorVictor Stinner <victor.stinner@gmail.com>2015-03-31 10:08:09 (GMT)
committerVictor Stinner <victor.stinner@gmail.com>2015-03-31 10:08:09 (GMT)
commitb31017331994abcc6af3ce2febfc33593c3d7fec (patch)
treeab3097d35ec2ba5c001a81781f8726e3603ea73c /Lib/test/test_selectors.py
parent45ca48b03d08cdc9959e0577bc84b845931112f6 (diff)
downloadcpython-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.py37
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: