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。
223 lines
8.5 KiB
Python
223 lines
8.5 KiB
Python
"""nodes/frame_extract.py 的模块级测试（数据 → 测试过程 → 验证结果）。
|
||
|
||
被测模块：`nodes/frame_extract.py`（按帧间隔抽帧 + crop 裁切 + 帧清单），
|
||
可独立调用。抽帧用例使用模块 `data/` 下的真实视频调用真实 ffmpeg；
|
||
帧号排序用例构造真实命名的帧文件（重现 14236 帧任务的错位场景）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from nodes.frame_extract import (
|
||
DEFAULT_CROP,
|
||
_frame_step,
|
||
_parse_crop,
|
||
_parse_progress_line,
|
||
_sorted_frame_files,
|
||
invoke,
|
||
)
|
||
from wov_sdk.models import InvokeRequest
|
||
|
||
# 模块专用真实素材：10 秒测试视频（1280x720, 25fps）。
|
||
DATA_DIR = Path(__file__).resolve().parent / "data"
|
||
TEST_VIDEO = DATA_DIR / "subtitle_10s.mp4"
|
||
|
||
|
||
def _request(tmp_path: Path, video: Path | None, **params) -> InvokeRequest:
|
||
"""构造真实请求；video 为 None 时表示不传 video_uri。"""
|
||
inputs = {} if video is None else {"video_uri": str(video)}
|
||
return InvokeRequest(
|
||
run_id="run-test",
|
||
node_instance_id="frame-extract-1",
|
||
params=params,
|
||
inputs=inputs,
|
||
output_dir=str(tmp_path / "out"),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 参数解析与换算（纯函数）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_default_crop_is_bottom_quarter() -> None:
|
||
"""默认裁切区域为画面底部 1/4（字幕很少出现在上半部分）。"""
|
||
# 数据：模块默认常量。
|
||
# 测试过程与验证结果
|
||
assert DEFAULT_CROP == [0, 0.75, 1, 0.25]
|
||
|
||
|
||
def test_parse_crop_accepts_sequence_and_tuple() -> None:
|
||
"""crop 接受四元素序列（列表/元组），返回浮点比例列表。"""
|
||
# 数据：列表形式与元组形式。
|
||
# 测试过程与验证结果
|
||
assert _parse_crop([0.1, 0.2, 0.3, 0.4]) == [0.1, 0.2, 0.3, 0.4]
|
||
assert _parse_crop((0, 0.75, 1, 0.25)) == [0.0, 0.75, 1.0, 0.25]
|
||
|
||
|
||
def test_parse_crop_rejects_invalid_input() -> None:
|
||
"""非法 crop（长度不对/非数值/越界）返回 None，由调用处报错。"""
|
||
# 数据：五类非法输入（含 JSON 字符串——解析由调用方负责，节点只收序列）。
|
||
# 测试过程与验证结果
|
||
assert _parse_crop("not json") is None
|
||
assert _parse_crop("[0.1, 0.2, 0.3, 0.4]") is None
|
||
assert _parse_crop([0.1, 0.2]) is None
|
||
assert _parse_crop([0.1, 0.2, "x", 0.4]) is None
|
||
assert _parse_crop([0.1, 0.2, 1.5, 0.4]) is None
|
||
|
||
|
||
def test_frame_step_rounds_interval_times_fps() -> None:
|
||
"""帧间隔换算：step = round(间隔秒 × fps)，至少为 1。"""
|
||
# 数据：25fps 下 0.5s → 12.5 → 12；极短间隔至少 1。
|
||
# 测试过程与验证结果
|
||
assert _frame_step(25.0, 0.5) == 12
|
||
assert _frame_step(25.0, 0.04) == 1
|
||
assert _frame_step(30.0, 1.0) == 30
|
||
|
||
|
||
def test_parse_progress_line_extracts_frame_number() -> None:
|
||
"""ffmpeg -progress 的 frame=N 行被解析为整数，其他行忽略。"""
|
||
# 数据：真实的 -progress 输出行。
|
||
# 测试过程与验证结果
|
||
assert _parse_progress_line("frame=123") == 123
|
||
assert _parse_progress_line("fps=25.0") is None
|
||
assert _parse_progress_line("frame=abc") is None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 帧号自然排序（14236 帧事故回归）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_frame_files_sorted_numerically_not_lexicographically(tmp_path: Path) -> None:
|
||
"""帧文件按帧号数值排序：4 位与 5 位编号混排时不会回退（真实事故回归）。
|
||
|
||
背景：ffmpeg 的 %04d 在超过 9999 帧后扩为 5 位，字典序会把
|
||
frame_10000 排到 frame_9999 之前，导致时间轴与图像错位
|
||
（run_339ec7ee437f 的 14236 帧任务实测）。
|
||
"""
|
||
# 数据：构造跨越 9999 边界的真实帧文件名。
|
||
frames_dir = tmp_path / "frames"
|
||
frames_dir.mkdir()
|
||
for number in (1, 9999, 10000, 10009, 1009, 14236):
|
||
(frames_dir / f"frame_{number:04d}.png").write_bytes(b"png")
|
||
|
||
# 测试过程
|
||
ordered = [int(p.stem.split("_")[1]) for p in _sorted_frame_files(frames_dir)]
|
||
|
||
# 验证结果：严格按数值升序（含 1009 < 9999 < 10000 < 10009）。
|
||
assert ordered == [1, 1009, 9999, 10000, 10009, 14236]
|
||
|
||
|
||
def test_frame_files_sorted_returns_empty_for_empty_dir(tmp_path: Path) -> None:
|
||
"""空目录返回空列表。"""
|
||
# 数据：空目录。
|
||
frames_dir = tmp_path / "frames"
|
||
frames_dir.mkdir()
|
||
|
||
# 测试过程与验证结果
|
||
assert _sorted_frame_files(frames_dir) == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# invoke：真实抽帧
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_invoke_extracts_frames_with_manifest(tmp_path: Path) -> None:
|
||
"""真实视频按间隔抽帧：产出帧图与清单，清单时间轴与帧数一致。"""
|
||
# 数据：10 秒 25fps 视频，间隔 2 秒（step=50）。
|
||
assert TEST_VIDEO.is_file(), f"缺少测试视频 {TEST_VIDEO}"
|
||
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, TEST_VIDEO, interval_seconds=2.0))
|
||
|
||
# 验证结果：产物清单存在，帧数与 10s/2s=5 一致，时间轴递增。
|
||
assert response.status == "completed", response.error
|
||
manifest_path = Path(response.outputs["frames_manifest"])
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
frames = manifest["frames"] if isinstance(manifest, dict) else manifest
|
||
assert len(frames) >= 4
|
||
times = [f["time"] for f in frames]
|
||
assert times == sorted(times)
|
||
assert response.outputs["frame_count"] == len(frames)
|
||
|
||
|
||
def _png_size(path: Path) -> tuple[int, int]:
|
||
"""从 PNG 文件头读取宽高（避免为测试引入图像库依赖）。
|
||
|
||
PNG 结构：8 字节签名 + 4 字节长度 + "IHDR" + 宽(4) + 高(4)，均为大端。
|
||
"""
|
||
data = path.read_bytes()[:24]
|
||
assert data[:8] == b"\x89PNG\r\n\x1a\n", "不是合法 PNG"
|
||
assert data[12:16] == b"IHDR"
|
||
return int.from_bytes(data[16:20], "big"), int.from_bytes(data[20:24], "big")
|
||
|
||
|
||
def test_invoke_crops_to_bottom_region(tmp_path: Path) -> None:
|
||
"""默认裁切底部 1/4：帧图高度明显小于原视频（1280x720 → 约 320 高）。"""
|
||
# 数据：真实视频 + 默认 crop。
|
||
if not TEST_VIDEO.is_file():
|
||
pytest.skip(f"缺少测试视频 {TEST_VIDEO}")
|
||
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, TEST_VIDEO, interval_seconds=5.0))
|
||
|
||
# 验证结果：帧图高度约 720*0.25=180（720p 内压缩后可能更小），宽度保持宽扁形。
|
||
assert response.status == "completed", response.error
|
||
frames = sorted((tmp_path / "out" / "frames").glob("frame_*.png"))
|
||
assert frames
|
||
width, height = _png_size(frames[0])
|
||
assert height <= 320
|
||
assert width > height # 底部字幕条，仍是宽扁形
|
||
|
||
|
||
def test_invoke_respects_custom_crop(tmp_path: Path) -> None:
|
||
"""自定义 crop 生效：裁切区域变化体现在帧图尺寸上。"""
|
||
# 数据：取画面上半部分 1/4 高。
|
||
if not TEST_VIDEO.is_file():
|
||
pytest.skip(f"缺少测试视频 {TEST_VIDEO}")
|
||
|
||
# 测试过程
|
||
response = invoke(_request(
|
||
tmp_path, TEST_VIDEO, interval_seconds=5.0, crop=[0, 0, 1, 0.25],
|
||
))
|
||
|
||
# 验证结果：成功产出帧图。
|
||
assert response.status == "completed", response.error
|
||
assert list((tmp_path / "out" / "frames").glob("frame_*.png"))
|
||
|
||
|
||
def test_invoke_fails_without_video_uri(tmp_path: Path) -> None:
|
||
"""缺少 video_uri 时失败。"""
|
||
# 数据：空输入。
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, None))
|
||
|
||
# 验证结果
|
||
assert response.status == "failed"
|
||
assert "video_uri" in (response.error or "")
|
||
|
||
|
||
def test_invoke_fails_when_video_missing(tmp_path: Path) -> None:
|
||
"""视频文件不存在时失败。"""
|
||
# 数据：不存在的路径。
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, tmp_path / "nope.mp4"))
|
||
|
||
# 验证结果
|
||
assert response.status == "failed"
|
||
assert "not found" in (response.error or "")
|
||
|
||
|
||
def test_invoke_rejects_invalid_interval_and_crop(tmp_path: Path) -> None:
|
||
"""非法 interval / crop 参数被拒绝，不产出残缺帧序列。"""
|
||
# 数据：零间隔与非法 crop。
|
||
# 测试过程与验证结果
|
||
assert invoke(_request(tmp_path, TEST_VIDEO, interval_seconds=0)).status == "failed"
|
||
assert invoke(_request(tmp_path, TEST_VIDEO, crop="bad")).status == "failed"
|