Class _DescriptorLock

tests/test_fork_backends.py:28–43

class _DescriptorLock(BaseFileLock):
    acquire_error: BaseException | None = None
    descriptor: int | None = None
    fail_release: bool = False

    def _acquire(self) -> None:  # pragma: needs fork
        self.descriptor = self._context.lock_file_fd = os.open(self.lock_file, os.O_CREAT | os.O_RDWR, 0o600)
        if self.acquire_error is not None:
            raise self.acquire_error

    def _release(self) -> None:  # pragma: needs fork
        if self.fail_release:
            msg = "unlock failed"
            raise RuntimeError(msg)
        os.close(cast("int", self._context.lock_file_fd))
        self._context.lock_file_fd = None


class _CoroutineDescriptorLock(BaseAsyncFileLock):