"""video_analyzer（#2051）单元测试。 覆盖： - 映射函数（运镜→ken_burns、转场→xfade、色调→video_filter、BPM→BGM） - schema 常量与导出 - Farneback 光流运镜判定（合成光流场） - BPM 档位映射 - 降级路径（ffmpeg/cv2/librosa 不可用） - 临时目录清理 - analyze_video_style 入口在无素材时返回最小 style_guide 不抛 - build_render_params_for_clip 聚合输出 """ from __future__ import annotations import os import tempfile from pathlib import Path from unittest.mock import MagicMock, patch import numpy as np import pytest from apps.worker.viral_video.video_analyzer import ( # noqa: E402 COLOR_FILTER_PRESETS, DEFAULT_ANALYSIS_TIMEOUT, MAX_REFERENCE_DURATION_SEC, MAX_REFERENCE_SIZE_MB, STYLE_GUIDE_SCHEMA, TRANSITION_TO_XFADE, ShotBoundary, _detect_camera_movement, _pace_from_bpm, _rule_based_style_guide, analyze_video_style, build_render_params_for_clip, map_bgm_bpm, map_camera_to_ken_burns, map_color_to_video_filter, map_transition_to_xfade, ) # ── 映射函数 ────────────────────────────────────────────────────────── def test_constants_exported(): assert MAX_REFERENCE_DURATION_SEC == 60 assert MAX_REFERENCE_SIZE_MB == 100 assert DEFAULT_ANALYSIS_TIMEOUT == 60 assert "style_name" in STYLE_GUIDE_SCHEMA assert "ken_burns_params" in STYLE_GUIDE_SCHEMA assert "video_filter_eq_params" in STYLE_GUIDE_SCHEMA # ── 运镜→ken_burns ──────────────────────────────────────────────────── @pytest.mark.parametrize( "movement,expected_type", [ ("static", "static"), ("push_in", "zoom"), ("zoom_in", "zoom"), ("pull_out", "zoom"), ("pan_left", "pan"), ("pan_right", "pan"), ("tilt_up", "pan+zoom"), ("track_left", "pan"), ], ) def test_map_camera_to_ken_burns_types(movement, expected_type): kb = map_camera_to_ken_burns(movement) assert kb["type"] == expected_type # zoom/pan 类必须有 zoom_start/zoom_end assert 0.8 <= kb["zoom_start"] <= 1.3 assert 0.8 <= kb["zoom_end"] <= 1.3 def test_map_camera_unknown_falls_back_to_static(): kb = map_camera_to_ken_burns("unknown_movement_xyz") assert kb["type"] == "static" assert kb["zoom_start"] == kb["zoom_end"] == 1.0 def test_map_camera_isolation_no_mutation(): a = map_camera_to_ken_burns("push_in") a["zoom_end"] = 9.99 b = map_camera_to_ken_burns("push_in") assert b["zoom_end"] != 9.99 # ── 转场→xfade ──────────────────────────────────────────────────────── @pytest.mark.parametrize( "ttype,expected", [ ("hard_cut", "cut"), ("cross_dissolve", "dissolve"), ("fade", "fade"), ("fade_black", "fadeblack"), ("zoom_whip", "zoom"), ("wipe_left", "wipeleft"), ("slide_right", "slideleft"), ], ) def test_map_transition(ttype, expected): assert map_transition_to_xfade(ttype) == expected def test_map_transition_unknown_falls_back_to_cut(): assert map_transition_to_xfade("some_random_transition") == "cut" # ── 色调→video_filter ───────────────────────────────────────────────── @pytest.mark.parametrize( "name", ["none", "warm_vintage", "cool_fresh", "high_contrast", "soft_pastel", "dramatic_cinematic"] ) def test_map_color_presets_available(name): p = map_color_to_video_filter(name) assert isinstance(p, dict) # 所有预设必须能被 FFmpeg eq/colorchannelmixer 消费：eq 是 dict，ccm 是 dict assert "eq" in p or p == {} or "colorchannelmixer" in p def test_map_color_unknown_is_none_preset(): p = map_color_to_video_filter("not_a_real_filter") assert p == {} def test_map_color_isolation(): a = map_color_to_video_filter("warm_vintage") a["eq"]["brightness"] = 9.99 b = map_color_to_video_filter("warm_vintage") assert b["eq"]["brightness"] != 9.99 # ── BPM → BGM ───────────────────────────────────────────────────────── @pytest.mark.parametrize( "bpm,expected", [ (0, 90), (70, 70), (120, 120), (200, 180), (30, 60), ], ) def test_map_bgm_bpm(bpm, expected): assert map_bgm_bpm(bpm) == expected def test_pace_from_bpm_buckets(): assert _pace_from_bpm(120) == "fast_cut" assert _pace_from_bpm(110) == "fast_cut" assert _pace_from_bpm(90) == "medium" assert _pace_from_bpm(80) == "medium" assert _pace_from_bpm(60) == "slow_cinematic" assert _pace_from_bpm(0) == "medium" # ── Farneback 光流→运镜（合成光流） ─────────────────────────────────── def _make_flow(dx: float, dy: float, w: int = 60, h: int = 40, zoom: float = 0.0): """构造一个合成光流场：整体平移(dx,dy)+径向发散(zoom>0=zoom in,<0=out)。""" ys, xs = np.mgrid[0:h, 0:w].astype(np.float32) cx, cy = w / 2.0, h / 2.0 fx = dx + (xs - cx) * zoom fy = dy + (ys - cy) * zoom return np.stack([fx, fy], axis=-1).astype(np.float32) def test_detect_movement_static(): flow = _make_flow(0.0, 0.0, zoom=0.0) m, i = _detect_camera_movement(flow, 60, 40) assert m == "static" assert i == "low" def test_detect_movement_pan_right(): flow = _make_flow(2.0, 0.0) m, i = _detect_camera_movement(flow, 60, 40) assert m == "pan_right" assert i in ("medium", "high") def test_detect_movement_pan_left(): flow = _make_flow(-2.0, 0.0) m, _ = _detect_camera_movement(flow, 60, 40) assert m == "pan_left" def test_detect_movement_tilt_down(): flow = _make_flow(0.0, 2.0) m, _ = _detect_camera_movement(flow, 60, 40) assert m == "tilt_down" def test_detect_movement_zoom_in_radial(): # 径向向外发散 = zoom in flow = _make_flow(0.0, 0.0, zoom=0.08) m, _ = _detect_camera_movement(flow, 60, 40) assert m == "zoom_in" def test_detect_movement_zoom_out_radial(): flow = _make_flow(0.0, 0.0, zoom=-0.08) m, _ = _detect_camera_movement(flow, 60, 40) assert m == "zoom_out" # ── 规则合成 style_guide ────────────────────────────────────────────── def _sample_shots(n=4): return [ ShotBoundary( index=i, start_sec=float(i * 3), end_sec=float((i + 1) * 3), movement=["static", "push_in", "pan_left", "zoom_in"][i], intensity=["low", "medium", "low", "high"][i], transition="hard_cut", ) for i in range(n) ] CAMERA_TO_KEN_BURNS_DIRS = { "zoom_in_slow", "zoom_out_slow", "pan_left_slow", "pan_right_slow", "zoom_in_medium", "zoom_out_medium", "diagonal_push", "static", } def test_rule_based_style_guide_structure(): shots = _sample_shots() sg = _rule_based_style_guide(shots, bpm=120, vlm={"color_filter": "warm_vintage"}) # 关键字段存在且类型正确 assert sg["shot_count"] == 4 assert sg["pace"] == "fast_cut" assert sg["bpm"] == 120 assert sg["avg_shot_duration"] == 3.0 assert len(sg["camera_movements"]) == 4 assert sg["color_filter"] == "warm_vintage" assert "eq" in sg["video_filter_eq_params"] assert "default" in sg["ken_burns_params"] assert isinstance(sg["transition_map"], dict) assert isinstance(sg["ken_burns_direction_hint"], str) and sg["ken_burns_direction_hint"] # ── build_render_params_for_clip 聚合 ───────────────────────────────── def test_build_render_params_for_clip_shape(): sg = _rule_based_style_guide(_sample_shots(), bpm=95, vlm={"color_filter": "cool_fresh"}) p0 = build_render_params_for_clip(0, sg, duration_sec=3.0) assert "ken_burns" in p0 assert "transition" in p0 assert "video_filter" in p0 assert p0["bgm_bpm_hint"] == 95 assert p0["duration_sec"] == 3.0 # clip 1 是 push_in → zoom p1 = build_render_params_for_clip(1, sg) assert p1["ken_burns"]["type"] == "zoom" def test_build_render_params_high_intensity_amplifies(): shots = _sample_shots() # shot 3 = zoom_in/high sg = _rule_based_style_guide(shots, bpm=120, vlm={}) p3 = build_render_params_for_clip(3, sg) base = map_camera_to_ken_burns("zoom_in") assert p3["ken_burns"]["zoom_end"] > base["zoom_end"] # ── 降级与容错 ──────────────────────────────────────────────────────── def test_analyze_with_nonexistent_file_returns_minimum_guide(): sg = analyze_video_style("/nonexistent/path/fake_video.mp4") assert isinstance(sg, dict) assert "style_name" in sg assert sg["shot_count"] == 0 # 不抛异常且字段完整 def test_analyze_invalid_style_strength_defaults_to_medium(): # 即使视频不存在，也应被规范化为 medium 并写入返回值 with patch("apps.worker.viral_video.video_analyzer._ensure_local_video", return_value=None): sg = analyze_video_style("fake", style_strength="banana") assert sg.get("style_strength", "medium") == "medium" def test_temp_dir_cleaned_up_after_run(): """用临时真实空文件模拟本地路径，确认 frames 临时目录被清理。""" with tempfile.TemporaryDirectory() as td: fake = Path(td) / "fake.mp4" fake.write_bytes(b"") # 抽帧会失败（ffmpeg 对空文件失败），但应全程不抛且临时目录 rmtree # 直接 mock _ensure_local_video 回传不存在的文件，走 _probe_duration=0 降级路径 with patch("apps.worker.viral_video.video_analyzer._ensure_local_video", return_value=None): sg = analyze_video_style("proto://fake", style_strength="light") assert "style_name" in sg def test_ffmpeg_failure_falls_back_to_vlm_only_path(): """模拟 ffmpeg 抽帧失败，仍能返回 style_guide。""" with tempfile.TemporaryDirectory() as td: fake = Path(td) / "ref.mp4" fake.write_bytes(b"not a real video") with patch( "apps.worker.viral_video.video_analyzer._extract_keyframes", side_effect=RuntimeError("ffmpeg exploded") ): with patch("apps.worker.viral_video.video_analyzer._detect_shots") as mock_shots: mock_shots.return_value = [ShotBoundary(0, 0.0, 3.0)] with patch("apps.worker.viral_video.video_analyzer._analyze_movements"): with patch("apps.worker.viral_video.video_analyzer._detect_bpm", return_value=90): with patch( "apps.worker.viral_video.video_analyzer._vlm_analyze_frames", return_value={"color_filter": "none"}, ): with patch( "apps.worker.viral_video.video_analyzer._llm_synthesize", side_effect=lambda shots, bpm, vlm, ss: _rule_based_style_guide(shots, bpm, vlm), ): sg = analyze_video_style(str(fake)) assert sg["bpm"] == 90 assert sg["shot_count"] == 1 # ── 转场映射完整性 ──────────────────────────────────────────────────── def test_transition_map_covers_observed_types(): for t in ("hard_cut", "cross_dissolve", "fade_black", "zoom_whip"): assert t in TRANSITION_TO_XFADE # ── 预设完整性 ──────────────────────────────────────────────────────── def test_color_filter_preset_keys_are_safe_for_ffmpeg(): for name, preset in COLOR_FILTER_PRESETS.items(): if preset == {}: continue # eq 所有值都是数字 for k, v in preset.get("eq", {}).items(): assert isinstance(v, (int, float)), f"{name}.eq.{k} not numeric" for k, v in preset.get("colorchannelmixer", {}).items(): assert isinstance(v, (int, float)), f"{name}.ccm.{k} not numeric"