tests/test_io_utils.py

from __future__ import annotations

import errno
import json
import os
from pathlib import Path

import autocad_bench.common.io as io_utils


def test_atomic_write_json_retries_transient_replace_failure(
    monkeypatch, tmp_path: Path
) -> None:
    attempts = 0
    real_replace = os.replace
    sleeps: list[float] = []

    def flaky_replace(source, destination) -> None:
        nonlocal attempts
        attempts += 1
        if attempts < 3:
            raise PermissionError(errno.EPERM, "transient replace denial")
        real_replace(source, destination)

    monkeypatch.setattr(io_utils.os, "replace", flaky_replace)
    target = tmp_path / "state.json"

    io_utils.atomic_write_json(
        target,
        {"state": "running"},
        sleep=sleeps.append,
    )

    assert attempts == 3
    assert sleeps == [0.05, 0.1]
    assert json.loads(target.read_text(encoding="utf-8")) == {"state": "running"}
    assert list(tmp_path.glob(".*.tmp")) == []