"""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} 不应触发转码"