vrsub/tests/app/test_storage/test_atomic_copy.py

"""src/wov_app/storage.py 的模块级测试（数据 → 测试过程 → 验证结果）。

被测模块：`src/wov_app/storage.py`（产物文件原子复制），被 scheduler 与
batch 复用，也可独立调用。用例使用真实文件系统（tmp_path）验证复制完整性与
失败时对原目标的保护。
"""

from __future__ import annotations

import stat
from pathlib import Path

import pytest

from wov_app.storage import atomic_copy


def test_atomic_copy_creates_target_with_same_content(tmp_path: Path) -> None:
    """复制后目标存在且内容与源一致（源保留）。"""
    # 数据：真实的源文件（含中文内容）。
    source = tmp_path / "src.srt"
    source.write_text("1\n00:00:01,000 --> 00:00:02,000\n你好\n", encoding="utf-8")
    target = tmp_path / "out" / "movie.CN.srt"

    # 测试过程
    atomic_copy(source, target)

    # 验证结果：源保留、目标内容一致。
    assert source.is_file()
    assert target.read_text(encoding="utf-8") == source.read_text(encoding="utf-8")


def test_atomic_copy_creates_parent_directories(tmp_path: Path) -> None:
    """目标父目录不存在时自动创建（产物目录可能尚未建立）。"""
    # 数据：深层不存在的目录。
    source = tmp_path / "a.txt"
    source.write_text("x", encoding="utf-8")
    target = tmp_path / "deep" / "nested" / "dir" / "a.txt"

    # 测试过程
    atomic_copy(source, target)

    # 验证结果
    assert target.is_file()
    assert target.parent.is_dir()


def test_atomic_copy_overwrites_existing_target(tmp_path: Path) -> None:
    """目标已存在时被完整替换（旧内容不残留）。"""
    # 数据：已存在的旧目标（内容更长，用于检测残留）。
    source = tmp_path / "new.srt"
    source.write_text("新", encoding="utf-8")
    target = tmp_path / "movie.CN.srt"
    target.write_text("旧的非常长的内容" * 100, encoding="utf-8")

    # 测试过程
    atomic_copy(source, target)

    # 验证结果：内容被完全替换，无旧内容残留。
    assert target.read_text(encoding="utf-8") == "新"


def test_atomic_copy_preserves_binary_content(tmp_path: Path) -> None:
    """二进制内容（如 ASS 的 BOM/CRLF）逐字节一致。"""
    # 数据：带 BOM 与 CRLF 的字节序列。
    payload = "\ufeff[Script Info]\r\nTitle: 测试\r\n".encode("utf-8")
    source = tmp_path / "src.ass"
    source.write_bytes(payload)
    target = tmp_path / "dst.ass"

    # 测试过程
    atomic_copy(source, target)

    # 验证结果：逐字节一致。
    assert target.read_bytes() == payload


def test_atomic_copy_cleans_temp_file_on_success(tmp_path: Path) -> None:
    """成功后不留下临时文件（目录里只有源与目标）。"""
    # 数据：源文件。
    source = tmp_path / "src.txt"
    source.write_text("x", encoding="utf-8")
    target = tmp_path / "dst.txt"

    # 测试过程
    atomic_copy(source, target)

    # 验证结果：无 *.tmp 残留。
    assert not list(tmp_path.glob("*.tmp"))


def test_atomic_copy_keeps_target_and_cleans_temp_on_failure(tmp_path: Path) -> None:
    """复制失败时保留原目标、清理临时文件（R03：不暴露写了一半的成品）。"""
    # 数据：源文件不存在（触发 copy2 失败），目标已存在有效内容。
    missing = tmp_path / "missing.txt"
    target = tmp_path / "movie.CN.srt"
    target.write_text("原有有效成品", encoding="utf-8")

    # 测试过程与验证结果：抛错，原目标未被破坏，无临时文件残留。
    with pytest.raises(FileNotFoundError):
        atomic_copy(missing, target)
    assert target.read_text(encoding="utf-8") == "原有有效成品"
    assert not list(tmp_path.glob("*.tmp"))


def test_atomic_copy_exists_with_missing_target_on_failure(tmp_path: Path) -> None:
    """复制失败且目标原本不存在时，不产生残缺目标文件。"""
    # 数据：源不存在、目标不存在。
    missing = tmp_path / "missing.txt"
    target = tmp_path / "new.CN.srt"

    # 测试过程与验证结果
    with pytest.raises(FileNotFoundError):
        atomic_copy(missing, target)
    assert not target.exists()
    assert not list(tmp_path.glob("*.tmp"))


def test_atomic_copy_preserves_source_permissions_semantics(tmp_path: Path) -> None:
    """目标可读（copy2 保留元数据，权限不会导致后续读取失败）。"""
    # 数据：普通源文件。
    source = tmp_path / "src.txt"
    source.write_text("内容", encoding="utf-8")
    target = tmp_path / "dst.txt"

    # 测试过程
    atomic_copy(source, target)

    # 验证结果：目标可读且非空。
    assert target.read_text(encoding="utf-8") == "内容"
    assert target.stat().st_size > 0