cat-shark
8a715a8064
按"测试规则"重写 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。
289 lines
9.7 KiB
Python
289 lines
9.7 KiB
Python
"""nodes/ass.py 的模块级测试（数据 → 测试过程 → 验证结果）。
|
||
|
||
被测模块：`nodes/ass.py`（SRT → VR 双目 ASS，含统一样式出口），被
|
||
srt-to-dual-eye-ass 节点与历史字幕统脚本共用，可独立调用。
|
||
|
||
覆盖：SRT 解析（含多行与畸形行）、ASS 头与样式行、左右眼对照与零视差、
|
||
顶部安全区（an8 + MarginV=700）、透明度、margin_top 参数化、invoke 全流程。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
from nodes.ass import (
|
||
DEFAULT_MARGIN_TOP,
|
||
_ass_header,
|
||
ass_header,
|
||
dialogue_line,
|
||
invoke,
|
||
parse_srt,
|
||
style_row,
|
||
write_ass,
|
||
)
|
||
from wov_sdk.models import InvokeRequest
|
||
|
||
# 模块专用测试数据：两条（第二条为多行）SRT。
|
||
SAMPLE_SRT = """1
|
||
00:00:01,000 --> 00:00:02,000
|
||
第一行
|
||
第二行
|
||
|
||
2
|
||
00:00:04,000 --> 00:00:06,000
|
||
第三行
|
||
"""
|
||
|
||
|
||
def _request(tmp_path: Path, srt_text: str, **params) -> InvokeRequest:
|
||
"""构造真实 InvokeRequest：把输入 SRT 落到临时目录，输出目录同目录下。"""
|
||
srt_path = tmp_path / "input.srt"
|
||
srt_path.write_text(srt_text, encoding="utf-8")
|
||
return InvokeRequest(
|
||
run_id="run-test",
|
||
node_instance_id="ass-1",
|
||
params=params,
|
||
inputs={"cn_srt_uri": str(srt_path)},
|
||
output_dir=str(tmp_path / "out"),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# SRT 解析
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_parse_srt_converts_comma_to_dot_and_joins_multiline() -> None:
|
||
"""SRT 时间戳逗号转 ASS 点号；多行正文用 \\N 连接。"""
|
||
# 数据：两条，第一条两行正文。
|
||
text = SAMPLE_SRT
|
||
|
||
# 测试过程
|
||
entries = parse_srt(text)
|
||
|
||
# 验证结果：条数、时间戳格式与多行连接符。
|
||
assert len(entries) == 2
|
||
assert entries[0][0] == "00:00:01.000"
|
||
assert entries[0][1] == "00:00:02.000"
|
||
assert entries[0][2] == r"第一行\N第二行"
|
||
|
||
|
||
def test_parse_srt_skips_malformed_timeline() -> None:
|
||
"""缺少时间轴分隔符的畸形条目被跳过，不产生错误条目。"""
|
||
# 数据：一条畸形（无 -->）+ 一条正常。
|
||
text = "1\n00:00:01,000\n坏条目\n\n2\n00:00:04,000 --> 00:00:06,000\n正常\n"
|
||
|
||
# 测试过程
|
||
entries = parse_srt(text)
|
||
|
||
# 验证结果：只解析出正常条目。
|
||
assert len(entries) == 1
|
||
assert entries[0][2] == "正常"
|
||
|
||
|
||
def test_parse_srt_empty_text() -> None:
|
||
"""空文本解析为空列表。"""
|
||
# 数据：空字符串。
|
||
# 测试过程与验证结果
|
||
assert parse_srt("") == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 样式：单一事实来源
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_ass_header_contains_resolution_and_styles() -> None:
|
||
"""ASS 头包含分辨率、左右眼两条样式与必需字段。"""
|
||
# 数据：VR 常用分辨率 3840x1920。
|
||
# 测试过程
|
||
header = ass_header(3840, 1920)
|
||
|
||
# 验证结果
|
||
assert "PlayResX: 3840" in header
|
||
assert "PlayResY: 1920" in header
|
||
assert "Style: LeftEye," in header
|
||
assert "Style: RightEye," in header
|
||
assert "[Script Info]" in header and "[V4+ Styles]" in header and "[Events]" in header
|
||
|
||
|
||
def test_style_row_top_aligned_with_margin() -> None:
|
||
"""样式行使用 an8 顶部对齐 + MarginV 顶部安全边距（默认 700）。"""
|
||
# 数据：左眼样式，宽 3840，默认边距。
|
||
# 测试过程
|
||
row = style_row("LeftEye", 3840)
|
||
|
||
# 验证结果：末尾字段为 Alignment=8、MarginL=50、MarginR=1920（右眼区）、MarginV=700。
|
||
assert row.endswith(",8,50,1920,700,1")
|
||
|
||
|
||
def test_style_row_uses_translucent_fill_and_outline() -> None:
|
||
"""文字填充约 70% 透明、描边半透明黑，降低对画面的遮挡。"""
|
||
# 数据：任一样式行。
|
||
# 测试过程
|
||
row = style_row("LeftEye", 3840)
|
||
|
||
# 验证结果
|
||
assert "&HB3FFFFFF" in row
|
||
assert "&H80000000" in row
|
||
assert "Arial" in row
|
||
|
||
|
||
def test_default_margin_top_is_700() -> None:
|
||
"""默认顶部安全边距常量是 700（2026-09 调整，历史 120 已废弃）。"""
|
||
# 数据：模块常量。
|
||
# 测试过程与验证结果
|
||
assert DEFAULT_MARGIN_TOP == 700
|
||
|
||
|
||
def test_dialogue_line_prefixes_top_alignment() -> None:
|
||
"""Dialogue 行固定前缀 {\\an8}，保证每条字幕落在顶部安全区。"""
|
||
# 数据：一条字幕。
|
||
# 测试过程
|
||
line = dialogue_line("LeftEye", "0:00:01.00", "0:00:02.00", "你好")
|
||
|
||
# 验证结果
|
||
assert line == r"Dialogue: 0,0:00:01.00,0:00:02.00,LeftEye,,0,0,0,,{\an8}你好"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 写出：左右眼与零视差
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_write_ass_emits_both_eyes_with_identical_text(tmp_path: Path) -> None:
|
||
"""每个条目输出左右眼两行 Dialogue，文本与水平相对位置完全一致（零视差）。"""
|
||
# 数据：单条字幕。
|
||
entries = parse_srt("1\n00:00:01,000 --> 00:00:02,000\n台词\n")
|
||
output = tmp_path / "out.ass"
|
||
|
||
# 测试过程
|
||
write_ass(entries, output, "3840x1920")
|
||
content = output.read_text(encoding="utf-8")
|
||
|
||
# 验证结果：两条 Dialogue（LeftEye/RightEye），文本各出现一次。
|
||
dialogue = [line for line in content.splitlines() if line.startswith("Dialogue:")]
|
||
assert len(dialogue) == 2
|
||
assert "LeftEye" in dialogue[0] and "RightEye" in dialogue[1]
|
||
assert dialogue[0].split(",,")[-1] == dialogue[1].split(",,")[-1]
|
||
|
||
|
||
def test_write_ass_custom_margin_top_changes_style_row(tmp_path: Path) -> None:
|
||
"""margin_top 参数化到样式行（不同分辨率/内容可微调顶部边距）。"""
|
||
# 数据：自定义边距 300。
|
||
entries = parse_srt(SAMPLE_SRT)
|
||
output = tmp_path / "out.ass"
|
||
|
||
# 测试过程
|
||
write_ass(entries, output, "3840x1920", margin_top=300)
|
||
content = output.read_text(encoding="utf-8")
|
||
|
||
# 验证结果：样式行的 MarginV 为 300。
|
||
assert ",8,50,1920,300,1" in content
|
||
|
||
|
||
def test_write_ass_multiple_entries_keep_order(tmp_path: Path) -> None:
|
||
"""多条字幕按输入顺序输出（时间轴顺序不被打乱）。"""
|
||
# 数据：两条字幕。
|
||
entries = parse_srt(SAMPLE_SRT)
|
||
output = tmp_path / "out.ass"
|
||
|
||
# 测试过程
|
||
write_ass(entries, output, "3840x1920")
|
||
dialogue = [
|
||
line for line in output.read_text(encoding="utf-8").splitlines()
|
||
if line.startswith("Dialogue:")
|
||
]
|
||
|
||
# 验证结果：第 1、2 行属于第一条，第 3、4 行属于第二条。
|
||
assert "00:00:01.000" in dialogue[0] and "00:00:02.000" in dialogue[1]
|
||
assert "00:00:04.000" in dialogue[2] and "00:00:06.000" in dialogue[3]
|
||
|
||
|
||
def test_ass_header_compat_interface_matches_new_header() -> None:
|
||
"""兼容接口 _ass_header 与新版 ass_header 对同一分辨率输出一致（样式不漂移）。"""
|
||
# 数据：同一分辨率字符串与整数宽高。
|
||
# 测试过程与验证结果
|
||
assert _ass_header("3840x1920") == ass_header(3840, 1920)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# invoke 全流程
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_invoke_writes_ass_and_returns_uri(tmp_path: Path) -> None:
|
||
"""invoke 读取 cn_srt_uri，写出 dual_eye.ass 并返回产物 URI。"""
|
||
# 数据：真实输入 SRT 文件 + 默认参数。
|
||
request = _request(tmp_path, SAMPLE_SRT)
|
||
|
||
# 测试过程
|
||
response = invoke(request)
|
||
|
||
# 验证结果：状态、产物存在、内容含左右眼。
|
||
assert response.status == "completed"
|
||
artifact = Path(response.outputs["ass_uri"])
|
||
assert artifact.is_file()
|
||
assert artifact.name == "dual_eye.ass"
|
||
assert "LeftEye" in artifact.read_text(encoding="utf-8")
|
||
|
||
|
||
def test_invoke_honors_resolution_and_margin_params(tmp_path: Path) -> None:
|
||
"""invoke 透传 resolution 与 margin_top 参数到产物。"""
|
||
# 数据：自定义分辨率 1920x1080 与边距 250。
|
||
request = _request(tmp_path, SAMPLE_SRT, resolution="1920x1080", margin_top=250)
|
||
|
||
# 测试过程
|
||
response = invoke(request)
|
||
|
||
# 验证结果
|
||
content = Path(response.outputs["ass_uri"]).read_text(encoding="utf-8")
|
||
assert "PlayResX: 1920" in content
|
||
assert "PlayResY: 1080" in content
|
||
assert "250,1" in content
|
||
|
||
|
||
def test_invoke_default_margin_top_is_700(tmp_path: Path) -> None:
|
||
"""未传 margin_top 时使用默认 700。"""
|
||
# 数据：不传 margin_top。
|
||
request = _request(tmp_path, SAMPLE_SRT)
|
||
|
||
# 测试过程
|
||
response = invoke(request)
|
||
|
||
# 验证结果
|
||
content = Path(response.outputs["ass_uri"]).read_text(encoding="utf-8")
|
||
assert ",8,50,1920,700,1" in content
|
||
|
||
|
||
def test_invoke_fails_without_input_uri(tmp_path: Path) -> None:
|
||
"""缺少 cn_srt_uri 输入时返回 failed 并说明原因。"""
|
||
# 数据：空输入。
|
||
request = InvokeRequest(
|
||
run_id="r", node_instance_id="n", params={}, inputs={}, output_dir=str(tmp_path)
|
||
)
|
||
|
||
# 测试过程
|
||
response = invoke(request)
|
||
|
||
# 验证结果
|
||
assert response.status == "failed"
|
||
assert "cn_srt_uri" in (response.error or "")
|
||
|
||
|
||
def test_invoke_fails_when_input_file_missing(tmp_path: Path) -> None:
|
||
"""输入指向不存在的文件时返回 failed。"""
|
||
# 数据：不存在的路径。
|
||
request = InvokeRequest(
|
||
run_id="r", node_instance_id="n", params={},
|
||
inputs={"cn_srt_uri": str(tmp_path / "missing.srt")},
|
||
output_dir=str(tmp_path / "out"),
|
||
)
|
||
|
||
# 测试过程
|
||
response = invoke(request)
|
||
|
||
# 验证结果
|
||
assert response.status == "failed"
|
||
assert "not found" in (response.error or "")
|