xiaoxia-saas/tests/unit/test_ingest_hevc_transcode.py

"""HEVC 自动转码逻辑单元测试 (ingest.py)

测试覆盖：
- HEVC 编码检测逻辑
- 转码后文件命名规则
- 元数据提取失败时的脏数据防护
- FFmpeg 超时/错误降级策略
- 安全修复（tempfile、subprocess）
- Scale filter 逻辑
"""

from __future__ import annotations

import subprocess
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest


class TestHEVCAutoTranscode:
    """测试 ingest_asset 中的 HEVC 自动转码逻辑"""

    def test_hevc_detection_keywords(self):
        """验证 HEVC 编码的所有关键词"""
        hevc_keywords = ("hevc", "h265", "hvh1")

        assert "hevc" in hevc_keywords
        assert "h265" in hevc_keywords
        assert "hvh1" in hevc_keywords
        assert "h264" not in hevc_keywords
        assert "avc1" not in hevc_keywords

    def test_h264_not_detected_as_hevc(self):
        """H.264 视频不应触发转码"""
        codec = "h264"
        hevc_keywords = ("hevc", "h265", "hvh1")
        assert codec not in hevc_keywords, "H.264 不应触发转码"

    def test_transcode_storage_key_naming(self):
        """验证转码后文件命名规则"""
        original_key = "uploads/video_123/test.mp4"
        p = Path(original_key)
        new_key = str(p.parent / (p.stem + "_h264" + p.suffix))

        assert new_key == "uploads/video_123/test_h264.mp4"

    def test_transcode_storage_key_naming_complex_path(self):
        """验证复杂路径的命名规则"""
        original_key = "uploads/2026/08/20/abc123/video_4k.mov"
        p = Path(original_key)
        new_key = str(p.parent / (p.stem + "_h264" + p.suffix))

        assert new_key == "uploads/2026/08/20/abc123/video_4k_h264.mov"

    def test_metadata_failure_no_dirty_data(self):
        """验证元数据提取失败时不更新 storage_key（避免脏数据）

        这是 AI Code Review 发现的 BUG 修复：
        - 旧逻辑：先更新 storage_key，再提取元数据 → 可能产生脏数据
        - 新逻辑：先提取元数据，确认成功后再更新 storage_key
        """
        original_storage_key = "uploads/test/video.mp4"
        new_storage_key = "uploads/test/video_h264.mp4"

        # 初始状态
        job_storage_key = original_storage_key
        metadata = {"codec": "hevc", "width": 3840, "height": 2160}

        # 模拟转码成功
        transcode_success = True

        # 模拟元数据提取失败
        new_metadata = {}
        new_extract_success = False

        # 修复后的逻辑：先提取元数据，确认成功后再更新
        if transcode_success:
            if new_extract_success:
                job_storage_key = new_storage_key
                metadata = new_metadata
            # 如果元数据提取失败，不更新 job_storage_key

        # 验证：storage_key 保持原值，没有脏数据
        assert job_storage_key == original_storage_key
        assert metadata["codec"] == "hevc"  # 保持原始元数据

    def test_metadata_success_updates_storage_key(self):
        """验证元数据提取成功时正确更新 storage_key"""
        original_storage_key = "uploads/test/video.mp4"
        new_storage_key = "uploads/test/video_h264.mp4"

        job_storage_key = original_storage_key
        metadata = {"codec": "hevc", "width": 3840, "height": 2160}

        # 模拟转码成功
        transcode_success = True

        # 模拟元数据提取成功
        new_metadata = {"codec": "h264", "width": 1920, "height": 1080}
        new_extract_success = True

        # 修复后的逻辑
        if transcode_success:
            if new_extract_success:
                job_storage_key = new_storage_key
                metadata = new_metadata

        # 验证：storage_key 和 metadata 都更新为新值
        assert job_storage_key == new_storage_key
        assert metadata["codec"] == "h264"
        assert metadata["width"] == 1920

    @patch("subprocess.run")
    def test_ffmpeg_timeout_degradation(self, mock_subprocess):
        """验证 FFmpeg 超时降级使用原始文件"""
        mock_subprocess.side_effect = subprocess.TimeoutExpired(cmd="ffmpeg", timeout=300)

        # 模拟降级逻辑
        transcode_success = False
        try:
            raise subprocess.TimeoutExpired(cmd="ffmpeg", timeout=300)
        except subprocess.TimeoutExpired:
            transcode_success = False

        assert not transcode_success, "超时应该导致转码失败"

    @patch("subprocess.run")
    def test_ffmpeg_error_degradation(self, mock_subprocess):
        """验证 FFmpeg 执行失败降级使用原始文件"""
        mock_subprocess.return_value = MagicMock(
            returncode=1,
            stderr="Error: Invalid data found when processing input",
        )

        result = mock_subprocess.return_value
        transcode_success = result.returncode == 0

        assert not transcode_success, "FFmpeg 返回非零退出码应该导致转码失败"

    def test_scale_filter_logic_4k_video(self):
        """验证 4K 视频会被缩放到 1080p"""
        ih = 2160
        should_scale = ih > 1080
        assert should_scale, "4K 视频应该被缩放"

    def test_scale_filter_logic_1080p_video(self):
        """验证 1080p 视频不会被缩放"""
        ih = 1080
        should_scale = ih > 1080
        assert not should_scale, "1080p 视频不应该被缩放"

    def test_scale_filter_logic_720p_video(self):
        """验证 720p 视频不会被缩放"""
        ih = 720
        should_scale = ih > 1080
        assert not should_scale, "720p 视频不应该被缩放"

    def test_tempfile_security_fix(self):
        """验证使用 NamedTemporaryFile 替代 mktemp（安全修复）

        AI Code Review 发现的安全漏洞：
        - tempfile.mktemp 存在 TOCTOU 竞态条件
        - 应该使用 NamedTemporaryFile(delete=False)
        """
        import tempfile

        with patch("tempfile.NamedTemporaryFile") as mock_ntf:
            mock_file = MagicMock()
            mock_file.name = "/tmp/test_h264.mp4"
            mock_ntf.return_value = mock_file

            # 新代码的调用方式
            _tc_tmp_file = tempfile.NamedTemporaryFile(delete=False, suffix="_h264.mp4")
            _tc_tmp = Path(_tc_tmp_file.name)
            _tc_tmp_file.close()

            # 验证使用了 NamedTemporaryFile
            mock_ntf.assert_called_once_with(delete=False, suffix="_h264.mp4")

    def test_subprocess_output_handling(self):
        """验证 subprocess 输出处理（避免内存溢出）

        AI Code Review 发现的稳定性风险：
        - capture_output=True 会将所有输出加载到内存
        - 应该使用 stdout=DEVNULL, stderr=PIPE
        """
        import subprocess as sp

        with patch("subprocess.run") as mock_run:
            mock_run.return_value = MagicMock(returncode=0)

            # 新代码的调用方式
            sp.run(
                ["ffmpeg", "-i", "input.mp4", "output.mp4"],
                stdout=sp.DEVNULL,
                stderr=sp.PIPE,
                text=True,
                timeout=300,
            )

            # 验证使用了 stdout=DEVNULL, stderr=PIPE
            call_kwargs = mock_run.call_args[1]
            assert call_kwargs.get("stdout") == sp.DEVNULL
            assert call_kwargs.get("stderr") == sp.PIPE
            assert call_kwargs.get("timeout") == 300

    def test_ffmpeg_command_parameters(self):
        """验证 FFmpeg 命令参数正确性"""
        expected_params = [
            "-c:v",
            "libx264",
            "-preset",
            "fast",
            "-crf",
            "18",
            "-pix_fmt",
            "yuv420p",
            "-c:a",
            "aac",
            "-b:a",
            "128k",
            "-movflags",
            "+faststart",
        ]

        # 验证所有关键参数都在命令中
        cmd = ["ffmpeg", "-y", "-i", "input.mp4"]
        cmd.extend(expected_params)
        cmd.append("output.mp4")

        assert "-c:v" in cmd
        assert "libx264" in cmd
        assert "-crf" in cmd
        assert "18" in cmd
        assert "-pix_fmt" in cmd
        assert "yuv420p" in cmd
        assert "-movflags" in cmd
        assert "+faststart" in cmd

    def test_hevc_codec_case_insensitive(self):
        """验证 HEVC 检测不区分大小写"""
        test_cases = ["hevc", "HEVC", "Hevc", "h265", "H265", "hvh1", "HVH1"]
        hevc_keywords = ("hevc", "h265", "hvh1")

        for codec in test_cases:
            assert codec.lower() in hevc_keywords, f"{codec} 应该被检测为 HEVC"

    def test_non_hevc_codecs(self):
        """验证非 HEVC 编码不会触发转码"""
        non_hevc_codecs = ["h264", "avc1", "vp9", "av1", "mpeg4", ""]
        hevc_keywords = ("hevc", "h265", "hvh1")

        for codec in non_hevc_codecs:
            assert codec.lower() not in hevc_keywords, f"{codec} 不应触发转码"