[Python-checkins] bpo-31479: Always reset the signal alarm in tests (#3588)
Victor Stinner
webhook-mailer at python.org
Tue Sep 19 12:36:57 EDT 2017
https://github.com/python/cpython/commit/9abee722d448c1c00c7d4e11ce242ec7b13e5c49
commit: 9abee722d448c1c00c7d4e11ce242ec7b13e5c49
branch: master
author: Victor Stinner <victor.stinner at gmail.com>
committer: GitHub <noreply at github.com>
date: 2017-09-19T09:36:54-07:00
summary:
bpo-31479: Always reset the signal alarm in tests (#3588)
* bpo-31479: Always reset the signal alarm in tests
Use "try: ... finally: signal.signal(0)" pattern to make sure that
tests don't "leak" a pending fatal signal alarm.
* Move two more alarm() calls into the try block
Fix also typo: replace signal.signal(0) with signal.alarm(0)
* Move another signal.alarm() into the try block
files:
M Lib/test/signalinterproctester.py
M Lib/test/test_io.py
M Lib/test/test_pty.py
M Lib/test/test_selectors.py
M Lib/test/test_socket.py
M Lib/test/test_threadsignals.py
diff --git a/Lib/test/signalinterproctester.py b/Lib/test/signalinterproctester.py
index d3ae170983b..877be517c0e 100644
--- a/Lib/test/signalinterproctester.py
+++ b/Lib/test/signalinterproctester.py
@@ -74,10 +74,13 @@ def test_interprocess_signal(self):
# Nothing should happen: SIGUSR2 is ignored
child.wait()
- signal.alarm(1)
- self.wait_signal(None, 'SIGALRM', KeyboardInterrupt)
- self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 1,
- 'SIGALRM': 0})
+ try:
+ signal.alarm(1)
+ self.wait_signal(None, 'SIGALRM', KeyboardInterrupt)
+ self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 1,
+ 'SIGALRM': 0})
+ finally:
+ signal.alarm(0)
if __name__ == "__main__":
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py
index d4685dd22e1..ce4ed1b8f6a 100644
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -3940,6 +3940,7 @@ def on_alarm(*args):
if isinstance(exc, RuntimeError):
self.assertTrue(str(exc).startswith("reentrant call"), str(exc))
finally:
+ signal.alarm(0)
wio.close()
os.close(r)
@@ -3968,6 +3969,7 @@ def alarm_handler(sig, frame):
# - third raw read() returns b"bar"
self.assertEqual(decode(rio.read(6)), "foobar")
finally:
+ signal.alarm(0)
rio.close()
os.close(w)
os.close(r)
@@ -4035,6 +4037,7 @@ def alarm2(sig, frame):
self.assertIsNone(error)
self.assertEqual(N, sum(len(x) for x in read_results))
finally:
+ signal.alarm(0)
write_finished = True
os.close(w)
os.close(r)
diff --git a/Lib/test/test_pty.py b/Lib/test/test_pty.py
index b6e2ed51f61..f283e1930b7 100644
--- a/Lib/test/test_pty.py
+++ b/Lib/test/test_pty.py
@@ -51,14 +51,11 @@ class PtyTest(unittest.TestCase):
def setUp(self):
# isatty() and close() can hang on some platforms. Set an alarm
# before running the test to make sure we don't hang forever.
- self.old_alarm = signal.signal(signal.SIGALRM, self.handle_sig)
+ old_alarm = signal.signal(signal.SIGALRM, self.handle_sig)
+ self.addCleanup(signal.signal, signal.SIGALRM, old_alarm)
+ self.addCleanup(signal.alarm, 0)
signal.alarm(10)
- def tearDown(self):
- # remove alarm, restore old alarm handler
- signal.alarm(0)
- signal.signal(signal.SIGALRM, self.old_alarm)
-
def handle_sig(self, sig, frame):
self.fail("isatty hung")
diff --git a/Lib/test/test_selectors.py b/Lib/test/test_selectors.py
index f2594a6b121..79ac25c12d4 100644
--- a/Lib/test/test_selectors.py
+++ b/Lib/test/test_selectors.py
@@ -399,17 +399,19 @@ def handler(*args):
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)
+ try:
+ 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)
+ 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)
+ finally:
+ signal.alarm(0)
@unittest.skipUnless(hasattr(signal, "alarm"),
"signal.alarm() required for this test")
@@ -421,17 +423,19 @@ def test_select_interrupt_noraise(self):
orig_alrm_handler = signal.signal(signal.SIGALRM, lambda *args: None)
self.addCleanup(signal.signal, signal.SIGALRM, orig_alrm_handler)
- self.addCleanup(signal.alarm, 0)
- signal.alarm(1)
+ try:
+ signal.alarm(1)
- s.register(rd, selectors.EVENT_READ)
- t = time()
- # 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)
+ s.register(rd, selectors.EVENT_READ)
+ t = time()
+ # 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)
+ finally:
+ signal.alarm(0)
class ScalableSelectorMixIn:
diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py
index 01502c805cc..84ec4291335 100644
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -3854,7 +3854,6 @@ def setUp(self):
orig_alrm_handler = signal.signal(signal.SIGALRM,
lambda signum, frame: 1 / 0)
self.addCleanup(signal.signal, signal.SIGALRM, orig_alrm_handler)
- self.addCleanup(self.setAlarm, 0)
# Timeout for socket operations
timeout = 4.0
@@ -3891,9 +3890,12 @@ def setUp(self):
def checkInterruptedRecv(self, func, *args, **kwargs):
# Check that func(*args, **kwargs) raises
# errno of EINTR when interrupted by a signal.
- self.setAlarm(self.alarm_time)
- with self.assertRaises(ZeroDivisionError) as cm:
- func(*args, **kwargs)
+ try:
+ self.setAlarm(self.alarm_time)
+ with self.assertRaises(ZeroDivisionError) as cm:
+ func(*args, **kwargs)
+ finally:
+ self.setAlarm(0)
def testInterruptedRecvTimeout(self):
self.checkInterruptedRecv(self.serv.recv, 1024)
@@ -3948,10 +3950,13 @@ def checkInterruptedSend(self, func, *args, **kwargs):
# Check that func(*args, **kwargs), run in a loop, raises
# OSError with an errno of EINTR when interrupted by a
# signal.
- with self.assertRaises(ZeroDivisionError) as cm:
- while True:
- self.setAlarm(self.alarm_time)
- func(*args, **kwargs)
+ try:
+ with self.assertRaises(ZeroDivisionError) as cm:
+ while True:
+ self.setAlarm(self.alarm_time)
+ func(*args, **kwargs)
+ finally:
+ self.setAlarm(0)
# Issue #12958: The following tests have problems on OS X prior to 10.7
@support.requires_mac_ver(10, 7)
@@ -4675,8 +4680,8 @@ def alarm_handler(signal, frame):
raise Alarm
old_alarm = signal.signal(signal.SIGALRM, alarm_handler)
try:
- signal.alarm(2) # POSIX allows alarm to be up to 1 second early
try:
+ signal.alarm(2) # POSIX allows alarm to be up to 1 second early
foo = self.serv.accept()
except socket.timeout:
self.fail("caught timeout instead of Alarm")
diff --git a/Lib/test/test_threadsignals.py b/Lib/test/test_threadsignals.py
index 67a1c581d79..1ad6c63fea2 100644
--- a/Lib/test/test_threadsignals.py
+++ b/Lib/test/test_threadsignals.py
@@ -56,9 +56,11 @@ def test_signals(self):
# wait for it return.
if signal_blackboard[signal.SIGUSR1]['tripped'] == 0 \
or signal_blackboard[signal.SIGUSR2]['tripped'] == 0:
- signal.alarm(1)
- signal.pause()
- signal.alarm(0)
+ try:
+ signal.alarm(1)
+ signal.pause()
+ finally:
+ signal.alarm(0)
self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped'], 1)
self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped_by'],
@@ -98,6 +100,7 @@ def test_lock_acquire_interruption(self):
# after timeout return of lock.acquire() (which can fool assertRaises).
self.assertLess(dt, 3.0)
finally:
+ signal.alarm(0)
signal.signal(signal.SIGALRM, oldalrm)
@unittest.skipIf(USING_PTHREAD_COND,
@@ -131,6 +134,7 @@ def other_thread():
# See rationale above in test_lock_acquire_interruption
self.assertLess(dt, 3.0)
finally:
+ signal.alarm(0)
signal.signal(signal.SIGALRM, oldalrm)
def acquire_retries_on_intr(self, lock):
More information about the Python-checkins
mailing list