Function _close_after_commit

tests/test_soft_stale.py:645–672

(mocker: MockerFixture, lock: SoftFileLock)

@contextmanager
def _close_after_commit(mocker: MockerFixture, lock: SoftFileLock) -> Iterator[tuple[OSError, list[int]]]:
    real_close = os.close
    close_error = OSError(EINTR, "close failed")
    attempts: list[int] = []
    # close is an attribute of the shared os module, so this patch binds for every close in the process while the
    # window is open, including one a GC-run finalizer (a multiprocessing pipe from an earlier test, say) makes on this
    # very thread. Fail only the descriptor this lock lets go of: it clears its bookkeeping right before the one close
    # attempt, so record the descriptor as it does.
    released: list[int | None] = []
    real_mark_released = lock._mark_descriptor_released

    def mark_released() -> None:
        released.append(lock._context.lock_file_fd)
        real_mark_released()

    def close(fd: int) -> None:
        real_close(fd)
        if fd not in released:
            return
        attempts.append(fd)
        raise close_error

    mocker.patch.object(lock, "_mark_descriptor_released", side_effect=mark_released)
    close_mock = mocker.patch("filelock._soft.os.close", side_effect=close)
    try:
        yield close_error, attempts
    finally:
        mocker.stop(close_mock)


def test_soft_windows_unlink_gives_up_after_every_attempt_is_denied(tmp_path: Path, mocker: MockerFixture) -> None: