bpo-31479: Always reset the signal alarm in tests

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__":

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)

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")
 

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:

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")

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):