bpo-43843: libregrtest uses threading.excepthook (GH-25400)

test.libregrtest now marks a test as ENV_CHANGED (altered the
execution environment) if a thread raises an exception but does not
catch it. It sets a hook on threading.excepthook. Use
--fail-env-changed option to mark the test as failed.

libregrtest regrtest_unraisable_hook() explicitly flushs
sys.stdout, sys.stderr and sys.__stderr__.

1 files changed, 32 insertions, 1 deletions

diff --git a/Lib/test/test_regrtest.py b/Lib/test/test_regrtest.py
index 38321e0..054776c 100644
--- a/Lib/test/test_regrtest.py
+++ b/Lib/test/test_regrtest.py
@@ -1236,7 +1236,7 @@ class ArgsTestCase(BaseTestCase):
 
     def test_unraisable_exc(self):
         # --fail-env-changed must catch unraisable exception.
-        # The exceptioin must be displayed even if sys.stderr is redirected.
+        # The exception must be displayed even if sys.stderr is redirected.
         code = textwrap.dedent(r"""
             import unittest
             import weakref
@@ -1267,6 +1267,37 @@ class ArgsTestCase(BaseTestCase):
         self.assertIn("Warning -- Unraisable exception", output)
         self.assertIn("Exception: weakref callback bug", output)
 
+    def test_threading_excepthook(self):
+        # --fail-env-changed must catch uncaught thread exception.
+        # The exception must be displayed even if sys.stderr is redirected.
+        code = textwrap.dedent(r"""
+            import threading
+            import unittest
+            from test.support import captured_stderr
+
+            class MyObject:
+                pass
+
+            def func_bug():
+                raise Exception("bug in thread")
+
+            class Tests(unittest.TestCase):
+                def test_threading_excepthook(self):
+                    with captured_stderr() as stderr:
+                        thread = threading.Thread(target=func_bug)
+                        thread.start()
+                        thread.join()
+                    self.assertEqual(stderr.getvalue(), '')
+        """)
+        testname = self.create_test(code=code)
+
+        output = self.run_tests("--fail-env-changed", "-v", testname, exitcode=3)
+        self.check_executed_tests(output, [testname],
+                                  env_changed=[testname],
+                                  fail_env_changed=True)
+        self.assertIn("Warning -- Uncaught thread exception", output)
+        self.assertIn("Exception: bug in thread", output)
+
     def test_cleanup(self):
         dirname = os.path.join(self.tmptestdir, "test_python_123")
         os.mkdir(dirname)