Function _recursive_acquisition_script

tests/test_read_write_fork.py:440–481

()

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

        import sys

        from filelock import ReadWriteLock

        lock_path, operation = sys.argv[1:]
        lock = ReadWriteLock(lock_path)
        armed = True

        from typing import TYPE_CHECKING

        if TYPE_CHECKING:
            from _typeshed import Unused

        def audit_hook(event: str, _args: Unused) -> None:
            if armed and event == "sqlite3.connect":
                if operation == "acquire":
                    lock.acquire_read()
                elif operation == "release":
                    lock.release()
                else:
                    lock.close()

        sys.addaudithook(audit_hook)
        try:
            lock.acquire_read()
        except RuntimeError as error:
            assert str(error) == (
                f"Cannot {operation} ReadWriteLock on {lock_path} while acquisition is active in this thread"
            )
        else:
            raise AssertionError(f"recursive {operation} succeeded")

        armed = False
        with lock.read_lock():
            pass
        """
    )


def _ctypes_signature_script() -> str:  # pragma: <3.12 cover