summaryrefslogtreecommitdiffstats
path: root/Lib
diff options
context:
space:
mode:
Diffstat (limited to 'Lib')
-rw-r--r--Lib/test/signalinterproctester.py11
-rw-r--r--Lib/test/test_io.py3
-rw-r--r--Lib/test/test_pty.py9
-rw-r--r--Lib/test/test_selectors.py40
-rw-r--r--Lib/test/test_socket.py23
-rw-r--r--Lib/test/test_threadsignals.py10
6 files changed, 56 insertions, 40 deletions
diff --git a/Lib/test/signalinterproctester.py b/Lib/test/signalinterproctester.py
index d3ae170..877be51 100644
--- a/Lib/test/signalinterproctester.py
+++ b/Lib/test/signalinterproctester.py
@@ -74,10 +74,13 @@ class InterProcessSignalTests(unittest.TestCase):
# 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 d4685dd..ce4ed1b8 100644
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -3940,6 +3940,7 @@ class SignalsTest(unittest.TestCase):
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 @@ class SignalsTest(unittest.TestCase):
# - 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 @@ class SignalsTest(unittest.TestCase):
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 b6e2ed5..f283e19 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 f2594a6..79ac25c 100644
--- a/Lib/test/test_selectors.py
+++ b/Lib/test/test_selectors.py
@@ -399,17 +399,19 @@ class BaseSelectorTestCase(unittest.TestCase):
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 @@ class BaseSelectorTestCase(unittest.TestCase):
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 01502c8..84ec429 100644
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -3854,7 +3854,6 @@ class InterruptedTimeoutBase(unittest.TestCase):
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 @@ class InterruptedRecvTimeoutTest(InterruptedTimeoutBase, UDPTestBase):
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 @@ class InterruptedSendTimeoutTest(InterruptedTimeoutBase,
# 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 @@ class TCPTimeoutTest(SocketTCPTest):
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 67a1c58..1ad6c63 100644
--- a/Lib/test/test_threadsignals.py
+++ b/Lib/test/test_threadsignals.py
@@ -56,9 +56,11 @@ class ThreadSignals(unittest.TestCase):
# 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 @@ class ThreadSignals(unittest.TestCase):
# 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 @@ class ThreadSignals(unittest.TestCase):
# 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):
generated by cgit v0.12 at 2024-12-11 12:51:53 (GMT)