Function _reentrant_finalizer_script

tests/test_read_write_fork.py:358–401

()

def _reentrant_finalizer_script() -> str:  # pragma: needs fork
    return textwrap.dedent(
        """
        from __future__ import annotations

        import asyncio
        import gc
        import sys
        import weakref
        from threading import Condition, Event
        from types import CodeType, FrameType
        from typing import Final

        from filelock import AsyncReadWriteLock, ReadWriteLock

        class _FinalizerTarget:
            cycle: _FinalizerTarget

        _FINALIZED: Final[Event] = Event()

        def _close_lock() -> None:
            ReadWriteLock(sys.argv[2], is_singleton=False).close()
            _FINALIZED.set()

        _TARGET = _FinalizerTarget()
        _TARGET.cycle = _TARGET
        weakref.finalize(_TARGET, _close_lock)
        del _TARGET

        _NOTIFY_CODE: Final[CodeType] = Condition.notify_all.__code__

        def _collect_at_notify(frame: FrameType, _event: str, _argument: None) -> None:
            if frame.f_code is _NOTIFY_CODE:
                sys.settrace(None)
                gc.collect()

        sys.settrace(_collect_at_notify)
        _LOCK: Final[AsyncReadWriteLock] = AsyncReadWriteLock(sys.argv[1], is_singleton=False)
        sys.settrace(None)

        assert _FINALIZED.is_set()
        asyncio.run(_LOCK.close())
        """
    )


def _recursive_construction_script() -> str: