Changeset 391 for python/trunk/Lib/test/test_signal.py

Timestamp:

Mar 19, 2014, 11:31:01 PM (11 years ago)

Author:

dmik

Message:

python: Merge vendor 2.7.6 to trunk.

Location:

python/trunk

Files:

• . (modified) (1 prop)
• Lib/test/test_signal.py (modified) (17 diffs)

python/trunk/Lib/test/test_signal.py

@@ -1 +1 @@
 import unittest
 from test import test_support
-from contextlib import closing, nested
+from contextlib import closing
 import gc
 import pickle
@@ -10 +10 @@
 import sys, os, time, errno
 
-if sys.platform[:3] in ('win', 'os2') or sys.platform == 'riscos':
-    raise test_support.TestSkipped("Can't test signal on %s" % \
-                                   sys.platform)
+if sys.platform in ('os2', 'riscos'):
+    raise unittest.SkipTest("Can't test signal on %s" % sys.platform)
 
 
@@ -38 +37 @@
 
 
+@unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
 class InterProcessSignalTests(unittest.TestCase):
     MAX_DURATION = 20   # Entire test should last at most 20 sec.
@@ -110 +110 @@
             self.wait(child)
             time.sleep(1)  # Give the signal time to be delivered.
-            self.fail('HandlerBCalled exception not thrown')
+            self.fail('HandlerBCalled exception not raised')
         except HandlerBCalled:
             self.assertTrue(self.b_called)
@@ -140 +140 @@
                       " didn't arrive after another second.")
 
+    # Issue 3864. Unknown if this affects earlier versions of freebsd also.
+    @unittest.skipIf(sys.platform=='freebsd6',
+        'inter process signals not reliable (do not mix well with threading) '
+        'on freebsd6')
     def test_main(self):
         # This function spawns a child process to insulate the main
@@ -145 +149 @@
         # communicates with that child process over a pipe and
         # re-raises information about any exceptions the child
-        # throws. The real work happens in self.run_test().
+        # raises. The real work happens in self.run_test().
         os_done_r, os_done_w = os.pipe()
-        with nested(closing(os.fdopen(os_done_r)),
-                    closing(os.fdopen(os_done_w, 'w'))) as (done_r, done_w):
+        with closing(os.fdopen(os_done_r)) as done_r, \
+             closing(os.fdopen(os_done_w, 'w')) as done_w:
             child = os.fork()
             if child == 0:
@@ -183 +187 @@
 
 
+@unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
 class BasicSignalTests(unittest.TestCase):
     def trivial_signal_handler(self, *args):
@@ -199 +204 @@
     def test_getsignal(self):
         hup = signal.signal(signal.SIGHUP, self.trivial_signal_handler)
-        self.assertEquals(signal.getsignal(signal.SIGHUP),
-                          self.trivial_signal_handler)
+        self.assertEqual(signal.getsignal(signal.SIGHUP),
+                         self.trivial_signal_handler)
         signal.signal(signal.SIGHUP, hup)
-        self.assertEquals(signal.getsignal(signal.SIGHUP), hup)
-
-
+        self.assertEqual(signal.getsignal(signal.SIGHUP), hup)
+
+
+@unittest.skipUnless(sys.platform == "win32", "Windows specific")
+class WindowsSignalTests(unittest.TestCase):
+    def test_issue9324(self):
+        # Updated for issue #10003, adding SIGBREAK
+        handler = lambda x, y: None
+        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
+                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
+                    signal.SIGTERM):
+            # Set and then reset a handler for signals that work on windows
+            signal.signal(sig, signal.signal(sig, handler))
+
+        with self.assertRaises(ValueError):
+            signal.signal(-1, handler)
+
+        with self.assertRaises(ValueError):
+            signal.signal(7, handler)
+
+
+class WakeupFDTests(unittest.TestCase):
+
+    def test_invalid_fd(self):
+        fd = test_support.make_bad_fd()
+        self.assertRaises(ValueError, signal.set_wakeup_fd, fd)
+
+
+@unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
 class WakeupSignalTests(unittest.TestCase):
     TIMEOUT_FULL = 10
@@ -218 +249 @@
         time.sleep(self.TIMEOUT_FULL)
         mid_time = time.time()
