Method test_fork_asyncio_run

Lib/test/test_asyncio/test_unix_events.py:1297–1313

(self)

    @hashlib_helper.requires_hashdigest('md5')
    @support.skip_if_sanitizer("TSAN doesn't support threads after fork", thread=True)
    def test_fork_asyncio_run(self):
        self.addCleanup(multiprocessing_cleanup_tests)

        ctx = multiprocessing.get_context('fork')
        manager = ctx.Manager()
        self.addCleanup(manager.shutdown)
        result = manager.Value('i', 0)

        async def child_main():
            await asyncio.sleep(0.1)
            result.value = 42

        process = ctx.Process(target=lambda: asyncio.run(child_main()))
        process.start()
        process.join()

        self.assertEqual(result.value, 42)

    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
    @hashlib_helper.requires_hashdigest('md5')