Function _fork_script

tests/test_read_write_fork.py:577–672

()

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

        import gc
        import os
        import sqlite3
        import sys
        from typing import TYPE_CHECKING

        if TYPE_CHECKING:
            from _typeshed import Unused

        lock_path, mode, fork_name, audit_hook_mode = sys.argv[1:]
        if audit_hook_mode == "reject":
            def reject_filelock_audit_hook(event: str, _args: Unused) -> None:
                if event == "sys.addaudithook":
                    raise RuntimeError("audit hook registration rejected")

            sys.addaudithook(reject_filelock_audit_hook)

        from filelock import ReadWriteLock, Timeout

        lock = ReadWriteLock(lock_path, is_singleton=False)
        acquire = lock.acquire_read if mode == "read" else lock.acquire_write
        proxy = acquire()
        parent_to_child_read, parent_to_child_write = os.pipe()
        child_to_parent_read, child_to_parent_write = os.pipe()

        child_pid = getattr(os, fork_name)()
        if child_pid == 0:
            os.close(parent_to_child_write)
            os.close(child_to_parent_read)
            try:
                with proxy:
                    pass
                try:
                    lock.acquire_read()
                except RuntimeError as error:
                    assert "was invalidated by fork()" in str(error)
                else:
                    raise AssertionError("inherited acquisition succeeded")
                lock.release()
                lock.close()
                del proxy
                del lock
                gc.collect()
                sqlite_connects = 0

                def count_sqlite_connects(event: str, _args: Unused) -> None:
                    global sqlite_connects
                    if event == "sqlite3.connect":
                        sqlite_connects += 1

                sys.addaudithook(count_sqlite_connects)
                try:
                    ReadWriteLock(lock_path, is_singleton=False)