"""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