pypy/module/thread/test/test_fork.py::AppTestFork::()::test_nested_import_lock_fork
self = <pypy.interpreter.typedef.W_ObjectObjectUserDictWeakrefable object at 0x000000010913fa98>
def test_nested_import_lock_fork(self):
"""Check fork() in main thread works while the main thread is doing an import"""
# Issue 9573: this used to trigger RuntimeError in the child process
import imp
import os
import time
if not hasattr(os, 'fork'):
skip("No fork on this platform")
def fork_with_import_lock(level):
release = 0
in_child = False
try:
try:
for i in range(level):
imp.acquire_lock()
release += 1
pid = os.fork()
in_child = not pid
finally:
for i in range(release):
imp.release_lock()
except RuntimeError:
if in_child:
if verbose > 1:
print("RuntimeError in child")
os._exit(1)
raise
if in_child:
os._exit(0)
for i in range(10):
spid, status = os.waitpid(pid, os.WNOHANG)
if spid == pid:
break
time.sleep(1.0)
assert spid == pid
assert status == 0
# Check this works with various levels of nested
# import in the main thread
for level in range(5):
> fork_with_import_lock(level)
[/Users/matti/build-worker-arm64/own-macos-arm64/build/pypy/module/thread/test/test_fork.py:93]:44:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
level = W_IntObject(2)
def fork_with_import_lock(level):
release = 0
in_child = False
try:
try:
for i in range(level):
imp.acquire_lock()
release += 1
pid = os.fork()
in_child = not pid
finally:
for i in range(release):
imp.release_lock()
except RuntimeError:
if in_child:
if verbose > 1:
print("RuntimeError in child")
os._exit(1)
raise
if in_child:
os._exit(0)
for i in range(10):
spid, status = os.waitpid(pid, os.WNOHANG)
if spid == pid:
break
time.sleep(1.0)
assert spid == pid
> assert status == 0
E (application-level) AssertionError
[/Users/matti/build-worker-arm64/own-macos-arm64/build/pypy/module/thread/test/test_fork.py:93]:39: AssertionError
builder: own-macos-arm64 build #1313+
test: pypy/module/thread/test/test_fork.py::AppTestFork::()::test_nested_import_lock_fork