Function _async_fork_script

tests/test_read_write_fork.py:945–1029

()

def _async_fork_script() -> str:
    return textwrap.dedent(
        r"""
        from __future__ import annotations

        import asyncio
        import gc
        import os
        import sqlite3
        import sys
        import warnings

        from filelock import AsyncAcquireReadWriteReturnProxy, AsyncReadWriteLock, ReadWriteLock, Timeout

        warnings.filterwarnings("ignore", category=DeprecationWarning, message=r".*fork\(\).*")
        lock_path, state = sys.argv[1:]

        async def set_up_parent() -> tuple[AsyncReadWriteLock, AsyncAcquireReadWriteReturnProxy | None]:
            parent_lock = AsyncReadWriteLock(lock_path, is_singleton=False)
            parent_proxy = await parent_lock.acquire_write() if state == "held" else None
            return parent_lock, parent_proxy

        lock, proxy = asyncio.run(set_up_parent())
        child_pid = os.fork()
        if child_pid == 0:
            async def check_child() -> None:
                if proxy is not None:
                    async with proxy:
                        pass
                try:
                    await lock.acquire_read()
                except RuntimeError as error:
                    assert "was invalidated by fork()" in str(error)
                else:
                    raise AssertionError("inherited async lock acquired")
                await lock.release()
                await lock.close()
                if sys.implementation.name == "pypy":
                    try:
                        AsyncReadWriteLock(lock_path, is_singleton=False)
                    except RuntimeError as error:
                        assert str(error) == (
                            "ReadWriteLock is unavailable in a PyPy fork child; exec or exit before using it"
                        )
                    else:
                        raise AssertionError("PyPy child constructed an async lock after fork")
                elif state == "idle":
                    fresh_lock = AsyncReadWriteLock(lock_path, is_singleton=False)
                    async with fresh_lock.write_lock():
                        pass
                    await fresh_lock.close()
                else:
                    try:
                        ReadWriteLock(lock_path, is_singleton=False)
                    except RuntimeError as error:
                        expected = (
                            "ReadWriteLock is unavailable in a PyPy fork child"
                            if sys.implementation.name == "pypy"