mirror of https://github.com/python/cpython
bpo-37069: tests use catch_unraisable_exception() (GH-13762)
Modify test_coroutines, test_cprofile, test_generators, test_raise, test_ssl and test_yield_from to use support.catch_unraisable_exception() rather than support.captured_stderr(). test_thread: remove test_save_exception_state_on_error() which is now updated. test_unraisable_exception() checks that sys.unraisablehook() is called to handle _thread.start_new_thread() exception. test_cprofile now rely on unittest for test discovery: replace support.run_unittest() with unittest.main().
This commit is contained in:
parent
13136e83a6
commit
0025350294
|
@ -2032,11 +2032,17 @@ class CoroutineTest(unittest.TestCase):
|
||||||
def test_fatal_coro_warning(self):
|
def test_fatal_coro_warning(self):
|
||||||
# Issue 27811
|
# Issue 27811
|
||||||
async def func(): pass
|
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")
|
warnings.filterwarnings("error")
|
||||||
func()
|
coro = func()
|
||||||
|
# only store repr() to avoid keeping the coroutine alive
|
||||||
|
coro_repr = repr(coro)
|
||||||
|
coro = None
|
||||||
support.gc_collect()
|
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):
|
def test_for_assign_raising_stop_async_iteration(self):
|
||||||
class BadTarget:
|
class BadTarget:
|
||||||
|
|
|
@ -1,13 +1,13 @@
|
||||||
"""Test suite for the cProfile module."""
|
"""Test suite for the cProfile module."""
|
||||||
|
|
||||||
import sys
|
import sys
|
||||||
from test.support import run_unittest, TESTFN, unlink
|
|
||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
# rip off all interesting stuff from test_profile
|
# rip off all interesting stuff from test_profile
|
||||||
import cProfile
|
import cProfile
|
||||||
from test.test_profile import ProfileTest, regenerate_expected_output
|
from test.test_profile import ProfileTest, regenerate_expected_output
|
||||||
from test.support.script_helper import assert_python_failure, assert_python_ok
|
from test.support.script_helper import assert_python_failure, assert_python_ok
|
||||||
|
from test import support
|
||||||
|
|
||||||
|
|
||||||
class CProfileTest(ProfileTest):
|
class CProfileTest(ProfileTest):
|
||||||
|
@ -18,24 +18,18 @@ class CProfileTest(ProfileTest):
|
||||||
def get_expected_output(self):
|
def get_expected_output(self):
|
||||||
return _ProfileOutput
|
return _ProfileOutput
|
||||||
|
|
||||||
# Issue 3895.
|
|
||||||
def test_bad_counter_during_dealloc(self):
|
def test_bad_counter_during_dealloc(self):
|
||||||
|
# bpo-3895
|
||||||
import _lsprof
|
import _lsprof
|
||||||
# Must use a file as StringIO doesn't trigger the bug.
|
|
||||||
orig_stderr = sys.stderr
|
with support.catch_unraisable_exception() as cm:
|
||||||
try:
|
|
||||||
with open(TESTFN, 'w') as file:
|
|
||||||
sys.stderr = file
|
|
||||||
try:
|
|
||||||
obj = _lsprof.Profiler(lambda: int)
|
obj = _lsprof.Profiler(lambda: int)
|
||||||
obj.enable()
|
obj.enable()
|
||||||
obj = _lsprof.Profiler(1)
|
obj = _lsprof.Profiler(1)
|
||||||
obj.disable()
|
obj.disable()
|
||||||
obj.clear()
|
obj.clear()
|
||||||
finally:
|
|
||||||
sys.stderr = orig_stderr
|
self.assertEqual(cm.unraisable.exc_type, TypeError)
|
||||||
finally:
|
|
||||||
unlink(TESTFN)
|
|
||||||
|
|
||||||
def test_profile_enable_disable(self):
|
def test_profile_enable_disable(self):
|
||||||
prof = self.profilerclass()
|
prof = self.profilerclass()
|
||||||
|
@ -70,12 +64,10 @@ class TestCommandLine(unittest.TestCase):
|
||||||
self.assertGreater(rc, 0)
|
self.assertGreater(rc, 0)
|
||||||
self.assertIn(b"option -s: invalid choice: 'demo'", err)
|
self.assertIn(b"option -s: invalid choice: 'demo'", err)
|
||||||
|
|
||||||
def test_main():
|
|
||||||
run_unittest(CProfileTest, TestCommandLine)
|
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
if '-r' not in sys.argv:
|
if '-r' not in sys.argv:
|
||||||
test_main()
|
unittest.main()
|
||||||
else:
|
else:
|
||||||
regenerate_expected_output(__file__, CProfileTest)
|
regenerate_expected_output(__file__, CProfileTest)
|
||||||
|
|
||||||
|
|
|
@ -2051,15 +2051,17 @@ RuntimeError: generator ignored GeneratorExit
|
||||||
|
|
||||||
Our ill-behaved code should be invoked during GC:
|
Our ill-behaved code should be invoked during GC:
|
||||||
|
|
||||||
>>> import sys, io
|
>>> with support.catch_unraisable_exception() as cm:
|
||||||
>>> old, sys.stderr = sys.stderr, io.StringIO()
|
... g = f()
|
||||||
>>> g = f()
|
... next(g)
|
||||||
>>> next(g)
|
... del g
|
||||||
>>> del g
|
...
|
||||||
>>> "RuntimeError: generator ignored GeneratorExit" in sys.stderr.getvalue()
|
... cm.unraisable.exc_type == RuntimeError
|
||||||
|
... "generator ignored GeneratorExit" in str(cm.unraisable.exc_value)
|
||||||
|
... cm.unraisable.exc_traceback is not None
|
||||||
|
True
|
||||||
|
True
|
||||||
True
|
True
|
||||||
>>> sys.stderr = old
|
|
||||||
|
|
||||||
|
|
||||||
And errors thrown during closing should propagate:
|
And errors thrown during closing should propagate:
|
||||||
|
|
||||||
|
|
|
@ -459,9 +459,12 @@ class TestContext(unittest.TestCase):
|
||||||
self.assertNotEqual(e.__context__, None)
|
self.assertNotEqual(e.__context__, None)
|
||||||
self.assertIsInstance(e.__context__, AttributeError)
|
self.assertIsInstance(e.__context__, AttributeError)
|
||||||
|
|
||||||
with support.captured_output("stderr"):
|
with support.catch_unraisable_exception() as cm:
|
||||||
f()
|
f()
|
||||||
|
|
||||||
|
self.assertEqual(ZeroDivisionError, cm.unraisable.exc_type)
|
||||||
|
|
||||||
|
|
||||||
class TestRemovedFunctionality(unittest.TestCase):
|
class TestRemovedFunctionality(unittest.TestCase):
|
||||||
def test_tuples(self):
|
def test_tuples(self):
|
||||||
try:
|
try:
|
||||||
|
|
|
@ -4051,13 +4051,15 @@ class ThreadedTests(unittest.TestCase):
|
||||||
1/0
|
1/0
|
||||||
server_context.set_servername_callback(cb_raising)
|
server_context.set_servername_callback(cb_raising)
|
||||||
|
|
||||||
with self.assertRaises(ssl.SSLError) as cm, \
|
with support.catch_unraisable_exception() as catch:
|
||||||
support.captured_stderr() as stderr:
|
with self.assertRaises(ssl.SSLError) as cm:
|
||||||
stats = server_params_test(client_context, server_context,
|
stats = server_params_test(client_context, server_context,
|
||||||
chatty=False,
|
chatty=False,
|
||||||
sni_name='supermessage')
|
sni_name='supermessage')
|
||||||
self.assertEqual(cm.exception.reason, 'SSLV3_ALERT_HANDSHAKE_FAILURE')
|
|
||||||
self.assertIn("ZeroDivisionError", stderr.getvalue())
|
self.assertEqual(cm.exception.reason,
|
||||||
|
'SSLV3_ALERT_HANDSHAKE_FAILURE')
|
||||||
|
self.assertEqual(catch.unraisable.exc_type, ZeroDivisionError)
|
||||||
|
|
||||||
@needs_sni
|
@needs_sni
|
||||||
def test_sni_callback_wrong_return_type(self):
|
def test_sni_callback_wrong_return_type(self):
|
||||||
|
@ -4069,13 +4071,15 @@ class ThreadedTests(unittest.TestCase):
|
||||||
return "foo"
|
return "foo"
|
||||||
server_context.set_servername_callback(cb_wrong_return_type)
|
server_context.set_servername_callback(cb_wrong_return_type)
|
||||||
|
|
||||||
with self.assertRaises(ssl.SSLError) as cm, \
|
with support.catch_unraisable_exception() as catch:
|
||||||
support.captured_stderr() as stderr:
|
with self.assertRaises(ssl.SSLError) as cm:
|
||||||
stats = server_params_test(client_context, server_context,
|
stats = server_params_test(client_context, server_context,
|
||||||
chatty=False,
|
chatty=False,
|
||||||
sni_name='supermessage')
|
sni_name='supermessage')
|
||||||
|
|
||||||
|
|
||||||
self.assertEqual(cm.exception.reason, 'TLSV1_ALERT_INTERNAL_ERROR')
|
self.assertEqual(cm.exception.reason, 'TLSV1_ALERT_INTERNAL_ERROR')
|
||||||
self.assertIn("TypeError", stderr.getvalue())
|
self.assertEqual(catch.unraisable.exc_type, TypeError)
|
||||||
|
|
||||||
def test_shared_ciphers(self):
|
def test_shared_ciphers(self):
|
||||||
client_context, server_context, hostname = testing_context()
|
client_context, server_context, hostname = testing_context()
|
||||||
|
|
|
@ -133,27 +133,6 @@ class ThreadRunningTests(BasicThreadTest):
|
||||||
time.sleep(POLL_SLEEP)
|
time.sleep(POLL_SLEEP)
|
||||||
self.assertEqual(thread._count(), orig)
|
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 test_unraisable_exception(self):
|
||||||
def task():
|
def task():
|
||||||
started.release()
|
started.release()
|
||||||
|
|
|
@ -11,6 +11,7 @@ import unittest
|
||||||
import inspect
|
import inspect
|
||||||
|
|
||||||
from test.support import captured_stderr, disable_gc, gc_collect
|
from test.support import captured_stderr, disable_gc, gc_collect
|
||||||
|
from test import support
|
||||||
|
|
||||||
class TestPEP380Operation(unittest.TestCase):
|
class TestPEP380Operation(unittest.TestCase):
|
||||||
"""
|
"""
|
||||||
|
@ -562,11 +563,12 @@ class TestPEP380Operation(unittest.TestCase):
|
||||||
self.assertEqual(next(gi), 1)
|
self.assertEqual(next(gi), 1)
|
||||||
gi.throw(AttributeError)
|
gi.throw(AttributeError)
|
||||||
|
|
||||||
with captured_stderr() as output:
|
with support.catch_unraisable_exception() as cm:
|
||||||
gi = g()
|
gi = g()
|
||||||
self.assertEqual(next(gi), 1)
|
self.assertEqual(next(gi), 1)
|
||||||
gi.close()
|
gi.close()
|
||||||
self.assertIn('ZeroDivisionError', output.getvalue())
|
|
||||||
|
self.assertEqual(ZeroDivisionError, cm.unraisable.exc_type)
|
||||||
|
|
||||||
def test_exception_in_initial_next_call(self):
|
def test_exception_in_initial_next_call(self):
|
||||||
"""
|
"""
|
||||||
|
|
|
@ -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`.
|
Loading…
Reference in New Issue