[2.7] bpo-30764: test_subprocess uses SuppressCrashReport (#2405) (#2412)

vstinner · web-flow · commit 2097b9e0ef32 · 2017-06-27T00:00:51.000+02:00
* bpo-30764: Backport support.SuppressCrashReport Backport test.support.SuppressCrashReport context-manager from master. Drop the Windows implementation since it depends on msvcrt.CrtSetReportMode() which isn't available on Python 2.7. * bpo-30764: test_subprocess uses SuppressCrashReport (#2405) bpo-30764, bpo-29335: test_child_terminated_in_stopped_state() of test_subprocess now uses support.SuppressCrashReport() to prevent the creation of a core dump on FreeBSD. (cherry picked from commit cdee3f1)
diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py
@@ -23,6 +23,7 @@
 import struct
 import sysconfig
 import types
+
 try:
     import thread
 except ImportError:
@@ -43,7 +44,8 @@
            "threading_cleanup", "reap_threads", "start_threads", "cpython_only",
            "check_impl_detail", "get_attribute", "py3k_bytes",
            "import_fresh_module", "threading_cleanup", "reap_children",
-           "strip_python_stderr", "IPV6_ENABLED", "run_with_tz"]
+           "strip_python_stderr", "IPV6_ENABLED", "run_with_tz",
+           "SuppressCrashReport"]
 
 class Error(Exception):
     """Base class for regression test exceptions."""
@@ -1842,3 +1844,78 @@ def python_is_optimized():
         if opt.startswith('-O'):
             final_opt = opt
     return final_opt not in ('', '-O0', '-Og')
+
+
+class SuppressCrashReport:
+    """Try to prevent a crash report from popping up.
+
+    On Windows, don't display the Windows Error Reporting dialog.  On UNIX,
+    disable the creation of coredump file.
+    """
+    old_value = None
+    old_modes = None
+
+    def __enter__(self):
+        """On Windows, disable Windows Error Reporting dialogs using
+        SetErrorMode.
+
+        On UNIX, try to save the previous core file size limit, then set
+        soft limit to 0.
+        """
+        if sys.platform.startswith('win'):
+            # TODO: backport the Windows implementation
+            pass
+        else:
+            try:
+                import resource
+            except ImportError:
+                resource = None
+
+            if resource is not None:
+                try:
+                    self.old_value = resource.getrlimit(resource.RLIMIT_CORE)
+                    resource.setrlimit(resource.RLIMIT_CORE,
+                                       (0, self.old_value[1]))
+                except (ValueError, OSError):
+                    pass
+
+            if sys.platform == 'darwin':
+                # Check if the 'Crash Reporter' on OSX was configured
+                # in 'Developer' mode and warn that it will get triggered
+                # when it is.
+                #
+                # This assumes that this context manager is used in tests
+                # that might trigger the next manager.
+                cmd = ['/usr/bin/defaults', 'read',
+                       'com.apple.CrashReporter', 'DialogType']
+                proc = subprocess.Popen(cmd,
+                                        stdout=subprocess.PIPE,
+                                        stderr=subprocess.PIPE)
+                with proc:
+                    stdout = proc.communicate()[0]
+                if stdout.strip() == b'developer':
+                    sys.stdout.write("this test triggers the Crash Reporter, "
+                                     "that is intentional")
+                    sys.stdout.flush()
+
+        return self
+
+    def __exit__(self, *ignore_exc):
+        """Restore Windows ErrorMode or core file behavior to initial value."""
+        if self.old_value is None:
+            return
+
+        if sys.platform.startswith('win'):
+            # TODO: backport the Windows implementation
+            pass
+        else:
+            try:
+                import resource
+            except ImportError:
+                resource = None
+
+            if resource is not None:
+                try:
+                    resource.setrlimit(resource.RLIMIT_CORE, self.old_value)
+                except (ValueError, OSError):
+                    pass
diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py
@@ -10,11 +10,14 @@
 import time
 import re
 import sysconfig
+import textwrap
 
 try:
     import ctypes
 except ImportError:
     ctypes = None
+else:
+    import ctypes.util
 
 try:
     import resource
@@ -1262,43 +1265,39 @@ def test_pipe_cloexec(self):
 
         self.assertEqual(p2.returncode, 0, "Unexpected error: " + repr(stderr))
 
-    _libc_file_extensions = {
-      'Linux': 'so.6',
-      'Darwin': 'dylib',
-    }
-    @unittest.skipIf(not ctypes, 'ctypes module required.')
-    @unittest.skipIf(platform.uname()[0] not in _libc_file_extensions,
-                     'Test requires a libc this code can load with ctypes.')
-    @unittest.skipIf(not sys.executable, 'Test requires sys.executable.')
+    @unittest.skipIf(not ctypes, 'ctypes module required')
+    @unittest.skipIf(not sys.executable, 'Test requires sys.executable')
     def test_child_terminated_in_stopped_state(self):
         """Test wait() behavior when waitpid returns WIFSTOPPED; issue29335."""
         PTRACE_TRACEME = 0  # From glibc and MacOS (PT_TRACE_ME).
-        libc_name = 'libc.' + self._libc_file_extensions[platform.uname()[0]]
+        libc_name = ctypes.util.find_library('c')
         libc = ctypes.CDLL(libc_name)
         if not hasattr(libc, 'ptrace'):
-            raise unittest.SkipTest('ptrace() required.')
-        test_ptrace = subprocess.Popen(
-            [sys.executable, '-c', """if True:
-             import ctypes
-             libc = ctypes.CDLL({libc_name!r})
-             libc.ptrace({PTRACE_TRACEME}, 0, 0)
-             """.format(libc_name=libc_name, PTRACE_TRACEME=PTRACE_TRACEME)
-            ])
-        if test_ptrace.wait() != 0:
-            raise unittest.SkipTest('ptrace() failed - unable to test.')
-        child = subprocess.Popen(
-            [sys.executable, '-c', """if True:
+            raise unittest.SkipTest('ptrace() required')
+
+        code = textwrap.dedent("""
              import ctypes
+             from test.support import SuppressCrashReport
+
              libc = ctypes.CDLL({libc_name!r})
              libc.ptrace({PTRACE_TRACEME}, 0, 0)
-             libc.printf(ctypes.c_char_p(0xdeadbeef))  # Crash the process.
-             """.format(libc_name=libc_name, PTRACE_TRACEME=PTRACE_TRACEME)
-            ])
+        """.format(libc_name=libc_name, PTRACE_TRACEME=PTRACE_TRACEME))
+
+        child = subprocess.Popen([sys.executable, '-c', code])
+        if child.wait() != 0:
+            raise unittest.SkipTest('ptrace() failed - unable to test')
+
+        code += textwrap.dedent("""
+             with SuppressCrashReport():
+                # Crash the process
+                libc.printf(ctypes.c_char_p(0xdeadbeef))  # Crash the process.
+        """)
+        child = subprocess.Popen([sys.executable, '-c', code])
         try:
             returncode = child.wait()
-        except Exception as e:
+        except:
             child.kill()  # Clean up the hung stopped process.
-            raise e
+            raise
         self.assertNotEqual(0, returncode)
         self.assertLess(returncode, 0)  # signal death, likely SIGSEGV.
 
