[Python-checkins] cpython (3.2): Issue #12469: Run "wakeup" signal tests in subprocess to run the test in a

[victor.stinner]

http://hg.python.org/cpython/rev/e07b331bf489
changeset:   71195:e07b331bf489
branch:      3.2
parent:      71189:e924e51e9447
user:        Victor Stinner <victor.stinner at haypocalc.com>
date:        Mon Jul 04 17:35:10 2011 +0200
summary:
  Issue #12469: Run "wakeup" signal tests in subprocess to run the test in a
fresh process with only one thread and to not change signal handling of the
parent process.

files:
  Lib/test/test_signal.py |  105 ++++++++++++++++++---------
  Misc/NEWS               |    4 +
  2 files changed, 72 insertions(+), 37 deletions(-)


diff --git a/Lib/test/test_signal.py b/Lib/test/test_signal.py
--- a/Lib/test/test_signal.py
+++ b/Lib/test/test_signal.py
@@ -11,7 +11,7 @@
 import unittest
 from test import support
 from contextlib import closing
-from test.script_helper import spawn_python
+from test.script_helper import assert_python_ok, spawn_python
 
 if sys.platform in ('os2', 'riscos'):
     raise unittest.SkipTest("Can't test signal on %s" % sys.platform)
@@ -233,49 +233,80 @@
 
 @unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
 class WakeupSignalTests(unittest.TestCase):
-    TIMEOUT_FULL = 10
-    TIMEOUT_HALF = 5
+    def check_wakeup(self, test_body):
+        # use a subprocess to have only one thread and to not change signal
+        # handling of the parent process
+        code = """if 1:
+        import fcntl
+        import os
+        import signal
+
+        def handler(signum, frame):
+            pass
+
+        {}
+
+        signal.signal(signal.SIGALRM, handler)
+        read, write = os.pipe()
+        flags = fcntl.fcntl(write, fcntl.F_GETFL, 0)
+        flags = flags | os.O_NONBLOCK
+        fcntl.fcntl(write, fcntl.F_SETFL, flags)
+        signal.set_wakeup_fd(write)
+
+        test()
+
+        os.close(read)
+        os.close(write)
+        """.format(test_body)
+
+        assert_python_ok('-c', code)
 
     def test_wakeup_fd_early(self):
-        import select
+        self.check_wakeup("""def test():
+            import select
+            import time
 
-        signal.alarm(1)
-        before_time = time.time()
-        # We attempt to get a signal during the sleep,
-        # before select is called
-        time.sleep(self.TIMEOUT_FULL)
-        mid_time = time.time()
-        self.assertTrue(mid_time - before_time < self.TIMEOUT_HALF)
-        select.select([self.read], [], [], self.TIMEOUT_FULL)
-        after_time = time.time()
-        self.assertTrue(after_time - mid_time < self.TIMEOUT_HALF)
+            TIMEOUT_FULL = 10
+            TIMEOUT_HALF = 5
+
+            signal.alarm(1)
+            before_time = time.time()
+            # We attempt to get a signal during the sleep,
+            # before select is called
+            time.sleep(TIMEOUT_FULL)
+            mid_time = time.time()
+            dt = mid_time - before_time
+            if dt >= TIMEOUT_HALF:
+                raise Exception("%s >= %s" % (dt, TIMEOUT_HALF))
+            select.select([read], [], [], TIMEOUT_FULL)
+            after_time = time.time()
+            dt = after_time - mid_time
+            if dt >= TIMEOUT_HALF:
+                raise Exception("%s >= %s" % (dt, TIMEOUT_HALF))
+        """)
 
     def test_wakeup_fd_during(self):
-        import select
+        self.check_wakeup("""def test():
+            import select
+            import time
 
-        signal.alarm(1)
-        before_time = time.time()
-        # We attempt to get a signal during the select call
-        self.assertRaises(select.error, select.select,
-            [self.read], [], [], self.TIMEOUT_FULL)
-        after_time = time.time()
-        self.assertTrue(after_time - before_time < self.TIMEOUT_HALF)
+            TIMEOUT_FULL = 10
+            TIMEOUT_HALF = 5
 
-    def setUp(self):
-        import fcntl
-
-        self.alrm = signal.signal(signal.SIGALRM, lambda x,y:None)
-        self.read, self.write = os.pipe()
-        flags = fcntl.fcntl(self.write, fcntl.F_GETFL, 0)
-        flags = flags | os.O_NONBLOCK
-        fcntl.fcntl(self.write, fcntl.F_SETFL, flags)
-        self.old_wakeup = signal.set_wakeup_fd(self.write)
-
-    def tearDown(self):
-        signal.set_wakeup_fd(self.old_wakeup)
-        os.close(self.read)
-        os.close(self.write)
-        signal.signal(signal.SIGALRM, self.alrm)
+            signal.alarm(1)
+            before_time = time.time()
+            # We attempt to get a signal during the select call
+            try:
+                select.select([read], [], [], TIMEOUT_FULL)
+            except select.error:
+                pass
+            else:
+                raise Exception("select.error not raised")
+            after_time = time.time()
+            dt = after_time - before_time
+            if dt >= TIMEOUT_HALF:
+                raise Exception("%s >= %s" % (dt, TIMEOUT_HALF))
+        """)
 
 @unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
 class SiginterruptTest(unittest.TestCase):
diff --git a/Misc/NEWS b/Misc/NEWS
--- a/Misc/NEWS
+++ b/Misc/NEWS
@@ -38,6 +38,10 @@
 Tests
 -----
 
+- Issue #12469: Run "wakeup" signal tests in subprocess to run the test in a
+  fresh process with only one thread and to not change signal handling of the
+  parent process.
+
 - Issue #8716: Avoid crashes caused by Aqua Tk on OSX when attempting to run
   test_tk or test_ttk_guionly under a username that is not currently logged
   in to the console windowserver (as may be the case under buildbot or ssh).

-- 
Repository URL: http://hg.python.org/cpython