mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-10-21 14:12:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

bpo-40094: Enhance threading tests (GH-19260)

Browse source 
* 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.
This commit is contained in:
Victor Stinner 2020-03-31 21:49:44 +02:00 • committed by GitHub
parent 27c6231f58
commit a9f9687a7c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 39 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

37
Lib/test/test_thread.py
View file

@ -225,30 +225,31 @@ class TestForkInThread(unittest.TestCase):
@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: