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