-        self.assert_(mid_time - before_time < self.TIMEOUT_HALF)
+        self.assertTrue(mid_time - before_time < self.TIMEOUT_HALF)
         select.select([self.read], [], [], self.TIMEOUT_FULL)
         after_time = time.time()
-        self.assert_(after_time - mid_time < self.TIMEOUT_HALF)
+        self.assertTrue(after_time - mid_time < self.TIMEOUT_HALF)
 
     def test_wakeup_fd_during(self):
@@ -232 +263 @@
             [self.read], [], [], self.TIMEOUT_FULL)
         after_time = time.time()
-        self.assert_(after_time - before_time < self.TIMEOUT_HALF)
+        self.assertTrue(after_time - before_time < self.TIMEOUT_HALF)
 
     def setUp(self):
@@ -250 +281 @@
         signal.signal(signal.SIGALRM, self.alrm)
 
+@unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
 class SiginterruptTest(unittest.TestCase):
-    signum = signal.SIGUSR1
-    def readpipe_interrupted(self, cb):
+
+    def setUp(self):
+        """Install a no-op signal handler that can be set to allow
+        interrupts or not, and arrange for the original signal handler to be
+        re-installed when the test is finished.
+        """
+        self.signum = signal.SIGUSR1
+        oldhandler = signal.signal(self.signum, lambda x,y: None)
+        self.addCleanup(signal.signal, self.signum, oldhandler)
+
+    def readpipe_interrupted(self):
+        """Perform a read during which a signal will arrive.  Return True if the
+        read is interrupted by the signal and raises an exception.  Return False
+        if it returns normally.
+        """
+        # Create a pipe that can be used for the read.  Also clean it up
+        # when the test is over, since nothing else will (but see below for
+        # the write end).
         r, w = os.pipe()
+        self.addCleanup(os.close, r)
+
+        # Create another process which can send a signal to this one to try
+        # to interrupt the read.
         ppid = os.getpid()
         pid = os.fork()
 
-        oldhandler = signal.signal(self.signum, lambda x,y: None)
-        cb()
-        if pid==0:
-            # child code: sleep, kill, sleep. and then exit,
-            # which closes the pipe from which the parent process reads
+        if pid == 0:
+            # Child code: sleep to give the parent enough time to enter the
+            # read() call (there's a race here, but it's really tricky to
+            # eliminate it); then signal the parent process.  Also, sleep
+            # again to make it likely that the signal is delivered to the
+            # parent process before the child exits.  If the child exits
+            # first, the write end of the pipe will be closed and the test
+            # is invalid.
             try:
                 time.sleep(0.2)
@@ -267 +322 @@
                 time.sleep(0.2)
             finally:
+                # No matter what, just exit as fast as possible now.
                 exit_subprocess()
-
-        try:
+        else:
+            # Parent code.
+            # Make sure the child is eventually reaped, else it'll be a
+            # zombie for the rest of the test suite run.
+            self.addCleanup(os.waitpid, pid, 0)
+
+            # Close the write end of the pipe.  The child has a copy, so
+            # it's not really closed until the child exits.  We need it to
+            # close when the child exits so that in the non-interrupt case
+            # the read eventually completes, otherwise we could just close
+            # it *after* the test.
             os.close(w)
 
+            # Try the read and report whether it is interrupted or not to
+            # the caller.
             try:
-                d=os.read(r, 1)
+                d = os.read(r, 1)
                 return False
             except OSError, err:
@@ -279 +346 @@
                     raise
                 return True
-        finally:
-            signal.signal(self.signum, oldhandler)
-            os.waitpid(pid, 0)
 
     def test_without_siginterrupt(self):
-        i=self.readpipe_interrupted(lambda: None)
-        self.assertEquals(i, True)
+        """If a signal handler is installed and siginterrupt is not called
+        at all, when that signal arrives, it interrupts a syscall that's in
+        progress.
+        """
+        i = self.readpipe_interrupted()
+        self.assertTrue(i)
+        # Arrival of the signal shouldn't have changed anything.
+        i = self.readpipe_interrupted()
+        self.assertTrue(i)
 
     def test_siginterrupt_on(self):
