bundle/v0/tests/test_atomic_publish.py

from __future__ import annotations

from pathlib import Path

import pytest

from loss_aware_dro_repro.core import ContractError, publish_directory_atomically


def test_atomic_publish_retries_transient_permission_lock(tmp_path: Path):
    staging = tmp_path / ".staging"
    destination = tmp_path / "published"
    staging.mkdir()
    (staging / "result.json").write_text("{}\n", encoding="utf-8")
    attempts = 0
    delays: list[float] = []

    def transient_rename(source: Path, target: Path) -> None:
        nonlocal attempts
        attempts += 1
        if attempts < 3:
            raise PermissionError("simulated OneDrive sync lock")
        source.rename(target)

    publish_directory_atomically(
        staging,
        destination,
        _rename=transient_rename,
        _sleep=delays.append,
    )

    assert attempts == 3
    assert delays == [0.05, 0.1]
    assert not staging.exists()
    assert (destination / "result.json").read_text(encoding="utf-8") == "{}\n"


def test_atomic_publish_fails_closed_if_destination_appears(tmp_path: Path):
    staging = tmp_path / ".staging"
    destination = tmp_path / "published"
    staging.mkdir()

    def colliding_rename(source: Path, target: Path) -> None:
        target.mkdir()
        raise PermissionError("simulated competing publication")

    with pytest.raises(ContractError, match="appeared during publication"):
        publish_directory_atomically(
            staging,
            destination,
            _rename=colliding_rename,
            _sleep=lambda _: None,
        )

    assert staging.is_dir()
    assert destination.is_dir()


def test_atomic_publish_exhaustion_preserves_staging_for_caller_cleanup(
    tmp_path: Path,
):
    staging = tmp_path / ".staging"
    destination = tmp_path / "published"
    staging.mkdir()
    attempts = 0

    def always_locked(source: Path, target: Path) -> None:
        nonlocal attempts
        attempts += 1
        raise PermissionError("persistent lock")

    with pytest.raises(PermissionError, match="persistent lock"):
        publish_directory_atomically(
            staging,
            destination,
            attempts=3,
            _rename=always_locked,
            _sleep=lambda _: None,
        )

    assert attempts == 3
    assert staging.is_dir()
    assert not destination.exists()


def test_atomic_publish_rejects_cross_parent_move(tmp_path: Path):
    staging = tmp_path / "first" / ".staging"
    destination = tmp_path / "second" / "published"
    staging.mkdir(parents=True)
    destination.parent.mkdir()

    with pytest.raises(ContractError, match="one parent directory"):
        publish_directory_atomically(staging, destination)