[Python-checkins] bpo-31479: Always reset the signal alarm in tests (GH-3588) (GH-7312)
Victor Stinner
webhook-mailer at python.org
Fri Jun 1 09:23:13 EDT 2018
https://github.com/python/cpython/commit/95681c7a7ddd436ba7d6c10d1202c33dd6bd648b
commit: 95681c7a7ddd436ba7d6c10d1202c33dd6bd648b
branch: 3.6
author: Victor Stinner <vstinner at redhat.com>
committer: GitHub <noreply at github.com>
date: 2018-06-01T15:23:10+02:00
summary:
bpo-31479: Always reset the signal alarm in tests (GH-3588) (GH-7312)
* 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
(cherry picked from commit 9abee722d448c1c00c7d4e11ce242ec7b13e5c49)
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 d3ae170983bc..877be517c0ea 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 3f3b3902ea2d..22f79d7b9575 100644
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -3956,6 +3956,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)
@@ -3984,6 +3985,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)
@@ -4052,6 +4054,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 45ebea51a050..cd934bf652bd 100644
--- a/Lib/test/test_pty.py
+++ b/Lib/test/test_pty.py
@@ -69,14 +69,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 14ce91f3768c..23dfd06c94c9 100644
--- a/Lib/test/test_selectors.py
+++ b/Lib/test/test_selectors.py
@@ -372,17 +372,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")
@@ -394,17 +396,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 8b0441a8e52d..2d2ec04fc091 100644
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -3697,7 +3697,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
@@ -3734,9 +3733,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)
@@ -3792,10 +3794,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)
@@ -4525,8 +4530,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 9d927423756e..7d4d8c4106f8 100644
--- a/Lib/test/test_threadsignals.py
+++ b/Lib/test/test_threadsignals.py
@@ -54,9 +54,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'],
@@ -96,6 +98,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,
@@ -127,6 +130,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