-        i=self.readpipe_interrupted(lambda: signal.siginterrupt(self.signum, 1))
-        self.assertEquals(i, True)
+        """If a signal handler is installed and siginterrupt is called with
+        a true value for the second argument, when that signal arrives, it
+        interrupts a syscall that's in progress.
+        """
+        signal.siginterrupt(self.signum, 1)
+        i = self.readpipe_interrupted()
+        self.assertTrue(i)
+        # Arrival of the signal shouldn't have changed anything.
+        i = self.readpipe_interrupted()
+        self.assertTrue(i)
 
     def test_siginterrupt_off(self):
-        i=self.readpipe_interrupted(lambda: signal.siginterrupt(self.signum, 0))
-        self.assertEquals(i, False)
-
+        """If a signal handler is installed and siginterrupt is called with
+        a false value for the second argument, when that signal arrives, it
+        does not interrupt a syscall that's in progress.
+        """
+        signal.siginterrupt(self.signum, 0)
+        i = self.readpipe_interrupted()
+        self.assertFalse(i)
+        # Arrival of the signal shouldn't have changed anything.
+        i = self.readpipe_interrupted()
+        self.assertFalse(i)
+
+
+@unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
 class ItimerTest(unittest.TestCase):
     def setUp(self):
@@ -356 +445 @@
         self.assertEqual(self.hndl_called, True)
 
+    # Issue 3864. Unknown if this affects earlier versions of freebsd also.
+    @unittest.skipIf(sys.platform in ('freebsd6', 'netbsd5'),
+        'itimer not reliable (does not mix well with threading) on some BSDs.')
     def test_itimer_virtual(self):
         self.itimer = signal.ITIMER_VIRTUAL
@@ -362 +454 @@
 
         start_time = time.time()
-        while time.time() - start_time < 5.0:
+        while time.time() - start_time < 60.0:
             # use up some virtual time by doing real work
             _ = pow(12345, 67890, 10000019)
             if signal.getitimer(self.itimer) == (0.0, 0.0):
                 break # sig_vtalrm handler stopped this itimer
-        else:
-            self.fail('timeout waiting for sig_vtalrm signal; '
-                      'signal.getitimer(self.itimer) gives: %s' %
-                       (signal.getitimer(self.itimer),))
+        else: # Issue 8424
+            self.skipTest("timeout: likely cause: machine too slow or load too "
+                          "high")
 
         # virtual itimer should be (0.0, 0.0) now
-        self.assertEquals(signal.getitimer(self.itimer), (0.0, 0.0))
+        self.assertEqual(signal.getitimer(self.itimer), (0.0, 0.0))
         # and the handler should have been called
-        self.assertEquals(self.hndl_called, True)
-
+        self.assertEqual(self.hndl_called, True)
+
+    # Issue 3864. Unknown if this affects earlier versions of freebsd also.
+    @unittest.skipIf(sys.platform=='freebsd6',
+        'itimer not reliable (does not mix well with threading) on freebsd6')
     def test_itimer_prof(self):
         self.itimer = signal.ITIMER_PROF
@@ -383 +477 @@
 
         start_time = time.time()
-        while time.time() - start_time < 5.0:
+        while time.time() - start_time < 60.0:
             # do some work
             _ = pow(12345, 67890, 10000019)
             if signal.getitimer(self.itimer) == (0.0, 0.0):
                 break # sig_prof handler stopped this itimer
-        else:
-            self.fail('timeout waiting for sig_prof signal')
+        else: # Issue 8424
+            self.skipTest("timeout: likely cause: machine too slow or load too "
+                          "high")
 
         # profiling itimer should be (0.0, 0.0) now
-        self.assertEquals(signal.getitimer(self.itimer), (0.0, 0.0))
+        self.assertEqual(signal.getitimer(self.itimer), (0.0, 0.0))
         # and the handler should have been called
         self.assertEqual(self.hndl_called, True)
@@ -398 +493 @@
 def test_main():
     test_support.run_unittest(BasicSignalTests, InterProcessSignalTests,
-        WakeupSignalTests, SiginterruptTest, ItimerTest)
+                              WakeupFDTests, WakeupSignalTests,
+                              SiginterruptTest, ItimerTest,
+                              WindowsSignalTests)