python · vstinner · Jun 3, 2019 · Jun 3, 2019
diff --git a/Lib/test/test_coroutines.py b/Lib/test/test_coroutines.py
@@ -2032,11 +2032,17 @@ async def func(): pass
    def test_fatal_coro_warning(self):
        # Issue 27811
        async def func(): pass
-        with warnings.catch_warnings(), support.captured_stderr() as stderr:
+        with warnings.catch_warnings(), \
+             support.catch_unraisable_exception() as cm:
            warnings.filterwarnings("error")
-            func()
+            coro = func()
+            # only store repr() to avoid keeping the coroutine alive
+            coro_repr = repr(coro)
+            coro = None
            support.gc_collect()
-        self.assertIn("was never awaited", stderr.getvalue())
+
+            self.assertIn("was never awaited", str(cm.unraisable.exc_value))
+            self.assertEqual(repr(cm.unraisable.object), coro_repr)

    def test_for_assign_raising_stop_async_iteration(self):
        class BadTarget:

diff --git a/Lib/test/test_cprofile.py b/Lib/test/test_cprofile.py
@@ -1,13 +1,13 @@
 """Test suite for the cProfile module."""

 import sys
-from test.support import run_unittest, TESTFN, unlink
 import unittest

 # rip off all interesting stuff from test_profile
 import cProfile
 from test.test_profile import ProfileTest, regenerate_expected_output
 from test.support.script_helper import assert_python_failure, assert_python_ok
+from test import support


 class CProfileTest(ProfileTest):
@@ -18,24 +18,18 @@ class CProfileTest(ProfileTest):
    def get_expected_output(self):
        return _ProfileOutput

-    # Issue 3895.
    def test_bad_counter_during_dealloc(self):
+        # bpo-3895
        import _lsprof
-        # Must use a file as StringIO doesn't trigger the bug.
-        orig_stderr = sys.stderr
-        try:
-            with open(TESTFN, 'w') as file:
-                sys.stderr = file
-                try:
-                    obj = _lsprof.Profiler(lambda: int)
-                    obj.enable()
-                    obj = _lsprof.Profiler(1)
-                    obj.disable()
-                    obj.clear()
-                finally:
-                    sys.stderr = orig_stderr
-        finally:
-            unlink(TESTFN)
+
+        with support.catch_unraisable_exception() as cm:
+            obj = _lsprof.Profiler(lambda: int)
+            obj.enable()
+            obj = _lsprof.Profiler(1)
+            obj.disable()
+            obj.clear()
+
+            self.assertEqual(cm.unraisable.exc_type, TypeError)

    def test_profile_enable_disable(self):
        prof = self.profilerclass()
@@ -70,12 +64,10 @@ def test_sort(self):
        self.assertGreater(rc, 0)
        self.assertIn(b"option -s: invalid choice: 'demo'", err)

-def test_main():
-    run_unittest(CProfileTest, TestCommandLine)

 def main():
    if '-r' not in sys.argv:
-        test_main()
+        unittest.main()
    else:
        regenerate_expected_output(__file__, CProfileTest)


diff --git a/Lib/test/test_generators.py b/Lib/test/test_generators.py
@@ -2051,15 +2051,17 @@ def printsolution(self, x):

 Our ill-behaved code should be invoked during GC:

->>> import sys, io
->>> old, sys.stderr = sys.stderr, io.StringIO()
->>> g = f()
->>> next(g)
->>> del g
->>> "RuntimeError: generator ignored GeneratorExit" in sys.stderr.getvalue()
+>>> with support.catch_unraisable_exception() as cm:
+...     g = f()
+...     next(g)
+...     del g
+...
+...     cm.unraisable.exc_type == RuntimeError
+...     "generator ignored GeneratorExit" in str(cm.unraisable.exc_value)
+...     cm.unraisable.exc_traceback is not None
+True
+True
 True
->>> sys.stderr = old
-

 And errors thrown during closing should propagate:


diff --git a/Lib/test/test_raise.py b/Lib/test/test_raise.py
@@ -459,9 +459,12 @@ def f():
                self.assertNotEqual(e.__context__, None)
                self.assertIsInstance(e.__context__, AttributeError)

-        with support.captured_output("stderr"):
+        with support.catch_unraisable_exception() as cm:
            f()

+            self.assertEqual(ZeroDivisionError, cm.unraisable.exc_type)
+
+
 class TestRemovedFunctionality(unittest.TestCase):
    def test_tuples(self):
        try:

