"""pip_engine_pure 单元测试.""" from pathlib import Path import pytest from apps.worker.video_processing.pip_engine_pure import ( build_animation_filters, build_enable_expr, build_overlay_expr, build_pip_filters, build_pip_pre_filter, compute_pip_position, compute_pip_size, count_visible_layers, sort_layers_by_z_index, validate_pip_layer, ) from packages.domain.pip_config import ( ANIMATION_FADE, ANIMATION_SLIDE_BOTTOM, ANIMATION_SLIDE_LEFT, ANIMATION_SLIDE_RIGHT, ANIMATION_SLIDE_TOP, PiPLayerConfig, ) def _make_layer(**kwargs): """快速创建 PiPLayerConfig.""" layer = PiPLayerConfig() for k, v in kwargs.items(): setattr(layer, k, v) return layer # ── compute_pip_size ──────────────────────────────────────────────────────── class TestComputePipSize: def test_percentage_width_auto_height(self): layer = _make_layer(width="30%") w, h = compute_pip_size(layer, 1920, 1080) assert w == 576 # 1920 * 0.3 assert h == int(576 * 9 / 16) # 按16:9等比 def test_pixel_width_and_height(self): layer = _make_layer(width="400", height="300") w, h = compute_pip_size(layer, 1920, 1080) assert w == 400 assert h == 300 def test_int_width_and_height(self): layer = _make_layer(width=500, height=400) w, h = compute_pip_size(layer, 1920, 1080) assert w == 500 assert h == 400 def test_width_exceeds_output_clamped(self): layer = _make_layer(width="200%") w, h = compute_pip_size(layer, 1920, 1080) assert w == 1920 assert h <= 1080 def test_height_exceeds_output_clamped(self): layer = _make_layer(width="100", height="200%") w, h = compute_pip_size(layer, 1920, 1080) assert w == 100 assert h == 1080 def test_minimum_size(self): layer = _make_layer(width="0", height="0") w, h = compute_pip_size(layer, 1920, 1080) assert w >= 1 assert h >= 1 def test_empty_height_auto_ratio(self): layer = _make_layer(width="320", height="") w, h = compute_pip_size(layer, 1920, 1080) assert w == 320 assert h == int(320 * 9 / 16) # ── compute_pip_position ──────────────────────────────────────────────────── class TestComputePipPosition: def test_bottom_right_position(self): layer = _make_layer(position="bottom_right", margin=20) x, y = compute_pip_position(layer, 200, 150, 1920, 1080) assert x == 1920 - 200 - 20 assert y == 1080 - 150 - 20 def test_top_left_position(self): layer = _make_layer(position="top_left", margin=10) x, y = compute_pip_position(layer, 200, 150, 1920, 1080) assert x == 10 assert y == 10 def test_center_position(self): layer = _make_layer(position="center") x, y = compute_pip_position(layer, 200, 150, 1920, 1080) assert x == (1920 - 200) // 2 assert y == (1080 - 150) // 2 def test_custom_position(self): layer = _make_layer(position="custom", x=100, y=200) x, y = compute_pip_position(layer, 200, 150, 1920, 1080) assert x == 100 assert y == 200 def test_clamped_to_left_edge(self): # parse_size_value 有 max(1, value) 钳制,负数返回1 layer = _make_layer(position="custom", x=-100, y=0) x, y = compute_pip_position(layer, 200, 150, 1920, 1080) assert x >= 0 assert x < 200 def test_clamped_to_right_edge(self): layer = _make_layer(position="custom", x=9999, y=0) x, y = compute_pip_position(layer, 200, 150, 1920, 1080) assert x == 1920 - 200 def test_clamped_to_top_edge(self): # parse_size_value 有 max(1, value) 钳制,负数返回1 layer = _make_layer(position="custom", x=0, y=-50) x, y = compute_pip_position(layer, 200, 150, 1920, 1080) assert y >= 0 assert y < 150 def test_clamped_to_bottom_edge(self): layer = _make_layer(position="custom", x=0, y=9999) x, y = compute_pip_position(layer, 200, 150, 1920, 1080) assert y == 1080 - 150 # ── build_pip_pre_filter ──────────────────────────────────────────────────── class TestBuildPipPreFilter: def test_basic_scale_and_sar(self): layer = _make_layer(width="200", height="150") result = build_pip_pre_filter("[1:v]", layer, 200, 150, "pip_pre_0") assert result.startswith("[1:v]") assert "scale=200:150" in result assert "setsar=1" in result assert result.endswith("[pip_pre_0]") def test_with_corner_radius(self): layer = _make_layer(corner_radius=20) result = build_pip_pre_filter("[1:v]", layer, 200, 150, "pip_pre_0") assert "geq=" in result assert "format=yuva420p" in result def test_corner_radius_zero(self): layer = _make_layer(corner_radius=0) result = build_pip_pre_filter("[1:v]", layer, 200, 150, "pip_pre_0") assert "geq=" not in result def test_with_border(self): layer = _make_layer(border_width=5, border_color="red") result = build_pip_pre_filter("[1:v]", layer, 200, 150, "pip_pre_0") assert "pad=210:160:5:5:red" in result def test_border_zero(self): layer = _make_layer(border_width=0) result = build_pip_pre_filter("[1:v]", layer, 200, 150, "pip_pre_0") assert "pad=" not in result def test_with_opacity(self): layer = _make_layer(opacity=0.5) result = build_pip_pre_filter("[1:v]", layer, 200, 150, "pip_pre_0") assert "colorchannelmixer=aa=0.5" in result def test_full_opacity_no_alpha_filter(self): layer = _make_layer(opacity=1.0) result = build_pip_pre_filter("[1:v]", layer, 200, 150, "pip_pre_0") assert "colorchannelmixer" not in result def test_opacity_clamped(self): layer = _make_layer(opacity=-0.5) result = build_pip_pre_filter("[1:v]", layer, 200, 150, "pip_pre_0") assert "colorchannelmixer=aa=0.0" in result def test_with_fade_animation(self): layer = _make_layer( animation_in=ANIMATION_FADE, animation_out=ANIMATION_FADE, animation_duration=0.5, duration=3.0, ) result = build_pip_pre_filter("[1:v]", layer, 200, 150, "pip_pre_0") assert "fade=t=in:st=0:d=0.5:alpha=1" in result assert "fade=t=out" in result def test_all_features_combined(self): layer = _make_layer( corner_radius=15, border_width=3, border_color="blue", opacity=0.8, animation_in=ANIMATION_FADE, animation_duration=0.3, duration=5.0, ) result = build_pip_pre_filter("[1:v]", layer, 300, 200, "pip_out") assert "scale=300:200" in result assert "geq=" in result assert "pad=" in result assert "colorchannelmixer=aa=0.8" in result assert "fade=t=in" in result assert result.endswith("[pip_out]") # ── build_animation_filters ───────────────────────────────────────────────── class TestBuildAnimationFilters: def test_no_animation(self): layer = _make_layer() assert build_animation_filters(layer, 200, 150) == [] def test_fade_in_only(self): layer = _make_layer(animation_in=ANIMATION_FADE, animation_duration=0.5) result = build_animation_filters(layer, 200, 150) assert len(result) == 1 assert "fade=t=in:st=0:d=0.5:alpha=1" in result[0] def test_fade_out_requires_duration(self): layer = _make_layer(animation_out=ANIMATION_FADE, animation_duration=0.5) result = build_animation_filters(layer, 200, 150) # 没有duration,出场动画不生效 assert len(result) == 0 def test_fade_out_with_duration(self): layer = _make_layer( animation_out=ANIMATION_FADE, animation_duration=0.5, duration=5.0, ) result = build_animation_filters(layer, 200, 150) assert len(result) == 1 assert "fade=t=out" in result[0] assert "st=4.5" in result[0] # 5.0 - 0.5 def test_both_fade_animations(self): layer = _make_layer( animation_in=ANIMATION_FADE, animation_out=ANIMATION_FADE, animation_duration=0.3, duration=4.0, ) result = build_animation_filters(layer, 200, 150) assert len(result) == 2 assert "fade=t=in" in result[0] assert "fade=t=out" in result[1] def test_zero_duration_animation(self): layer = _make_layer(animation_in=ANIMATION_FADE, animation_duration=0) result = build_animation_filters(layer, 200, 150) assert result == [] def test_negative_animation_duration_clamped(self): layer = _make_layer(animation_in=ANIMATION_FADE, animation_duration=-1) result = build_animation_filters(layer, 200, 150) assert result == [] def test_slide_animation_not_in_this_function(self): # slide类动画不在这个函数处理 layer = _make_layer(animation_in=ANIMATION_SLIDE_LEFT, animation_duration=0.5) result = build_animation_filters(layer, 200, 150) assert result == [] # ── build_overlay_expr ────────────────────────────────────────────────────── class TestBuildOverlayExpr: def test_no_animation_static_position(self): layer = _make_layer() x_expr, y_expr = build_overlay_expr(layer, 100, 200, 200, 150, 1920, 1080) assert x_expr == "100" assert y_expr == "200" def test_slide_left_enter(self): layer = _make_layer(animation_in=ANIMATION_SLIDE_LEFT, animation_duration=0.5) x_expr, y_expr = build_overlay_expr(layer, 100, 200, 200, 150, 1920, 1080) assert "if(lt(t,0.5)" in x_expr assert y_expr == "200" def test_slide_right_enter(self): layer = _make_layer(animation_in=ANIMATION_SLIDE_RIGHT, animation_duration=0.5) x_expr, _ = build_overlay_expr(layer, 100, 200, 200, 150, 1920, 1080) assert "output_width" not in x_expr # 应该是具体数值 assert "if(lt(t,0.5)" in x_expr assert "1920" in x_expr def test_slide_top_enter(self): layer = _make_layer(animation_in=ANIMATION_SLIDE_TOP, animation_duration=0.5) x_expr, y_expr = build_overlay_expr(layer, 100, 200, 200, 150, 1920, 1080) assert x_expr == "100" assert "if(lt(t,0.5)" in y_expr def test_slide_bottom_enter(self): layer = _make_layer(animation_in=ANIMATION_SLIDE_BOTTOM, animation_duration=0.5) _, y_expr = build_overlay_expr(layer, 100, 200, 200, 150, 1920, 1080) assert "if(lt(t,0.5)" in y_expr assert "1080" in y_expr def test_slide_left_exit(self): layer = _make_layer( animation_out=ANIMATION_SLIDE_LEFT, animation_duration=0.5, duration=5.0, ) x_expr, _ = build_overlay_expr(layer, 100, 200, 200, 150, 1920, 1080) assert "if(gt(t,4.5)" in x_expr def test_slide_right_exit(self): layer = _make_layer( animation_out=ANIMATION_SLIDE_RIGHT, animation_duration=0.5, duration=5.0, ) x_expr, _ = build_overlay_expr(layer, 100, 200, 200, 150, 1920, 1080) assert "if(gt(t,4.5)" in x_expr def test_slide_top_exit(self): layer = _make_layer( animation_out=ANIMATION_SLIDE_TOP, animation_duration=0.5, duration=5.0, ) _, y_expr = build_overlay_expr(layer, 100, 200, 200, 150, 1920, 1080) assert "if(gt(t,4.5)" in y_expr def test_slide_bottom_exit(self): layer = _make_layer( animation_out=ANIMATION_SLIDE_BOTTOM, animation_duration=0.5, duration=5.0, ) _, y_expr = build_overlay_expr(layer, 100, 200, 200, 150, 1920, 1080) assert "if(gt(t,4.5)" in y_expr def test_exit_animation_no_duration_ignored(self): layer = _make_layer(animation_out=ANIMATION_SLIDE_LEFT, animation_duration=0.5) x_expr, y_expr = build_overlay_expr(layer, 100, 200, 200, 150, 1920, 1080) assert x_expr == "100" assert y_expr == "200" def test_zero_animation_duration_no_effect(self): layer = _make_layer(animation_in=ANIMATION_SLIDE_LEFT, animation_duration=0) x_expr, y_expr = build_overlay_expr(layer, 100, 200, 200, 150, 1920, 1080) assert x_expr == "100" assert y_expr == "200" # ── build_enable_expr ──────────────────────────────────────────────────────── class TestBuildEnableExpr: def test_no_start_no_duration_empty(self): layer = _make_layer(start_time=0.0) # duration默认是0.0 assert build_enable_expr(layer) == "" def test_with_start_and_duration(self): layer = _make_layer(start_time=2.0, duration=3.0) result = build_enable_expr(layer) assert result == ":enable='between(t,2.0,5.0)'" def test_with_start_only_no_duration(self): layer = _make_layer(start_time=5.0) result = build_enable_expr(layer) assert result == ":enable='gte(t,5.0)'" def test_negative_start_clamped(self): layer = _make_layer(start_time=-1.0, duration=5.0) result = build_enable_expr(layer) assert "between(t,0.0," in result def test_zero_duration_with_start(self): layer = _make_layer(start_time=3.0, duration=0.0) result = build_enable_expr(layer) assert "gte(t,3.0)" in result def test_zero_start_and_duration(self): layer = _make_layer(start_time=0, duration=0) assert build_enable_expr(layer) == "" # ── build_pip_filters ─────────────────────────────────────────────────────── class TestBuildPipFilters: def test_empty_layers(self): filter_parts, input_args, final = build_pip_filters("base", [], [], output_width=1920, output_height=1080) assert filter_parts == [] assert input_args == [] assert final == "base" def test_single_layer(self): layer = _make_layer(width="200", height="150", position="top_left", source="test.mp4") filter_parts, input_args, final = build_pip_filters( "base", [layer], ["/tmp/test.mp4"], output_width=1920, output_height=1080 ) assert len(filter_parts) == 2 # pre_filter + overlay assert input_args == ["-i", "/tmp/test.mp4"] assert final == "pip_combined_0" def test_multiple_layers(self): layers = [ _make_layer(width="100", height="100", position="top_left"), _make_layer(width="100", height="100", position="top_right"), ] sources = ["/tmp/a.mp4", "/tmp/b.mp4"] filter_parts, input_args, final = build_pip_filters( "base", layers, sources, output_width=1920, output_height=1080 ) assert len(filter_parts) == 4 # 2 pre + 2 overlay assert len(input_args) == 4 # 2 * (-i, path) assert final == "pip_combined_1" def test_mismatched_layers_and_sources(self): layer = _make_layer() with pytest.raises(ValueError, match="长度不一致"): build_pip_filters("base", [layer], [], output_width=1920, output_height=1080) def test_custom_base_input_index(self): layer = _make_layer(width="100", height="100", position="top_left") filter_parts, input_args, _ = build_pip_filters( "base", [layer], ["/a.mp4"], output_width=1920, output_height=1080, base_input_idx=5, ) assert "[5:v]" in filter_parts[0] assert len(input_args) == 2 def test_path_object_supported(self): layer = _make_layer(width="100", height="100", position="top_left") _, input_args, _ = build_pip_filters( "base", [layer], [Path("/tmp/test.mp4")], output_width=1920, output_height=1080, ) assert input_args[1] == "/tmp/test.mp4" def test_filter_parts_contain_correct_labels(self): layer = _make_layer(width="100", height="100", position="top_left") filter_parts, _, _ = build_pip_filters( "base", [layer], ["/a.mp4"], output_width=1920, output_height=1080, ) # pre filter输出 pip_pre_0 assert "[pip_pre_0]" in filter_parts[0] # overlay使用 pip_pre_0 作为输入 assert "[pip_pre_0]" in filter_parts[1] # overlay输出 pip_combined_0 assert "[pip_combined_0]" in filter_parts[1] # ── validate_pip_layer ────────────────────────────────────────────────────── class TestValidatePipLayer: def test_valid_layer(self): layer = _make_layer( source_type="asset_id", source="asset_123", width="25%", position="bottom_right", ) valid, err = validate_pip_layer(layer) assert valid is True assert err == "" def test_missing_source_type(self): layer = _make_layer(source_type="", source="abc", width="25%", position="top_left") valid, err = validate_pip_layer(layer) assert valid is False assert "source_type" in err def test_invalid_source_type(self): layer = _make_layer(source_type="invalid", source="abc", width="25%", position="top_left") valid, err = validate_pip_layer(layer) assert valid is False assert "source_type" in err def test_missing_source(self): layer = _make_layer(source_type="asset_id", source="", width="25%", position="top_left") valid, err = validate_pip_layer(layer) assert valid is False assert "source" in err def test_missing_width(self): layer = _make_layer(source_type="asset_id", source="abc", width=None, position="top_left") valid, err = validate_pip_layer(layer) assert valid is False assert "width" in err def test_empty_width(self): layer = _make_layer(source_type="asset_id", source="abc", width="", position="top_left") valid, err = validate_pip_layer(layer) assert valid is False assert "width" in err def test_invalid_position(self): layer = _make_layer(source_type="asset_id", source="abc", width="25%", position="invalid_pos") valid, err = validate_pip_layer(layer) assert valid is False assert "position" in err def test_invalid_opacity_high(self): layer = _make_layer(source_type="asset_id", source="abc", width="25%", opacity=2.0) valid, err = validate_pip_layer(layer) assert valid is False assert "opacity" in err def test_invalid_opacity_low(self): layer = _make_layer(source_type="asset_id", source="abc", width="25%", opacity=-0.5) valid, err = validate_pip_layer(layer) assert valid is False assert "opacity" in err def test_negative_corner_radius(self): layer = _make_layer(source_type="asset_id", source="abc", width="25%", corner_radius=-1) valid, err = validate_pip_layer(layer) assert valid is False assert "corner_radius" in err def test_negative_border_width(self): layer = _make_layer(source_type="asset_id", source="abc", width="25%", border_width=-1) valid, err = validate_pip_layer(layer) assert valid is False assert "border_width" in err def test_negative_animation_duration(self): layer = _make_layer(source_type="asset_id", source="abc", width="25%", animation_duration=-1) valid, err = validate_pip_layer(layer) assert valid is False assert "animation_duration" in err def test_negative_start_time(self): layer = _make_layer(source_type="asset_id", source="abc", width="25%", start_time=-1) valid, err = validate_pip_layer(layer) assert valid is False assert "start_time" in err def test_negative_duration(self): layer = _make_layer(source_type="asset_id", source="abc", width="25%", duration=-1) valid, err = validate_pip_layer(layer) assert valid is False assert "duration" in err def test_invalid_animation_in(self): layer = _make_layer(source_type="asset_id", source="abc", width="25%", animation_in="invalid") valid, err = validate_pip_layer(layer) assert valid is False assert "animation_in" in err def test_invalid_animation_out(self): layer = _make_layer(source_type="asset_id", source="abc", width="25%", animation_out="invalid") valid, err = validate_pip_layer(layer) assert valid is False assert "animation_out" in err def test_multiple_errors_joined(self): layer = _make_layer(source_type="", source="", width=None, position="bad") valid, err = validate_pip_layer(layer) assert valid is False assert err.count(";") >= 2 # 至少3个错误,2个分号 def test_all_source_types_valid(self): for st in ("local_path", "asset_id", "url"): layer = _make_layer(source_type=st, source="abc", width="25%") valid, _ = validate_pip_layer(layer) assert valid is True def test_all_positions_valid(self): for pos in ( "top_left", "top_center", "top_right", "center_left", "center", "center_right", "bottom_left", "bottom_center", "bottom_right", "custom", ): layer = _make_layer(source_type="asset_id", source="abc", width="25%", position=pos) valid, err = validate_pip_layer(layer) assert valid is True, f"position {pos} should be valid: {err}" # ── count_visible_layers ──────────────────────────────────────────────────── class TestCountVisibleLayers: def test_all_visible(self): layers = [ _make_layer(opacity=1.0), _make_layer(opacity=0.5), ] assert count_visible_layers(layers) == 2 def test_none_visible(self): layers = [ _make_layer(opacity=0.0), _make_layer(opacity=0.0), ] assert count_visible_layers(layers) == 0 def test_mixed(self): layers = [ _make_layer(opacity=1.0), _make_layer(opacity=0.0), _make_layer(opacity=0.1), _make_layer(opacity=0.0), ] assert count_visible_layers(layers) == 2 def test_empty_list(self): assert count_visible_layers([]) == 0 def test_negative_opacity_not_counted(self): # opacity < 0 也不算可见 layer = _make_layer(opacity=-0.5) assert count_visible_layers([layer]) == 0 # ── sort_layers_by_z_index ────────────────────────────────────────────────── class TestSortLayersByZIndex: def test_sorted_by_z_index(self): l1 = _make_layer(z_index=3) l2 = _make_layer(z_index=1) l3 = _make_layer(z_index=2) result = sort_layers_by_z_index([l1, l2, l3]) assert result[0].z_index == 1 assert result[1].z_index == 2 assert result[2].z_index == 3 def test_same_z_index_stable(self): l1 = _make_layer(z_index=5) l2 = _make_layer(z_index=5) l3 = _make_layer(z_index=5) result = sort_layers_by_z_index([l1, l2, l3]) # 稳定排序,保持原顺序 assert result[0] is l1 assert result[1] is l2 assert result[2] is l3 def test_negative_z_index(self): l1 = _make_layer(z_index=-5) l2 = _make_layer(z_index=0) l3 = _make_layer(z_index=5) result = sort_layers_by_z_index([l3, l1, l2]) assert result[0].z_index == -5 assert result[2].z_index == 5 def test_empty_list(self): assert sort_layers_by_z_index([]) == [] def test_single_layer(self): layer = _make_layer(z_index=10) result = sort_layers_by_z_index([layer]) assert len(result) == 1 assert result[0] is layer