diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py index 46c7d3394..c9ac3f79e 100755 --- a/apps/worker/worker_app/tasks/ingest.py +++ b/apps/worker/worker_app/tasks/ingest.py @@ -1,3 +1,4 @@ +import json import shutil import subprocess import tempfile @@ -150,6 +151,214 @@ def extract_media_metadata(file_url: str, media_type: str) -> tuple[dict, bool]: return metadata, success +# ── HEVC 自动转码辅助函数（模块级，便于单元测试）───────────────────────── +HEVC_CODECS = ("hevc", "h265", "hvh1") +# 转码目标：长边封顶 1920（只缩不放，与 validate 的 max_long_edge 一致）， +# 竖屏/横屏/超宽屏统一按长边等比缩放，短边自动按比例（-2 保证偶数）。 +TRANSCODE_MAX_LONG_EDGE = 1920 +TRANSCODE_TIMEOUT_SECONDS = 900 +# ffmpeg scale 滤镜中 if(...) 表达式内的逗号必须用 \, 转义， +# 否则逗号被当作 filter 分隔符解析，报 "No such filter" / Invalid size。 +# subprocess list 传参不经 shell，\ 在 Python 字符串里直接写一个字面反斜杠即可。 +# 横屏（iw>=ih）限宽 min(1920,iw)、高 -2 自适应；竖屏（ih>iw）限高、宽自适应； +# min() 保证小视频不放大。与 validate_transcode_output 的"长边 <= 1920"规则对齐， +# 超宽屏（如 4000x1000）短边不触发旧的短边缩放、长边超限被误降级的问题由此消除。 +_TRANSCODE_VF = ( + rf"scale=w=if(gte(iw\,ih)\,min({TRANSCODE_MAX_LONG_EDGE}\,iw)\,-2):" + rf"h=if(gt(ih\,iw)\,min({TRANSCODE_MAX_LONG_EDGE}\,ih)\,-2),format=yuv420p" +) + + +def is_hevc_codec(codec: str | None) -> bool: + """判断编码是否为 HEVC（不区分大小写）。""" + return (codec or "").lower() in HEVC_CODECS + + +def probe_rotation(path: str) -> int | None: + """ffprobe 读取视频旋转角度（display matrix side data）。 + + 返回 0/90/-90/180 等整数；无 side data 或探测失败返回 None。 + + 注意：旧实现同时请求 side_data 和 stream_tags 且取输出第一行， + iOS 文件会输出两行（如 "270\\n90"）导致取到错误值，现仅读 side_data。 + """ + try: + result = subprocess.run( + [ + "ffprobe", + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "side_data=rotation", + "-of", + "default=noprint_wrappers=1:nokey=1", + str(path), + ], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + timeout=60, + ) + first_line = (result.stdout or "").strip().split("\n")[0].strip() + if not first_line: + return None + return int(float(first_line)) + except (subprocess.TimeoutExpired, ValueError, OSError): + return None + + +def is_portrait_rotation(rotation: int | None) -> bool: + """rotation side data 为 ±90/270 时表示竖屏拍摄。 + + 注意：这只覆盖"存储横屏 + display matrix 旋转"的 iOS 风格视频； + 物理竖屏视频（Android 常见，存储即 h>w、rotation=None/0）不会命中， + 方向判定请用 is_portrait_video()。 + """ + return rotation in (90, 270, -90) + + +def is_portrait_video( + stored_width: int | None, + stored_height: int | None, + rotation: int | None, +) -> bool: + """按显示方向判断是否竖屏（显示高度 > 显示宽度）。 + + - rotation 为 90/270/-90 时，显示方向的宽高相对存储维度互换； + - rotation 为 0/180/None 时，显示方向即存储维度。 + + 这样两类竖屏都能正确识别： + - iOS：存储 1920x1080 + rotation=90 → 显示 1080x1920 竖屏 + - Android/物理竖屏：存储 1080x1920、无 rotation → 显示 1080x1920 竖屏 + 探测失败（维度为 None）时退回仅看 rotation，保证调用链不中断。 + """ + if not stored_width or not stored_height: + return is_portrait_rotation(rotation) + if is_portrait_rotation(rotation): + return stored_width > stored_height + return stored_height > stored_width + + +def build_transcode_vf() -> str: + """构建转码视频滤镜（竖屏/横屏统一，按显示长边封顶 1920、只缩不放）。 + + 依赖 ffmpeg 内置 autorotate（默认开启）按 display matrix 物理旋转画面， + 输出自动剥离 rotation side data；滤镜只做等比缩放，方向无关： + 横屏限宽、竖屏限高，短边 -2 自适应偶数，min() 保证小视频不放大。 + + 旧实现的问题： + - 显式 transpose=1 与 autorotate 叠加，竖屏被二次旋转成横屏； + - 竖屏沿用按高缩放表达式，1080x1920 被错误缩成 608x1080； + - 仅按短边 1080 触发缩放，超宽屏（如 4000x1000）长边超 1920 会被 + validate 拦截误降级，用户拿到浏览器无法播放的 HEVC 原文件。 + """ + return _TRANSCODE_VF + + +def probe_dimensions(path: str) -> tuple[int | None, int | None]: + """ffprobe 读取视频宽高（像素维度）。""" + try: + result = subprocess.run( + [ + "ffprobe", + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "stream=width,height", + "-of", + "csv=p=0:s=x", + str(path), + ], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + timeout=60, + ) + text = (result.stdout or "").strip().split("\n")[0].strip() + width_str, height_str = text.split("x") + return int(width_str), int(height_str) + except (subprocess.TimeoutExpired, ValueError, OSError): + return None, None + + +def probe_video_info(path: str) -> tuple[int | None, int | None, int | None]: + """一次 ffprobe 同时读取视频宽高与旋转角度（display matrix side data）。 + + 返回 (width, height, rotation)；探测失败对应位置为 None。 + 合并维度/角度两次探测，减少大文件、高并发下的 ffprobe 进程开销。 + rotation 仅取 stream side_data_list 的 Display Matrix（不读 tags.rotate， + 避免 iOS 文件 tag 值与 side data 双来源取错）。用 -show_streams 全量 JSON + 输出解析，兼容 ffmpeg 4.x/7.x（show_entries 嵌套 section 名跨版本不一致）。 + """ + try: + result = subprocess.run( + [ + "ffprobe", + "-v", + "error", + "-select_streams", + "v:0", + "-show_streams", + "-of", + "json", + str(path), + ], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + timeout=60, + ) + data = json.loads(result.stdout or "{}") + streams = data.get("streams") or [] + if not streams: + return None, None, None + stream = streams[0] + width = int(stream["width"]) if stream.get("width") else None + height = int(stream["height"]) if stream.get("height") else None + rotation = None + for side in stream.get("side_data_list") or []: + if side.get("side_data_type") == "Display Matrix" and side.get("rotation") is not None: + deg = int(round(float(side["rotation"]))) % 360 + # ffprobe：顺时针 90 拍摄输出 90，逆时针 90 输出 -90（归一为 270） + rotation = {0: 0, 90: 90, 180: 180, 270: -90}.get(deg, deg if deg in (90, 180) else None) + break + return width, height, rotation + except (subprocess.TimeoutExpired, ValueError, OSError, json.JSONDecodeError, KeyError, TypeError): + return None, None, None + + +def validate_transcode_output( + output_path: str, + expected_portrait: bool, + max_long_edge: int = TRANSCODE_MAX_LONG_EDGE, +) -> bool: + """校验转码产物方向与维度。 + + - 竖屏源：产物必须 height > width，且仍有 rotation side data 视为失败 + （播放器会二次旋转成横屏） + - 横屏源：产物必须 width >= height + - 长边不得超过 max_long_edge（只缩不放） + 校验失败时调用方应降级使用原始文件，不允许产出方向错误的文件覆盖。 + """ + width, height = probe_dimensions(output_path) + if not width or not height: + return False + if expected_portrait and height <= width: + return False + if not expected_portrait and width < height: + return False + if max(width, height) > max_long_edge: + return False + # 产物仍带 rotation side data 说明方向没有物理固化，播放器会再次旋转 + if probe_rotation(output_path) is not None: + return False + return True + + @celery_app.task(name="worker.ingest_asset") def ingest_asset(job_id: str) -> dict: """ @@ -245,15 +454,13 @@ def ingest_asset(job_id: str) -> dict: # 浏览器 WebCodecs 硬件解码 HEVC 输出黑帧，上传时自动转码 # 失败时降级使用原始文件，不阻塞上传流程 if media_type == "video" and local_file and local_file.exists(): - codec = (metadata.get("codec") or "").lower() - if codec in ("hevc", "h265", "hvh1"): + if is_hevc_codec(metadata.get("codec")): logger.info( "检测到 HEVC 编码 (codec=%s)，启动转码: job_id=%s", - codec, + metadata.get("codec"), job_id, ) _tc_tmp = None - _needs_rotation = False # ── Step 1: 磁盘空间检查（独立 try/except，失败仍尝试转码）── try: @@ -264,50 +471,21 @@ def ingest_asset(job_id: str) -> dict: except Exception as _disk_err: logger.warning("磁盘检查失败，仍尝试转码: job_id=%s err=%s", job_id, _disk_err) - # ── Step 2: ffprobe 旋转检测（独立 try/except，失败不阻塞转码）── - try: - _probe_cmd = [ - "ffprobe", - "-v", - "error", - "-select_streams", - "v:0", - "-show_entries", - "side_data=rotation", - "-show_entries", - "stream_tags=rotate", - "-of", - "default=noprint_wrappers=1:nokey=1", - str(local_file), - ] - _probe_result = subprocess.run( - _probe_cmd, - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, - text=True, - timeout=60, # 大文件在容器 overlay 文件系统上解析可能较慢 - ) - _rotation_str = (_probe_result.stdout or "").strip().split("\n")[0] - if _rotation_str in ("90", "270", "-90"): - _needs_rotation = True - logger.info( - "检测到竖屏视频 (rotation=%s)，将物理旋转画面: job_id=%s", - _rotation_str, - job_id, - ) - except subprocess.TimeoutExpired: - logger.warning( - "ffprobe 旋转检测超时(60s)，跳过旋转继续转码: job_id=%s", - job_id, - ) - _needs_rotation = False - except Exception as _probe_err: - logger.warning( - "ffprobe 旋转检测异常，跳过旋转继续转码: job_id=%s err=%s", - job_id, - _probe_err, - ) - _needs_rotation = False + # ── Step 2: 方向检测（按显示方向判定竖/横屏）────────────── + # 不能只看 rotation side data：Android 等设备的物理竖屏视频 + # 存储维度已是 h>w 且 rotation=0/None，只看 rotation 会误判横屏、 + # 套用横屏滤镜把 1080x1920 压成 608x1080，转码产物校验失败降级， + # 用户拿到 HEVC 原文件浏览器仍黑帧。 + _src_w, _src_h, _rotation = probe_video_info(str(local_file)) + _is_portrait = is_portrait_video(_src_w, _src_h, _rotation) + logger.info( + "视频方向检测: stored=%sx%s rotation=%s portrait=%s: job_id=%s", + _src_w, + _src_h, + _rotation, + _is_portrait, + job_id, + ) # ── Step 3: ffmpeg 转码（独立 try/except）── try: @@ -315,11 +493,11 @@ def ingest_asset(job_id: str) -> dict: _tc_tmp = Path(_tc_tmp_file.name) _tc_tmp_file.close() # 关闭文件描述符，ffmpeg 会自己打开 - # 构建 video filter：竖屏先旋转再缩放 - if _needs_rotation: - _vf = "transpose=1,scale='if(gt(ih,1080),-2,iw)':'if(gt(ih,1080),1080,ih)'" - else: - _vf = "scale='if(gt(ih,1080),-2,iw)':'if(gt(ih,1080),1080,ih)'" + # 旋转交给 ffmpeg 内置 autorotate（按 display matrix 物理旋转， + # 输出自动剥离 side data）；滤镜只做 1080p 等比"只缩不放"。 + # 注意不能再加 transpose：旧逻辑 autorotate + transpose 双重旋转， + # 竖屏被转成横屏；scale 表达式内逗号必须 \, 转义（见 build_transcode_vf）。 + _vf = build_transcode_vf() _cmd = [ "ffmpeg", @@ -333,7 +511,7 @@ def ingest_asset(job_id: str) -> dict: "-crf", "18", "-vf", - _vf + ",format=yuv420p", + _vf, "-colorspace", "bt709", "-color_primaries", @@ -344,54 +522,62 @@ def ingest_asset(job_id: str) -> dict: "yuv420p", "-level", "4.2", + "-c:a", + "aac", + "-b:a", + "128k", + "-movflags", + "+faststart", + str(_tc_tmp), ] - # 竖屏视频：清除旋转元数据 - if _needs_rotation: - _cmd.extend(["-metadata:s:v:0", "rotate=0"]) - _cmd.extend( - [ - "-c:a", - "aac", - "-b:a", - "128k", - "-movflags", - "+faststart", - str(_tc_tmp), - ] - ) _proc = subprocess.run( _cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, - timeout=900, + timeout=TRANSCODE_TIMEOUT_SECONDS, ) if _proc.returncode == 0 and _tc_tmp.exists() and _tc_tmp.stat().st_size > 0: - from video_processing.oss_helpers import upload_to_oss - - _p = Path(job.storage_key) - _new_key = str(_p.parent / (_p.stem + "_h264" + _p.suffix)) - _url = upload_to_oss(_tc_tmp, _new_key) - if _url: - # 先提取元数据，确认成功后再更新 storage_key（避免脏数据） - _new_metadata, _new_extract_success = extract_media_metadata( - str(_tc_tmp), - media_type, - ) - if _new_extract_success: - job.storage_key = _new_key - metadata = _new_metadata - extract_success = _new_extract_success - logger.info( - "HEVC→H.264 转码完成: job_id=%s key=%s", + # ── Step 4: 方向/维度校验，不符则降级，杜绝横屏文件覆盖 ── + if not validate_transcode_output(str(_tc_tmp), _is_portrait): + _w, _h = probe_dimensions(str(_tc_tmp)) + _rot = probe_rotation(str(_tc_tmp)) + logger.error( + "转码产物方向/维度校验失败，降级使用原始文件: " + "job_id=%s source_rotation=%s portrait=%s out=%sx%s out_rotation=%s", job_id, - _new_key[:80], + _rotation, + _is_portrait, + _w, + _h, + _rot, ) else: - logger.warning( - "转码文件上传 OSS 失败，使用原始文件: job_id=%s", - job_id, - ) + from video_processing.oss_helpers import upload_to_oss + + _p = Path(job.storage_key) + _new_key = str(_p.parent / (_p.stem + "_h264" + _p.suffix)) + _url = upload_to_oss(_tc_tmp, _new_key) + if _url: + # 先提取元数据，确认成功后再更新 storage_key（避免脏数据） + _new_metadata, _new_extract_success = extract_media_metadata( + str(_tc_tmp), + media_type, + ) + if _new_extract_success: + job.storage_key = _new_key + metadata = _new_metadata + extract_success = _new_extract_success + logger.info( + "HEVC→H.264 转码完成: job_id=%s key=%s", + job_id, + _new_key[:80], + ) + else: + logger.warning( + "转码文件上传 OSS 失败，使用原始文件: job_id=%s", + job_id, + ) else: _tail = _proc.stderr[-300:] if _proc.stderr else "" logger.warning( diff --git a/tests/unit/test_ingest_hevc_transcode.py b/tests/unit/test_ingest_hevc_transcode.py index 34daa9741..11b975923 100644 --- a/tests/unit/test_ingest_hevc_transcode.py +++ b/tests/unit/test_ingest_hevc_transcode.py @@ -1,255 +1,560 @@ """HEVC 自动转码逻辑单元测试 (ingest.py) -测试覆盖： -- HEVC 编码检测逻辑 -- 转码后文件命名规则 -- 元数据提取失败时的脏数据防护 -- FFmpeg 超时/错误降级策略 -- 安全修复（tempfile、subprocess） -- Scale filter 逻辑 +测试覆盖（全部调用生产代码真实函数）： +- 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 -from unittest.mock import MagicMock, patch import pytest +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker")) -class TestHEVCAutoTranscode: - """测试 ingest_asset 中的 HEVC 自动转码逻辑""" +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, +) - def test_hevc_detection_keywords(self): - """验证 HEVC 编码的所有关键词""" - hevc_keywords = ("hevc", "h265", "hvh1") +# ── 纯逻辑测试（不依赖 ffmpeg）────────────────────────────────────────────── - 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 不应触发转码" +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_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)) + 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" - assert new_key == "uploads/video_123/test_h264.mp4" + def test_hevc_keywords_constant(self): + assert HEVC_CODECS == ("hevc", "h265", "hvh1") - 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" +class TestPortraitRotation: + def test_portrait_rotations(self): + for rotation in (90, 270, -90): + assert is_portrait_rotation(rotation), f"rotation={rotation} 应为竖屏" - def test_metadata_failure_no_dirty_data(self): - """验证元数据提取失败时不更新 storage_key（避免脏数据） + def test_non_portrait_rotations(self): + for rotation in (0, 180, -180, None): + assert not is_portrait_rotation(rotation), f"rotation={rotation} 不应判定为竖屏" - 这是 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} +class TestIsPortraitVideo: + """按显示方向判定竖屏（存储维度 + rotation 互换）。""" - # 模拟转码成功 - transcode_success = True + 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 - # 模拟元数据提取失败 - new_metadata = {} - new_extract_success = False + 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 - # 修复后的逻辑：先提取元数据，确认成功后再更新 - if transcode_success: - if new_extract_success: - job_storage_key = new_storage_key - metadata = new_metadata - # 如果元数据提取失败，不更新 job_storage_key + def test_landscape_normal(self): + assert is_portrait_video(1920, 1080, None) is False + assert is_portrait_video(1920, 1080, 0) is False - # 验证：storage_key 保持原值，没有脏数据 - assert job_storage_key == original_storage_key - assert metadata["codec"] == "hevc" # 保持原始元数据 + 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_metadata_success_updates_storage_key(self): - """验证元数据提取成功时正确更新 storage_key""" - original_storage_key = "uploads/test/video.mp4" - new_storage_key = "uploads/test/video_h264.mp4" + 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 - job_storage_key = original_storage_key - metadata = {"codec": "hevc", "width": 3840, "height": 2160} - # 模拟转码成功 - transcode_success = True +class TestBuildTranscodeVF: + """统一转码滤镜：长边封顶 1920、只缩不放、方向无关。""" - # 模拟元数据提取成功 - new_metadata = {"codec": "h264", "width": 1920, "height": 1080} - new_extract_success = True + 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 - # 修复后的逻辑 - if transcode_success: - if new_extract_success: - job_storage_key = new_storage_key - metadata = new_metadata + def test_no_transpose_filter(self): + """不能显式 transpose：ffmpeg autorotate 已按 side data 物理旋转， + 再加 transpose 会双重旋转把竖屏转成横屏。""" + assert "transpose" not in build_transcode_vf() - # 验证：storage_key 和 metadata 都更新为新值 - assert job_storage_key == new_storage_key - assert metadata["codec"] == "h264" - assert metadata["width"] == 1920 + 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 封顶" - @patch("subprocess.run") - def test_ffmpeg_timeout_degradation(self, mock_subprocess): - """验证 FFmpeg 超时降级使用原始文件""" - mock_subprocess.side_effect = subprocess.TimeoutExpired(cmd="ffmpeg", timeout=300) + def test_format_yuv420p_suffix(self): + assert build_transcode_vf().endswith(",format=yuv420p") - # 模拟降级逻辑 - transcode_success = False - try: - raise subprocess.TimeoutExpired(cmd="ffmpeg", timeout=300) - except subprocess.TimeoutExpired: - transcode_success = False - assert not transcode_success, "超时应该导致转码失败" +class TestProbeVideoInfo: + """probe_video_info 合并探测：维度 + rotation 一次 ffprobe。""" - @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", + 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 - result = mock_subprocess.return_value - transcode_success = result.returncode == 0 + 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) - 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 视频应该被缩放" +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_scale_filter_logic_1080p_video(self): - """验证 1080p 视频不会被缩放""" - ih = 1080 - should_scale = ih > 1080 - assert not should_scale, "1080p 视频不应该被缩放" + 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_scale_filter_logic_720p_video(self): - """验证 720p 视频不会被缩放""" - ih = 720 - should_scale = ih > 1080 - assert not should_scale, "720p 视频不应该被缩放" + 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_tempfile_security_fix(self): - """验证使用 NamedTemporaryFile 替代 mktemp（安全修复） + 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) - AI Code Review 发现的安全漏洞： - - tempfile.mktemp 存在 TOCTOU 竞态条件 - - 应该使用 NamedTemporaryFile(delete=False) + 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）。 """ - import tempfile + 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 的素材 - with patch("tempfile.NamedTemporaryFile") as mock_ntf: - mock_file = MagicMock() - mock_file.name = "/tmp/test_h264.mp4" - mock_ntf.return_value = mock_file + r90 = tag_rotate(base_hevc, 90) + r270 = tag_rotate(base_hevc, 270) - # 新代码的调用方式 - _tc_tmp_file = tempfile.NamedTemporaryFile(delete=False, suffix="_h264.mp4") - _tc_tmp = Path(_tc_tmp_file.name) - _tc_tmp_file.close() + # 物理竖屏素材（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, + ) - # 验证使用了 NamedTemporaryFile - mock_ntf.assert_called_once_with(delete=False, suffix="_h264.mp4") + # 超宽屏 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, + ) - def test_subprocess_output_handling(self): - """验证 subprocess 输出处理（避免内存溢出） + yield { + "portrait_r90": r90, + "portrait_r270": r270, + "portrait_physical": physical, + "ultra_wide": ultra_wide, + "landscape": base_hevc, + "tmp_path": tmp_path, + } - 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 = [ +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", - "-c:a", - "aac", - "-b:a", - "128k", + "-an", "-movflags", "+faststart", - ] + str(dst), + ], + check=True, + ) + return is_portrait, rotation - # 验证所有关键参数都在命令中 - 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 +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 - def test_hevc_codec_case_insensitive(self): - """验证 HEVC 检测不区分大小写""" - test_cases = ["hevc", "HEVC", "Hevc", "h265", "H265", "hvh1", "HVH1"] - hevc_keywords = ("hevc", "h265", "hvh1") + 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 - for codec in test_cases: - assert codec.lower() in hevc_keywords, f"{codec} 应该被检测为 HEVC" + 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_non_hevc_codecs(self): - """验证非 HEVC 编码不会触发转码""" - non_hevc_codecs = ["h264", "avc1", "vp9", "av1", "mpeg4", ""] - hevc_keywords = ("hevc", "h265", "hvh1") + 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 - for codec in non_hevc_codecs: - assert codec.lower() not in hevc_keywords, f"{codec} 不应触发转码" + 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 diff --git a/tests/unit/test_ingest_hevc_transcode_task.py b/tests/unit/test_ingest_hevc_transcode_task.py new file mode 100644 index 000000000..a02a86bc1 --- /dev/null +++ b/tests/unit/test_ingest_hevc_transcode_task.py @@ -0,0 +1,291 @@ +"""ingest_asset 任务中 HEVC 转码主流程的任务级单元测试。 + +通过 mock subprocess / repository / OSS，验证： +- 转码成功 + 方向校验通过 → storage_key 改写为 *_h264 +- 方向校验失败（竖屏转出横屏）→ 降级原文件，storage_key 不变，error 日志 +- ffmpeg 非零退出 → 降级原文件 +- 非 HEVC 编码 → 不触发转码 +""" + +from __future__ import annotations + +import sys +import tempfile +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +# 在 import worker_app 模块前 mock 掉数据库连接和 celery（同 test_ingest_validation.py） +_mock_db_module = MagicMock() +_mock_db_module.SessionLocal = MagicMock() +sys.modules["worker_app.db"] = _mock_db_module +sys.modules["worker_app.core.config"] = MagicMock() + +_mock_celery_module = MagicMock() + + +def _passthrough_decorator(*args, **kwargs): + if len(args) == 1 and callable(args[0]): + return args[0] + return lambda f: f + + +_mock_celery_module.celery_app.task = MagicMock(side_effect=_passthrough_decorator) +sys.modules["worker_app.celery_app"] = _mock_celery_module + +# mock video_processing 子模块（主流程会 import 它们） +_oss_helpers_mock = MagicMock() +_thumbnail_mock = MagicMock() +sys.modules["video_processing.oss_helpers"] = _oss_helpers_mock +sys.modules["video_processing.thumbnail_generator"] = _thumbnail_mock + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker")) + +import pytest # noqa: E402 +from worker_app.tasks import ingest as ingest_mod # noqa: E402 + + +class _FakeJobRepo: + def __init__(self, db): + self.initial_job = SimpleNamespace( + id="job-1", + project_id="proj-1", + library_id="lib-1", + storage_key="uploads/proj/IMG_2281.MOV", + file_hash="hash-1", + status=None, + error_message=None, + result_asset_id=None, + updated_at=None, + ) + self.updated_job = None + + def get(self, job_id): + return self.initial_job + + def update(self, job): + # 生产代码在同一 job 对象上原地修改属性后传入 update； + # 这里捕获引用，断言时读最终状态 + self.updated_job = job + + @property + def final_job(self): + return self.updated_job or self.initial_job + + +class _FakeAssetRepo: + def __init__(self, db): + self.created = None + + def create(self, asset): + self.created = asset + + +def _video_metadata(codec="hevc"): + return { + "codec": codec, + "width": 1920, + "height": 1080, + "duration": 10.0, + "size_bytes": 5 * 1024 * 1024, + } + + +@pytest.fixture +def task_env(tmp_path): + """统一构造 ingest_asset 主流程的 mock 环境。返回控制句柄。 + + 测试中用 mocks = _start_patches(control) 启动，断言必须用 + mocks["upload"] 等 start() 返回的 mock；不能在 stop() 后读模块 + 属性（stop 后属性恢复为原 auto-mock，调用记录为 0）。 + """ + local_file = tmp_path / "local_hevc.MOV" + local_file.write_bytes(b"fake-hevc-source") + tc_out = tmp_path / "transcode_out_h264.mp4" + + control = { + "rotation_source": 90, # 源文件 rotation；None=横屏无 side data + "transcode_rc": 0, + "transcode_produces_file": True, + "validate_ok": True, + "upload_url": "https://oss.example.com/x_h264.MOV", + "codec": "hevc", + "source_dims": (1920, 1080), + "tc_out": tc_out, + "local_file": local_file, + } + + def fake_probe_rotation(path): + if Path(path).name == tc_out.name: + return None # 产物无 side data + return control["rotation_source"] + + def fake_probe_dimensions(path): + if Path(path).name == tc_out.name: + return (1080, 1920) if control["validate_ok"] else (1920, 1080) + return control["source_dims"] + + control["subprocess_calls"] = [] + + def fake_subprocess_run(cmd, **kwargs): + control["subprocess_calls"].append(list(cmd[:3])) + if cmd and cmd[0] == "ffmpeg" and "libx264" in cmd: + if control["transcode_rc"] != 0: + return SimpleNamespace(returncode=control["transcode_rc"], stderr="boom") + if control["transcode_produces_file"]: + Path(cmd[-1]).write_bytes(b"fake-h264-output") + return SimpleNamespace(returncode=0, stderr="") + return SimpleNamespace(returncode=0, stdout="", stderr="") + + def fake_ntf(*args, **kwargs): + mock_file = MagicMock() + mock_file.name = str(tc_out) if kwargs.get("suffix") == "_h264.mp4" else str(local_file) + mock_file.close = MagicMock() + # with ... as tmp: 让 __enter__ 返回自身，tmp.name 才是上面设置的路径 + mock_file.__enter__.return_value = mock_file + mock_file.__exit__.return_value = False + return mock_file + + job_repo = _FakeJobRepo(db=None) + asset_repo = _FakeAssetRepo(db=None) + + control["patchers"] = { + "session": patch.object(ingest_mod, "SessionLocal", return_value=MagicMock()), + "job_repo": patch.object(ingest_mod, "SQLAlchemyIngestJobRepository", return_value=job_repo), + "asset_repo": patch.object(ingest_mod, "SQLAlchemyAssetRepository", return_value=asset_repo), + "download": patch.object(ingest_mod, "download_asset", return_value=True), + "upload": patch.object( + sys.modules["video_processing.oss_helpers"], + "upload_to_oss", + return_value=control["upload_url"], + ), + "metadata": patch.object( + ingest_mod, + "extract_media_metadata", + side_effect=lambda path, mt: ( + (_video_metadata("h264"), True) + if Path(path).name == tc_out.name + else (_video_metadata(control["codec"]), True) + ), + ), + "rotation": patch.object(ingest_mod, "probe_rotation", side_effect=fake_probe_rotation), + "dimensions": patch.object(ingest_mod, "probe_dimensions", side_effect=fake_probe_dimensions), + "validate": patch.object( + ingest_mod, + "validate_transcode_output", + side_effect=lambda p, portrait: control["validate_ok"], + ), + "subprocess": patch.object(ingest_mod.subprocess, "run", side_effect=fake_subprocess_run), + "ntf": patch.object(tempfile, "NamedTemporaryFile", side_effect=fake_ntf), + # 缩略图生成跳过 + "thumb": patch( + "video_processing.thumbnail_generator.extract_first_frame", + side_effect=RuntimeError("skip thumb"), + ), + } + + control["job_repo"] = job_repo + control["asset_repo"] = asset_repo + return control + + +def _start_patches(control): + """启动全部 patcher，返回具名 mock dict（调用记录都在这些 mock 上）。""" + return {name: p.start() for name, p in control["patchers"].items()} + + +def _stop_patches(control): + for p in control["patchers"].values(): + p.stop() + + +class TestIngestHEVCTranscodeFlow: + def test_success_rewrites_storage_key(self, task_env): + """竖屏 HEVC 转码+校验通过 → storage_key 改写为 *_h264.MOV，asset READY 入库。""" + mocks = _start_patches(task_env) + try: + result = ingest_mod.ingest_asset("job-1") + finally: + _stop_patches(task_env) + + assert result["status"] == "completed" + assert task_env["job_repo"].final_job.storage_key == "uploads/proj/IMG_2281_h264.MOV" + assert task_env["asset_repo"].created is not None + # 转码产物上传 OSS 恰好一次，且上传的是 *_h264.MOV 新 key + mocks["upload"].assert_called_once() + uploaded_path, uploaded_key = mocks["upload"].call_args.args + assert uploaded_key == "uploads/proj/IMG_2281_h264.MOV" + assert str(uploaded_path).endswith("_h264.mp4") + + def test_validation_failure_keeps_original_file(self, task_env): + """竖屏转出横屏（校验失败）→ 降级原文件，storage_key 不变，打 error 日志。""" + task_env["validate_ok"] = False + mocks = _start_patches(task_env) + error_mock = MagicMock() + try: + with patch.object(ingest_mod.logger, "error", error_mock): + result = ingest_mod.ingest_asset("job-1") + finally: + _stop_patches(task_env) + + assert result["status"] == "completed" + assert task_env["job_repo"].final_job.storage_key == "uploads/proj/IMG_2281.MOV" + assert error_mock.called + assert "方向/维度校验失败" in error_mock.call_args[0][0] + # 校验失败：转码产物不得上传 OSS，杜绝横屏文件覆盖 + mocks["upload"].assert_not_called() + + def test_ffmpeg_nonzero_keeps_original(self, task_env): + """ffmpeg 返回非零 → 降级原文件，storage_key 不变。""" + task_env["transcode_rc"] = 1 + mocks = _start_patches(task_env) + try: + result = ingest_mod.ingest_asset("job-1") + finally: + _stop_patches(task_env) + + assert result["status"] == "completed" + assert task_env["job_repo"].final_job.storage_key == "uploads/proj/IMG_2281.MOV" + mocks["upload"].assert_not_called() + + def test_physical_portrait_no_rotation_still_transcodes(self, task_env): + """物理竖屏（存储 1080x1920、rotation=None，Android 风格）也必须判定竖屏 + 并转码改写 storage_key——回归旧逻辑只看 rotation 误判横屏的 bug。""" + task_env["source_dims"] = (1080, 1920) + task_env["rotation_source"] = None + mocks = _start_patches(task_env) + try: + result = ingest_mod.ingest_asset("job-1") + finally: + _stop_patches(task_env) + + assert result["status"] == "completed" + assert task_env["job_repo"].final_job.storage_key == "uploads/proj/IMG_2281_h264.MOV" + mocks["upload"].assert_called_once() + # 统一滤镜按长边 1920 封顶（横/竖分支都在），不应再出现按短边 1080 的旧表达式 + cmds = [] + for call in mocks["subprocess"].call_args_list: + cmd = call.args[0] if call.args else call.kwargs.get("cmd", []) + cmds.append(cmd) + vfs = [str(c) for c in cmds if c and c[0] == "ffmpeg" and "libx264" in c] + assert vfs, "应执行 libx264 转码" + assert any("min(1920" in vf for vf in vfs), f"应使用长边1920封顶滤镜: {vfs[0]}" + assert all("gt(ih,1080)" not in vf for vf in vfs), "不应再用短边1080旧表达式" + + def test_non_hevc_no_transcode(self, task_env): + """非 HEVC 编码（h264）→ 不触发 ffmpeg 转码。""" + task_env["codec"] = "h264" + task_env["rotation_source"] = None + mocks = _start_patches(task_env) + try: + result = ingest_mod.ingest_asset("job-1") + finally: + _stop_patches(task_env) + + assert result["status"] == "completed" + assert task_env["job_repo"].final_job.storage_key == "uploads/proj/IMG_2281.MOV" + mocks["upload"].assert_not_called() + # 所有 subprocess 调用都不应是 ffmpeg 转码 + for call in mocks["subprocess"].call_args_list: + cmd = call.args[0] if call.args else call.kwargs.get("cmd", []) + assert "libx264" not in cmd