test: 按模块重写测试代码,删除旧平铺结构

按"测试规则"重写 tests/:一个模块一个目录、用例按数据→过程→验证三段书写、
不保留全局 conftest.py、测试过程只调用真实生产代码。

结构(73 个文件、30 个模块目录、477 用例):
- tests/nodes/  15 个模块目录(srt/whisper/ass/ffmpeg/frame_extract/vlm/
  subtitle_ocr/llm/llm_filter/subtitle_cleanup/subtitle_correction/
  proper_nouns/adaptive_pool/vad_profiler/echo);
- tests/app/    11 个模块目录(db/scheduler/batch/maintenance/registry/seed/
  storage/config/logging/main/routers 三组 API);
- tests/sdk/test_models、tests/web/test_crop、tests/shared(公共设施)。

测试数据随模块目录入库(tests/**/data/),删除根级 testdata/;.gitignore
的 data/ 改为 /data/,否则会连带忽略 tests/**/data/ 导致测试数据无法入库。

顺带发现并修复三个真实缺陷:
- nodes/srt.py:相邻条目缺少空行时把下一条时间轴吞进正文(静默错位),
  改为正文行遇时间戳行即报错;
- src/wov_app/scheduler.py:_file_size 只捕获 OSError,含 \x00 的产物 URI
  抛 ValueError 导致任务误判失败,改为同时捕获;
- nodes/subtitle_correction.py:生产代码依赖测试包解析 SRT,
  改用生产模块 nodes/srt.py。

真实模型/服务集成测试按外部状态跳过:新增 tests/shared/gpu_memory.py
(运行时探测显存、CUDA OOM 转跳过)与 tests/shared/llm_service.py
(无 Key / 余额 / 限流转跳过)。全量 477 passed。
This commit is contained in:
2026-09-13 15:40:56 +08:00
parent 966f3e6b4b
commit 8a715a8064
139 changed files with 20810 additions and 10733 deletions
View File
+131
View File
@@ -0,0 +1,131 @@
"""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