xiaoxia-saas/tests/unit/test_ingest_hevc_transcode.py

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