"""HEVC 自动转码逻辑单元测试 (ingest.py) 测试覆盖（全部调用生产代码真实函数）： - HEVC 编码检测（is_hevc_codec） - ffprobe 旋转探测（probe_rotation）：单行 side_data 读取，不再取错 stream_tags 行 - 竖屏判定（is_portrait_rotation） - 转码滤镜构建（build_transcode_vf）：逗号 \\, 转义、无 transpose（避免与 ffmpeg 内置 autorotate 双重旋转）、竖屏按宽/横屏按高的 1080p 只缩不放 - 转码产物方向/维度校验（validate_transcode_output） - ffmpeg 端到端（有 ffmpeg 时）：竖屏 r90/r270 → 1080x1920 h264 无 side data； 横屏 → 1920x1080；缩略图方向为竖版 """ from __future__ import annotations import shutil import subprocess import sys from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker")) from worker_app.tasks.ingest import ( # noqa: E402 HEVC_CODECS, build_transcode_vf, is_hevc_codec, is_portrait_rotation, is_portrait_video, probe_dimensions, probe_rotation, probe_video_info, validate_transcode_output, ) # ── 纯逻辑测试（不依赖 ffmpeg）────────────────────────────────────────────── class TestHEVCCodecDetection: def test_hevc_keywords_detected(self): for codec in ("hevc", "h265", "hvh1", "HEVC", "H265", "Hevc", "HVH1"): assert is_hevc_codec(codec), f"{codec} 应该被识别为 HEVC" def test_non_hevc_codecs_not_detected(self): for codec in ("h264", "avc1", "vp9", "av1", "mpeg4", "", None): assert not is_hevc_codec(codec), f"{codec} 不应被识别为 HEVC" def test_hevc_keywords_constant(self): assert HEVC_CODECS == ("hevc", "h265", "hvh1") class TestPortraitRotation: def test_portrait_rotations(self): for rotation in (90, 270, -90): assert is_portrait_rotation(rotation), f"rotation={rotation} 应为竖屏" def test_non_portrait_rotations(self): for rotation in (0, 180, -180, None): assert not is_portrait_rotation(rotation), f"rotation={rotation} 不应判定为竖屏" class TestIsPortraitVideo: """按显示方向判定竖屏（存储维度 + rotation 互换）。""" def test_ios_style_stored_landscape_with_rotation90(self): # iOS：存储 1920x1080 + rotation=90 → 显示 1080x1920 竖屏 assert is_portrait_video(1920, 1080, 90) is True assert is_portrait_video(1920, 1080, -90) is True assert is_portrait_video(1920, 1080, 270) is True def test_physical_portrait_no_rotation(self): # Android/物理竖屏：存储 1080x1920、无 rotation → 竖屏（旧逻辑误判横屏） assert is_portrait_video(1080, 1920, None) is True assert is_portrait_video(1080, 1920, 0) is True def test_landscape_normal(self): assert is_portrait_video(1920, 1080, None) is False assert is_portrait_video(1920, 1080, 0) is False def test_rotation180_no_swap(self): # 180 度不互换宽高 assert is_portrait_video(1920, 1080, 180) is False assert is_portrait_video(1080, 1920, 180) is True def test_dimensions_unknown_falls_back_to_rotation(self): # 探测失败（None）退回仅看 rotation，不抛异常 assert is_portrait_video(None, None, 90) is True assert is_portrait_video(None, None, None) is False assert is_portrait_video(0, 0, 270) is True class TestBuildTranscodeVF: """统一转码滤镜：长边封顶 1920、只缩不放、方向无关。""" def test_comma_escaped_with_backslash(self): r"""scale 表达式内的逗号必须 \, 转义（否则报 Invalid size / No such filter）。""" vf = build_transcode_vf() assert "\\," in vf # 不应存在未转义的裸逗号（filter 分隔）出现在 if 表达式内 assert "gte(iw,ih)" not in vf assert "gt(ih,iw)" not in vf assert "min(1920,iw)" not in vf def test_no_transpose_filter(self): """不能显式 transpose：ffmpeg autorotate 已按 side data 物理旋转， 再加 transpose 会双重旋转把竖屏转成横屏。""" assert "transpose" not in build_transcode_vf() def test_long_edge_capped_1920_orientation_agnostic(self): """横屏限宽、竖屏限高，均 min(1920,...)，短边 -2 自适应。""" vf = build_transcode_vf() assert "gte(iw" in vf and "gt(ih" in vf, "横/竖分支都要在" assert vf.count("min(1920") == 2, "宽高分支都按长边 1920 封顶" def test_format_yuv420p_suffix(self): assert build_transcode_vf().endswith(",format=yuv420p") class TestProbeVideoInfo: """probe_video_info 合并探测：维度 + rotation 一次 ffprobe。""" def test_merges_dimensions_and_rotation(self, tmp_path): """真实 ffprobe：r90 素材 → (1920, 1080, 90)；r270 → rotation=-90；横屏 None。""" # 端到端素材由 hevc_sources fixture 构造，这里用独立小素材验证合并函数 base = tmp_path / "v.mp4" subprocess.run( [ FFMPEG, "-y", "-hide_banner", "-loglevel", "error", "-f", "lavfi", "-i", "color=c=white:s=320x240:d=1:r=15", "-c:v", "libx264", "-pix_fmt", "yuv420p", str(base), ], check=True, ) w, h, rot = probe_video_info(str(base)) assert (w, h) == (320, 240) assert rot is None def test_probe_failure_returns_triple_none(self): w, h, rot = probe_video_info("/nonexistent/path/fake.mp4") assert (w, h, rot) == (None, None, None) class TestValidateTranscodeOutput: def _probe_dimensions_called_with(self, monkeypatch, w, h, rotation=None): monkeypatch.setattr("worker_app.tasks.ingest.probe_dimensions", lambda p: (w, h)) monkeypatch.setattr("worker_app.tasks.ingest.probe_rotation", lambda p: rotation) def test_portrait_output_ok(self, monkeypatch): self._probe_dimensions_called_with(monkeypatch, 1080, 1920, None) assert validate_transcode_output("/tmp/fake.mp4", expected_portrait=True) def test_landscape_output_ok(self, monkeypatch): self._probe_dimensions_called_with(monkeypatch, 1920, 1080, None) assert validate_transcode_output("/tmp/fake.mp4", expected_portrait=False) def test_portrait_source_but_landscape_output_rejected(self, monkeypatch): """竖屏源转出横屏（双重旋转 bug 产物）必须判失败降级。""" self._probe_dimensions_called_with(monkeypatch, 1920, 1080, None) assert not validate_transcode_output("/tmp/fake.mp4", expected_portrait=True) def test_landscape_source_but_portrait_output_rejected(self, monkeypatch): self._probe_dimensions_called_with(monkeypatch, 1080, 1920, None) assert not validate_transcode_output("/tmp/fake.mp4", expected_portrait=False) def test_residual_rotation_side_data_rejected(self, monkeypatch): """产物仍带 rotation side data 会被播放器二次旋转，必须判失败。""" self._probe_dimensions_called_with(monkeypatch, 1080, 1920, rotation=90) assert not validate_transcode_output("/tmp/fake.mp4", expected_portrait=True) def test_long_edge_beyond_1920_rejected(self, monkeypatch): """只缩不放：长边不得超过 1920。""" self._probe_dimensions_called_with(monkeypatch, 3840, 2160, None) assert not validate_transcode_output("/tmp/fake.mp4", expected_portrait=False) def test_missing_dimensions_rejected(self, monkeypatch): self._probe_dimensions_called_with(monkeypatch, None, None, None) assert not validate_transcode_output("/tmp/fake.mp4", expected_portrait=False) # ── ffmpeg 端到端测试（无 ffmpeg/libx265 环境自动跳过）────────────────────── FFMPEG = shutil.which("ffmpeg") FFPROBE = shutil.which("ffprobe") def _has_x265() -> bool: if not FFMPEG: return False out = subprocess.run([FFMPEG, "-hide_banner", "-encoders"], capture_output=True, text=True).stdout return "libx265" in out pytestmark = pytest.mark.skipif( not (FFMPEG and FFPROBE and _has_x265()), reason="ffmpeg/ffprobe/libx265 不可用，跳过端到端转码测试", ) # 字体候选路径（Debian/Ubuntu/Alpine/macOS），找不到则省略 drawtext， # 仅靠顶部红条表达方向，测试断言不依赖文字。 _FONT_CANDIDATES = ( "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", "/usr/share/fonts/dejavu/DejaVuSans-Bold.ttf", "/usr/share/fonts/TTF/DejaVuSans-Bold.ttf", "/System/Library/Fonts/Supplemental/Arial Bold.ttf", ) FONT = next((f for f in _FONT_CANDIDATES if Path(f).exists()), None) @pytest.fixture def hevc_sources(tmp_path): """构造带方向标记的 HEVC 测试素材（横屏存储 + rotation side data，模拟 iPhone）。""" base = tmp_path / "base_landscape_h264.mp4" # 顶部红条标记画面方向；有字体时叠加 TOP 文字（仅人眼校验用，断言不依赖） draw = "drawbox=x=0:y=0:w=1920:h=200:color=red:t=fill" if FONT: draw += f",drawtext=fontfile={FONT}:text='TOP':fontsize=160:fontcolor=black:" "x=(w-tw)/2:y=30" subprocess.run( [ FFMPEG, "-y", "-hide_banner", "-loglevel", "error", "-f", "lavfi", "-i", "color=c=white:s=1920x1080:d=2:r=30", "-vf", draw, "-c:v", "libx264", "-pix_fmt", "yuv420p", str(base), ], check=True, ) base_hevc = tmp_path / "base_landscape_hevc.mp4" subprocess.run( [ FFMPEG, "-y", "-hide_banner", "-loglevel", "error", "-i", str(base), "-c:v", "libx265", "-tag:v", "hvc1", "-pix_fmt", "yuv420p", "-an", str(base_hevc), ], check=True, ) def tag_rotate(src: Path, rotation: int) -> Path: """给横屏 HEVC 素材打上旋转 display matrix，模拟 iPhone 竖屏。 ffmpeg 版本差异： - 5.1+（含 CI 的 7.x）：mp4 muxer 支持 -display_rotation 输出选项， 老式 -metadata rotate 在 stream copy 时不再写入 display matrix； - 4.x：不识别 -display_rotation，仍用 -metadata:s:v:0 rotate=。 两条路径都试，造完用 probe_rotation 自检；都失败则返回 None， 消费方跳过依赖旋转信息的断言（skip）。 """ out = tmp_path / f"portrait_r{rotation}_hevc.mp4" attempts = [ # ffmpeg 5.1+: display_rotation 输出选项 [ FFMPEG, "-y", "-hide_banner", "-loglevel", "error", "-i", str(src), "-c", "copy", "-display_rotation", str(rotation), str(out), ], # ffmpeg 4.x: 老式 rotate metadata [ FFMPEG, "-y", "-hide_banner", "-loglevel", "error", "-i", str(src), "-c", "copy", "-metadata:s:v:0", f"rotate={rotation}", str(out), ], ] for cmd in attempts: proc = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True) if proc.returncode == 0 and out.exists() and probe_rotation(str(out)) is not None: return out return None # 当前 ffmpeg 无法造出带 display matrix 的素材 r90 = tag_rotate(base_hevc, 90) r270 = tag_rotate(base_hevc, 270) # 物理竖屏素材（Android 风格）：直接生成 1080x1920 HEVC，无 rotation side data。 # 红条画在存储帧的顶部（短边 1080 一侧），方向断言只看维度。 physical = tmp_path / "portrait_physical_hevc.mp4" phys_draw = "drawbox=x=0:y=0:w=1080:h=120:color=red:t=fill" if FONT: phys_draw += f",drawtext=fontfile={FONT}:text='TOP':fontsize=120:fontcolor=black:" "x=(w-tw)/2:y=20" subprocess.run( [ FFMPEG, "-y", "-hide_banner", "-loglevel", "error", "-f", "lavfi", "-i", "color=c=white:s=1080x1920:d=2:r=30", "-vf", phys_draw, "-c:v", "libx265", "-tag:v", "hvc1", "-pix_fmt", "yuv420p", "-an", str(physical), ], check=True, ) # 超宽屏 HEVC（4000x1000，无 rotation）：旧滤镜短边 1000<1080 不缩放， # 长边 4000 超 validate 的 1920 上限被误降级；新滤镜长边封顶应缩到 1920x480。 ultra_wide = tmp_path / "ultra_wide_hevc.mp4" subprocess.run( [ FFMPEG, "-y", "-hide_banner", "-loglevel", "error", "-f", "lavfi", "-i", "color=c=red:s=4000x1000:d=1:r=30", "-c:v", "libx265", "-tag:v", "hvc1", "-pix_fmt", "yuv420p", "-an", str(ultra_wide), ], check=True, ) yield { "portrait_r90": r90, "portrait_r270": r270, "portrait_physical": physical, "ultra_wide": ultra_wide, "landscape": base_hevc, "tmp_path": tmp_path, } def _transcode_like_production(src: Path, dst: Path) -> tuple[bool, int | None]: """按生产代码相同方式执行转码，返回 (is_portrait, rotation)。""" # 与生产一致：合并探测 + 统一滤镜 width, height, rotation = probe_video_info(str(src)) is_portrait = is_portrait_video(width, height, rotation) vf = build_transcode_vf() subprocess.run( [ FFMPEG, "-y", "-hide_banner", "-loglevel", "error", "-i", str(src), "-c:v", "libx264", "-preset", "fast", "-crf", "18", "-vf", vf, "-pix_fmt", "yuv420p", "-an", "-movflags", "+faststart", str(dst), ], check=True, ) return is_portrait, rotation class TestTranscodeEndToEnd: def test_portrait_r90_hevc(self, hevc_sources): """竖屏 HEVC(rotation=90) 转码后：h264、1080x1920（h>w）、无 rotation side data。""" src = hevc_sources["portrait_r90"] if src is None: pytest.skip("当前 ffmpeg 无法生成带 rotation display matrix 的测试素材") dst = hevc_sources["tmp_path"] / "out_r90.mp4" is_portrait, rotation = _transcode_like_production(src, dst) assert rotation == 90 assert is_portrait is True width, height = probe_dimensions(str(dst)) assert (width, height) == (1080, 1920) assert height > width assert probe_rotation(str(dst)) is None assert validate_transcode_output(str(dst), True) is True codec = subprocess.run( [ FFPROBE, "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=codec_name", "-of", "default=noprint_wrappers=1:nokey=1", str(dst), ], capture_output=True, text=True, check=True, ).stdout.strip() assert codec == "h264" def test_portrait_r270_hevc(self, hevc_sources): """竖屏 HEVC(rotation=270→side_data 显示 -90) 同样转为 1080x1920。""" src = hevc_sources["portrait_r270"] if src is None: pytest.skip("当前 ffmpeg 无法生成带 rotation display matrix 的测试素材") dst = hevc_sources["tmp_path"] / "out_r270.mp4" is_portrait, rotation = _transcode_like_production(src, dst) assert rotation == -90 # ffprobe side_data 规范化为 -90 assert is_portrait is True width, height = probe_dimensions(str(dst)) assert (width, height) == (1080, 1920) assert height > width assert probe_rotation(str(dst)) is None assert validate_transcode_output(str(dst), True) is True def test_physical_portrait_hevc_no_rotation(self, hevc_sources): """物理竖屏 HEVC（存储 1080x1920、无 rotation side data，Android 风格） → 必须识别为竖屏，转出 1080x1920 h264；旧逻辑只看 rotation 会误判横屏、 套横屏滤镜压成 608x1080 并被 validate 拦截降级，用户拿到不能播放的 HEVC。""" src = hevc_sources["portrait_physical"] dst = hevc_sources["tmp_path"] / "out_physical.mp4" is_portrait, rotation = _transcode_like_production(src, dst) assert rotation is None assert is_portrait is True width, height = probe_dimensions(str(dst)) assert (width, height) == (1080, 1920) assert height > width assert probe_rotation(str(dst)) is None assert validate_transcode_output(str(dst), True) is True def test_landscape_hevc(self, hevc_sources): """横屏 HEVC 转码后：h264、1920x1080（w>h）、无 rotation side data。""" src = hevc_sources["landscape"] dst = hevc_sources["tmp_path"] / "out_land.mp4" is_portrait, rotation = _transcode_like_production(src, dst) assert rotation is None assert is_portrait is False width, height = probe_dimensions(str(dst)) assert (width, height) == (1920, 1080) assert width > height assert probe_rotation(str(dst)) is None assert validate_transcode_output(str(dst), False) is True def test_ultra_wide_long_edge_capped(self, hevc_sources): """超宽屏 4000x1000：长边必须封顶 1920（→1920x480），validate 通过。 回归旧滤镜只按短边触发缩放、长边超 1920 被误降级的 bug。""" src = hevc_sources["ultra_wide"] dst = hevc_sources["tmp_path"] / "out_ultra.mp4" is_portrait, rotation = _transcode_like_production(src, dst) assert rotation is None assert is_portrait is False width, height = probe_dimensions(str(dst)) assert max(width, height) <= 1920, f"长边应封顶 1920，实际 {width}x{height}" assert width == 1920 and height == 480 assert validate_transcode_output(str(dst), False) is True def test_thumbnail_portrait_direction(self, hevc_sources): """缩略图（thumbnail_generator 同款 ffmpeg 抽帧，依赖 autorotate）竖屏源→竖版图。""" src = hevc_sources["portrait_r90"] if src is None: pytest.skip("当前 ffmpeg 无法生成带 rotation display matrix 的测试素材") thumb = hevc_sources["tmp_path"] / "thumb.jpg" subprocess.run( [ FFMPEG, "-y", "-hide_banner", "-loglevel", "error", "-ss", "0.1", "-i", str(src), "-vframes", "1", "-vf", "scale=640:-1:force_original_aspect_ratio=decrease,format=yuvj420p", "-q:v", "2", str(thumb), ], check=True, ) width, height = probe_dimensions(str(thumb)) assert height > width, f"竖屏缩略图应为竖版，实际 {width}x{height}" def test_probe_rotation_reads_side_data_not_tag_line(self, hevc_sources): """回归：旧实现同时请求 side_data+stream_tags 输出两行且取第一行， r90 文件会取到 "270"；新实现只读 side_data，r90→90、r270→-90。""" r90, r270 = hevc_sources["portrait_r90"], hevc_sources["portrait_r270"] if r90 is None or r270 is None: pytest.skip("当前 ffmpeg 无法生成带 rotation display matrix 的测试素材") assert probe_rotation(str(r90)) == 90 assert probe_rotation(str(r270)) == -90 assert probe_rotation(str(hevc_sources["landscape"])) is None