diff --git a/Lib/test/test_ssl.py b/Lib/test/test_ssl.py
@@ -4051,13 +4051,15 @@ def cb_raising(ssl_sock, server_name, initial_context):
            1/0
        server_context.set_servername_callback(cb_raising)

-        with self.assertRaises(ssl.SSLError) as cm, \
-             support.captured_stderr() as stderr:
-            stats = server_params_test(client_context, server_context,
-                                       chatty=False,
-                                       sni_name='supermessage')
-        self.assertEqual(cm.exception.reason, 'SSLV3_ALERT_HANDSHAKE_FAILURE')
-        self.assertIn("ZeroDivisionError", stderr.getvalue())
+        with support.catch_unraisable_exception() as catch:
+            with self.assertRaises(ssl.SSLError) as cm:
+                stats = server_params_test(client_context, server_context,
+                                           chatty=False,
+                                           sni_name='supermessage')
+
+            self.assertEqual(cm.exception.reason,
+                             'SSLV3_ALERT_HANDSHAKE_FAILURE')
+            self.assertEqual(catch.unraisable.exc_type, ZeroDivisionError)

    @needs_sni
    def test_sni_callback_wrong_return_type(self):
@@ -4069,13 +4071,15 @@ def cb_wrong_return_type(ssl_sock, server_name, initial_context):
            return "foo"
        server_context.set_servername_callback(cb_wrong_return_type)

-        with self.assertRaises(ssl.SSLError) as cm, \
-             support.captured_stderr() as stderr:
-            stats = server_params_test(client_context, server_context,
-                                       chatty=False,
-                                       sni_name='supermessage')
-        self.assertEqual(cm.exception.reason, 'TLSV1_ALERT_INTERNAL_ERROR')
-        self.assertIn("TypeError", stderr.getvalue())
+        with support.catch_unraisable_exception() as catch:
+            with self.assertRaises(ssl.SSLError) as cm:
+                stats = server_params_test(client_context, server_context,
+                                           chatty=False,
+                                           sni_name='supermessage')
+
+
+            self.assertEqual(cm.exception.reason, 'TLSV1_ALERT_INTERNAL_ERROR')
+            self.assertEqual(catch.unraisable.exc_type, TypeError)

    def test_shared_ciphers(self):
        client_context, server_context, hostname = testing_context()

diff --git a/Lib/test/test_thread.py b/Lib/test/test_thread.py
@@ -133,27 +133,6 @@ def task():
                time.sleep(POLL_SLEEP)
            self.assertEqual(thread._count(), orig)

-    def test_save_exception_state_on_error(self):
-        # See issue #14474
-        def task():
-            started.release()
-            raise SyntaxError
-        def mywrite(self, *args):
-            try:
-                raise ValueError
-            except ValueError:
-                pass
-            real_write(self, *args)
-        started = thread.allocate_lock()
-        with support.captured_output("stderr") as stderr:
-            real_write = stderr.write
-            stderr.write = mywrite
-            started.acquire()
-            with support.wait_threads_exit():
-                thread.start_new_thread(task, ())
-                started.acquire()
-        self.assertIn("Traceback", stderr.getvalue())
-
    def test_unraisable_exception(self):
        def task():
            started.release()

diff --git a/Lib/test/test_yield_from.py b/Lib/test/test_yield_from.py
@@ -11,6 +11,7 @@
 import inspect

 from test.support import captured_stderr, disable_gc, gc_collect
+from test import support

 class TestPEP380Operation(unittest.TestCase):
    """
@@ -562,11 +563,12 @@ def g():
            self.assertEqual(next(gi), 1)
            gi.throw(AttributeError)

-        with captured_stderr() as output:
+        with support.catch_unraisable_exception() as cm:
            gi = g()
            self.assertEqual(next(gi), 1)
            gi.close()
-        self.assertIn('ZeroDivisionError', output.getvalue())
+
+            self.assertEqual(ZeroDivisionError, cm.unraisable.exc_type)

    def test_exception_in_initial_next_call(self):
        """

diff --git a/Misc/NEWS.d/next/Tests/2019-06-03-02-30-36.bpo-37069.rVtdLk.rst b/Misc/NEWS.d/next/Tests/2019-06-03-02-30-36.bpo-37069.rVtdLk.rst
@@ -0,0 +1,3 @@
+Modify test_coroutines, test_cprofile, test_generators, test_raise, test_ssl
+and test_yield_from to use :func:`test.support.catch_unraisable_exception`
+rather than :func:`test.support.captured_stderr`.