bpo-40094: Enhance threading tests (GH-19260)
* Rewrite test_thread.test_forkinthread() to use
support.wait_process() and wait for the child process in the main
thread, not in the spawned thread.
* test_threading now uses support.wait_process() and checks the child
process exit code to detect crashes.
diff --git a/Lib/test/test_thread.py b/Lib/test/test_thread.py
index 9f4801f..77e46f2 100644
--- a/Lib/test/test_thread.py
+++ b/Lib/test/test_thread.py
@@ -225,30 +225,31 @@
@unittest.skipUnless(hasattr(os, 'fork'), 'need os.fork')
@support.reap_threads
def test_forkinthread(self):
- status = "not set"
+ pid = None
- def thread1():
- nonlocal status
+ def fork_thread(read_fd, write_fd):
+ nonlocal pid
# fork in a thread
pid = os.fork()
- if pid == 0:
- # child
- try:
- os.close(self.read_fd)
- os.write(self.write_fd, b"OK")
- finally:
- os._exit(0)
- else:
- # parent
- os.close(self.write_fd)
- pid, status = os.waitpid(pid, 0)
+ if pid:
+ # parent process
+ return
+
+ # child process
+ try:
+ os.close(read_fd)
+ os.write(write_fd, b"OK")
+ finally:
+ os._exit(0)
with support.wait_threads_exit():
- thread.start_new_thread(thread1, ())
- self.assertEqual(os.read(self.read_fd, 2), b"OK",
- "Unable to fork() in thread")
- self.assertEqual(status, 0)
+ thread.start_new_thread(fork_thread, (self.read_fd, self.write_fd))
+ self.assertEqual(os.read(self.read_fd, 2), b"OK")
+ os.close(self.write_fd)
+
+ self.assertIsNotNone(pid)
+ support.wait_process(pid, exitcode=0)
def tearDown(self):
try: