Files
xiaoxia-saas/tests/unit/test_pip_engine_pure.py
xiaoxia 77a49e3365
CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled
CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled
CI/CD Pipeline / PR Build API Image (push) Has been cancelled
CI/CD Pipeline / PR Build Web Image (push) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
CI/CD Pipeline / CI Gate (push) Has been cancelled
test(wave201): concat_engine_pure 单测补全 +85测 (#1167)
2026-07-30 00:25:42 +08:00

681 lines
25 KiB
Python
Executable File

"""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