

def test_threaded_import_lock_fork(self):
    'Check fork() in main thread works while a subthread is doing an import'
    import_started = threading.Event()
    fake_module_name = 'fake test module'
    partial_module = 'partial'
    complete_module = 'complete'

    def importer():
        imp.acquire_lock()
        sys.modules[fake_module_name] = partial_module
        import_started.set()
        time.sleep(0.01)
        sys.modules[fake_module_name] = complete_module
        imp.release_lock()
    t = threading.Thread(target=importer)
    t.start()
    import_started.wait()
    pid = os.fork()
    try:
        if (not pid):
            m = __import__(fake_module_name)
            if (m == complete_module):
                os._exit(0)
            else:
                if (verbose > 1):
                    print('Child encountered partial module')
                os._exit(1)
        else:
            t.join()
            self.wait_impl(pid)
    finally:
        try:
            os.kill(pid, signal.SIGKILL)
        except OSError:
            pass

dataset/ETHPy150Open amrdraz/kodr/app/brython/www/src/Lib/test/test_fork1.py ForkTest.test_threaded_import_lock_fork/original