added tests, changed test_threading.ThreadTests.test_fork_at_exit to …

…fit new behaviour
python · gpshead · Jun 4, 2023 · May 24, 2023 · May 24, 2023 · May 24, 2023
commit 77443a04886a82fa26d19e8bd9739b08894895f0
diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py
@@ -4700,6 +4700,22 @@ def test_fork_warns_when_non_python_thread_exists(self):
         self.assertEqual(err.decode("utf-8"), "")
         self.assertEqual(out.decode("utf-8"), "")
 
+    def test_fork_at_exit(self):
+        code = """if 1:
+            import atexit
+            import os
+
+            def exit_handler():
+                pid = os.fork()
+                if pid != 0:
+                    print("shouldn't be printed")
+
+            atexit.register(exit_handler)
+        """
+        _, out, err = assert_python_ok("-c", code)
+        self.assertEqual(b"", out)
+        self.assertIn(b"can't fork at interpreter shutdown", err)
+
 
 # Only test if the C version is provided, otherwise TestPEP519 already tested
 # the pure Python implementation.

@@ -5,6 +5,7 @@
 from test.support import import_helper
 from test.support import os_helper
 from test.support import warnings_helper
+from test.support.script_helper import assert_python_ok
 import subprocess
 import sys
 import signal
@@ -3329,6 +3330,21 @@ def test_communicate_repeated_call_after_stdout_close(self):
             except subprocess.TimeoutExpired:
                 pass
 
+    def test_create_child_process_at_exit(self):
+        code = f"""if 1:
+        import atexit
+        import subprocess
+
+        def exit_handler():
+            subprocess.Popen({ZERO_RETURN_CMD})
+            print("shouldn't be printed")
+
+        atexit.register(exit_handler)
+        """
+        _, out, err = assert_python_ok("-c", code)
+        self.assertEqual(out, b'')
+        self.assertIn(b"can't create child process at interpreter shutdown", err)
+
 
 @unittest.skipUnless(mswindows, "Windows specific tests")
 class Win32ProcessTestCase(BaseTestCase):

diff --git a/Lib/test/test_threading.py b/Lib/test/test_threading.py
@@ -557,7 +557,7 @@ def exit_handler():
         """)
         _, out, err = assert_python_ok("-c", code)
         self.assertEqual(out, b'')
-        self.assertEqual(err.rstrip(), b'child process ok')
+        self.assertIn(b"can't fork at interpreter shutdown", err)
 
     @support.requires_fork()
     def test_dummy_thread_after_fork(self):
@@ -1048,6 +1048,22 @@ def import_threading():
         self.assertEqual(out, b'')
         self.assertEqual(err, b'')
 
+    def test_start_new_thread_at_exit(self):
+        code = """if 1:
+            import atexit
+            import _thread
+
+            def f():
+                print("shouldn't be printed")
+
+            def exit_handler():
+                _thread.start_new_thread(f, ())
+
+            atexit.register(exit_handler)
+        """
+        _, out, err = assert_python_ok("-c", code)
+        self.assertEqual(out, b'')
+        self.assertIn(b"can't create new thread at interpreter shutdown", err)
 
 class ThreadJoinOnShutdown(BaseTestCase):