Class BlockingLock

tests/test_fork_coordination.py:1136–1148

    parent_pid = os.getpid()

    class BlockingLock(BaseFileLock):
        def __init__(self, lock_file: str, *, is_singleton: bool = False) -> None:
            if os.getpid() == parent_pid:
                entered.set()
                release.wait(timeout=5)
            super().__init__(lock_file, thread_local=False, is_singleton=is_singleton)

        def _acquire(self) -> None:
            self._context.lock_file_fd = os.open(self.lock_file, os.O_CREAT | os.O_RDWR, 0o600)

        def _release(self) -> None:
            os.close(self._context.lock_file_fd if self._context.lock_file_fd is not None else -1)
            self._context.lock_file_fd = None

    worker = threading.Thread(target=BlockingLock, args=(str(tmp_path / "mutex.lock"),), kwargs={"is_singleton": True})
    worker.start()