Files
xiaoxia-saas/tests/unit/test_concat_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

580 lines
20 KiB
Python
Executable File

"""concat_engine_pure 单元测试."""
from pathlib import Path
from apps.worker.video_processing.concat_engine_pure import (
build_concat_filter,
build_fps_filter,
build_scale_pad_filter,
build_setpts_filter,
build_single_segment_filter_chain,
calculate_scaled_size,
can_use_stream_copy,
count_valid_segments,
estimate_total_duration,
format_fps_filter,
generate_concat_file_list,
parse_fps,
resolve_output_params,
validate_concat_config,
validate_video_path,
)
# ── parse_fps ────────────────────────────────────────────────────────────────
class TestParseFps:
def test_none_returns_default(self):
assert parse_fps(None) == 30.0
def test_integer_value(self):
assert parse_fps(30) == 30.0
assert parse_fps(24) == 24.0
def test_float_value(self):
assert parse_fps(29.97) == 29.97
def test_string_integer(self):
assert parse_fps("30") == 30.0
assert parse_fps(" 60 ") == 60.0 # 带空格
def test_string_fraction(self):
assert parse_fps("30/1") == 30.0
assert abs(parse_fps("24000/1001") - 23.976) < 0.01
def test_zero_denominator(self):
assert parse_fps("30/0") == 30.0
def test_empty_string(self):
assert parse_fps("") == 30.0
assert parse_fps(" ") == 30.0
def test_invalid_string(self):
assert parse_fps("abc") == 30.0
assert parse_fps("30fps") == 30.0
def test_negative_fps(self):
assert parse_fps(-30) == -30.0
def test_zero_fps(self):
assert parse_fps(0) == 0.0
# ── format_fps_filter ───────────────────────────────────────────────────────
class TestFormatFpsFilter:
def test_integer_fps(self):
assert format_fps_filter(30.0) == "fps=30"
def test_near_integer_fps(self):
# 接近整数时用整数形式(注意:int(fps)是截断不是四舍五入)
assert format_fps_filter(30.0001) == "fps=30"
assert format_fps_filter(30.0005) == "fps=30" # int(30.0005)=30
def test_non_integer_fps(self):
result = format_fps_filter(23.976)
assert result.startswith("fps=")
assert "23.976" in result
def test_float_precision(self):
result = format_fps_filter(29.97)
assert result.startswith("fps=")
# 三位小数
parts = result.split("=")[1]
assert len(parts.split(".")[1]) == 3
def test_one_fps(self):
assert format_fps_filter(1.0) == "fps=1"
# ── resolve_output_params ───────────────────────────────────────────────────
class TestResolveOutputParams:
def test_config_specified(self):
w, h, fps = resolve_output_params(1920, 1080, 60.0)
assert w == 1920
assert h == 1080
assert fps == 60.0
def test_fallback_to_first_video_info(self):
info = {"width": 1280, "height": 720, "r_frame_rate": "24/1"}
w, h, fps = resolve_output_params(0, 0, 0, info)
assert w == 1280
assert h == 720
assert fps == 24.0
def test_fallback_to_defaults(self):
w, h, fps = resolve_output_params(0, 0, 0)
assert w == 1080 # default_width
assert h == 1920 # default_height
assert fps == 30.0
def test_partial_config(self):
# 宽度配置了,高度和帧率用探测的
info = {"width": 1280, "height": 720, "r_frame_rate": "24/1"}
w, h, fps = resolve_output_params(1920, 0, 0, info)
assert w == 1920
assert h == 720
assert fps == 24.0
def test_custom_defaults(self):
w, h, fps = resolve_output_params(
0,
0,
0,
default_width=640,
default_height=480,
default_fps=25.0,
)
assert w == 640
assert h == 480
assert fps == 25.0
def test_minimum_size(self):
w, h, fps = resolve_output_params(0, 0, 0, {"width": 0, "height": 0, "r_frame_rate": "0/1"})
assert w >= 1
assert h >= 1
assert fps >= 1.0
def test_fps_fraction_in_info(self):
info = {"width": 1920, "height": 1080, "r_frame_rate": "24000/1001"}
_, _, fps = resolve_output_params(0, 0, 0, info)
assert abs(fps - 23.976) < 0.01
# ── calculate_scaled_size ───────────────────────────────────────────────────
class TestCalculateScaledSize:
def test_same_ratio(self):
sw, sh, ox, oy = calculate_scaled_size(1920, 1080, 1920, 1080)
assert sw == 1920
assert sh == 1080
assert ox == 0
assert oy == 0
def test_wider_source_pad_top_bottom(self):
# 源是16:9,目标是9:16竖屏 → 上下填黑边
sw, sh, ox, oy = calculate_scaled_size(1920, 1080, 1080, 1920)
assert sw == 1080 # 以宽度为准
assert sh == 607 # 1080 * 1080 / 1920 = 607.5 → 607
assert ox == 0
assert oy > 0 # 垂直居中
def test_taller_source_pad_left_right(self):
# 源是9:16竖屏,目标是16:9横屏 → 左右填黑边
sw, sh, ox, oy = calculate_scaled_size(1080, 1920, 1920, 1080)
assert sh == 1080 # 以高度为准
assert sw == 607 # 1080 * 1080 / 1920 = 607.5 → 607
assert ox > 0 # 水平居中
assert oy == 0
def test_zero_source_size(self):
sw, sh, ox, oy = calculate_scaled_size(0, 0, 1920, 1080)
assert sw == 1920
assert sh == 1080
assert ox == 0
assert oy == 0
def test_negative_source_size(self):
sw, sh, ox, oy = calculate_scaled_size(-1, -1, 1920, 1080)
assert sw == 1920
assert sh == 1080
assert ox == 0
assert oy == 0
def test_target_same_ratio_different_size(self):
# 比例相同,尺寸不同 → 直接缩放到目标大小
sw, sh, ox, oy = calculate_scaled_size(640, 360, 1920, 1080)
assert sw == 1920
assert sh == 1080
assert ox == 0
assert oy == 0
# ── can_use_stream_copy ─────────────────────────────────────────────────────
class TestCanUseStreamCopy:
def test_force_reencode_false(self):
assert can_use_stream_copy([], 1920, 1080, 30.0, force_reencode=True) is False
def test_empty_segments(self):
assert can_use_stream_copy([], 1920, 1080, 30.0) is False
def test_single_segment_matching_params(self):
segs = [{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"}]
assert can_use_stream_copy(segs, 1920, 1080, 30.0) is True
def test_multiple_segments_same_params(self):
segs = [
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
]
assert can_use_stream_copy(segs, 1920, 1080, 30.0) is True
def test_different_codec(self):
segs = [
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
{"codec_name": "hevc", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
]
assert can_use_stream_copy(segs, 1920, 1080, 30.0) is False
def test_different_resolution(self):
segs = [
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
{"codec_name": "h264", "width": 1280, "height": 720, "r_frame_rate": "30/1"},
]
assert can_use_stream_copy(segs, 1920, 1080, 30.0) is False
def test_different_fps(self):
segs = [
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "60/1"},
]
assert can_use_stream_copy(segs, 1920, 1080, 30.0) is False
def test_target_differs_from_source(self):
segs = [{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"}]
# 目标分辨率不同
assert can_use_stream_copy(segs, 1280, 720, 30.0) is False
# 目标帧率不同
assert can_use_stream_copy(segs, 1920, 1080, 60.0) is False
def test_fps_fraction_match(self):
segs = [{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "24000/1001"}]
assert can_use_stream_copy(segs, 1920, 1080, 23.976) is True
# ── generate_concat_file_list ───────────────────────────────────────────────
class TestGenerateConcatFileList:
def test_single_file(self):
result = generate_concat_file_list(["/tmp/video.mp4"])
assert result == "file '/tmp/video.mp4'\n"
def test_multiple_files(self):
result = generate_concat_file_list(["/a.mp4", "/b.mp4", "/c.mp4"])
lines = result.strip().split("\n")
assert len(lines) == 3
assert lines[0] == "file '/a.mp4'"
assert lines[1] == "file '/b.mp4'"
assert lines[2] == "file '/c.mp4'"
assert result.endswith("\n")
def test_escapes_single_quotes(self):
result = generate_concat_file_list(["/path/with'quote.mp4"])
# 单引号转义: '\''
assert "'\\''" in result
def test_empty_list(self):
result = generate_concat_file_list([])
assert result == "\n"
def test_path_with_spaces(self):
result = generate_concat_file_list(["/path/to/video file.mp4"])
assert "file '/path/to/video file.mp4'" in result
# ── build_scale_pad_filter ──────────────────────────────────────────────────
class TestBuildScalePadFilter:
def test_basic_filter(self):
result = build_scale_pad_filter(1920, 1080)
assert "scale=1920:1080" in result
assert "force_original_aspect_ratio=decrease" in result
assert "pad=1920:1080" in result
assert "black" in result
assert "(ow-iw)/2" in result
assert "(oh-ih)/2" in result
def test_different_resolution(self):
result = build_scale_pad_filter(1080, 1920)
assert "scale=1080:1920" in result
assert "pad=1080:1920" in result
def test_ignores_source_size(self):
# src_w/src_h 目前不影响输出,都是用表达式
result1 = build_scale_pad_filter(1920, 1080)
result2 = build_scale_pad_filter(1920, 1080, src_w=1280, src_h=720)
assert result1 == result2
# ── build_fps_filter ────────────────────────────────────────────────────────
class TestBuildFpsFilter:
def test_integer_fps(self):
assert build_fps_filter(30.0) == "fps=30"
def test_float_fps(self):
result = build_fps_filter(29.97)
assert result.startswith("fps=")
# ── build_setpts_filter ─────────────────────────────────────────────────────
class TestBuildSetptsFilter:
def test_returns_correct_string(self):
assert build_setpts_filter() == "setpts=PTS-STARTPTS"
# ── build_concat_filter ─────────────────────────────────────────────────────
class TestBuildConcatFilter:
def test_zero_inputs(self):
assert build_concat_filter(0) == ""
def test_single_input_with_audio(self):
result = build_concat_filter(1)
assert "[0:v][0:a]" in result
assert "concat=n=1:v=1:a=1" in result
assert "[concat_v][concat_a]" in result
def test_single_input_no_audio(self):
result = build_concat_filter(1, has_audio=False)
assert "[0:v]" in result
assert "concat=n=1:v=1:a=0" in result
assert "[concat_v]" in result
assert "[concat_a]" not in result
def test_multiple_inputs_with_audio(self):
result = build_concat_filter(3)
assert "[0:v][0:a][1:v][1:a][2:v][2:a]" in result
assert "concat=n=3:v=1:a=1" in result
def test_multiple_inputs_no_audio(self):
result = build_concat_filter(3, has_audio=False)
assert "[0:v][1:v][2:v]" in result
assert "concat=n=3:v=1:a=0" in result
def test_negative_inputs(self):
assert build_concat_filter(-1) == ""
# ── build_single_segment_filter_chain ───────────────────────────────────────
class TestBuildSingleSegmentFilterChain:
def test_with_audio(self):
result = build_single_segment_filter_chain(1920, 1080, 30.0, 0)
# 视频链
assert "[0:v]" in result
assert "[v0]" in result
assert "scale=1920:1080" in result
assert "fps=30" in result
assert "setpts=PTS-STARTPTS" in result
# 音频链
assert "[0:a]" in result
assert "[a0]" in result
assert "asetpts=PTS-STARTPTS" in result
# 用分号分隔
assert ";" in result
def test_without_audio(self):
result = build_single_segment_filter_chain(1920, 1080, 30.0, 2, has_audio=False)
assert "[2:v]" in result
assert "[v2]" in result
assert "[2:a]" not in result
assert ";" not in result # 没有音频就没有分号
def test_segment_index_propagated(self):
for idx in [0, 5, 10]:
result = build_single_segment_filter_chain(1920, 1080, 30.0, idx)
assert f"[{idx}:v]" in result
assert f"[v{idx}]" in result
# ── validate_concat_config ──────────────────────────────────────────────────
class TestValidateConcatConfig:
def test_valid_config(self):
config = {
"segments": [
{"video_path": "/a.mp4"},
{"video_path": "/b.mp4"},
],
"output_width": 1920,
"output_height": 1080,
"output_fps": 30,
}
valid, errors = validate_concat_config(config)
assert valid is True
assert errors == []
def test_no_segments(self):
valid, errors = validate_concat_config({})
assert valid is False
assert any("至少需要一个" in e for e in errors)
def test_empty_segments(self):
valid, errors = validate_concat_config({"segments": []})
assert valid is False
assert len(errors) >= 1
def test_missing_video_path(self):
config = {"segments": [{"video_path": ""}]}
valid, errors = validate_concat_config(config)
assert valid is False
assert any("video_path" in e for e in errors)
def test_multiple_missing_paths(self):
config = {
"segments": [
{"video_path": "/a.mp4"},
{"video_path": ""},
{"video_path": ""},
]
}
valid, errors = validate_concat_config(config)
assert valid is False
path_errors = [e for e in errors if "video_path" in e]
assert len(path_errors) == 2
def test_negative_output_width(self):
config = {"segments": [{"video_path": "/a.mp4"}], "output_width": -1}
valid, errors = validate_concat_config(config)
assert valid is False
assert any("output_width" in e for e in errors)
def test_negative_output_height(self):
config = {"segments": [{"video_path": "/a.mp4"}], "output_height": -1}
valid, errors = validate_concat_config(config)
assert valid is False
assert any("output_height" in e for e in errors)
def test_negative_output_fps(self):
config = {"segments": [{"video_path": "/a.mp4"}], "output_fps": -1}
valid, errors = validate_concat_config(config)
assert valid is False
assert any("output_fps" in e for e in errors)
def test_zero_output_params_valid(self):
# 0值表示未指定,是合法的
config = {
"segments": [{"video_path": "/a.mp4"}],
"output_width": 0,
"output_height": 0,
"output_fps": 0,
}
valid, errors = validate_concat_config(config)
assert valid is True
# ── validate_video_path ─────────────────────────────────────────────────────
class TestValidateVideoPath:
def test_empty_path(self):
valid, err = validate_video_path("", "/work")
assert valid is False
assert "不能为空" in err
def test_relative_path_valid(self):
valid, err = validate_video_path("video.mp4", "/work")
assert valid is True
assert err == ""
def test_relative_path_with_subdir(self):
valid, err = validate_video_path("sub/video.mp4", "/work")
assert valid is True
def test_path_traversal_rejected(self):
valid, err = validate_video_path("../secret.mp4", "/work")
assert valid is False
assert ".." in err
def test_nested_path_traversal_rejected(self):
valid, err = validate_video_path("sub/../../secret.mp4", "/work")
assert valid is False
def test_absolute_path_inside_workdir(self):
valid, err = validate_video_path("/work/sub/video.mp4", "/work")
assert valid is True
def test_absolute_path_outside_workdir(self):
valid, err = validate_video_path("/etc/passwd", "/work")
assert valid is False
assert "工作目录内" in err
def test_path_object_input(self):
valid, err = validate_video_path(Path("video.mp4"), Path("/work"))
assert valid is True
# ── estimate_total_duration ─────────────────────────────────────────────────
class TestEstimateTotalDuration:
def test_single_segment(self):
assert estimate_total_duration([{"duration": 10.5}]) == 10.5
def test_multiple_segments(self):
segs = [
{"duration": 10},
{"duration": 20.5},
{"duration": 5.5},
]
assert estimate_total_duration(segs) == 36.0
def test_empty_list(self):
assert estimate_total_duration([]) == 0.0
def test_missing_duration_field(self):
segs = [{"path": "a.mp4"}, {"duration": 10}]
assert estimate_total_duration(segs) == 10.0
def test_invalid_duration_skipped(self):
segs = [
{"duration": 10},
{"duration": "abc"},
{"duration": 20},
]
assert estimate_total_duration(segs) == 30.0
def test_string_duration(self):
segs = [{"duration": "15.5"}]
assert estimate_total_duration(segs) == 15.5
def test_negative_duration(self):
segs = [{"duration": -5}]
assert estimate_total_duration(segs) == -5.0
# ── count_valid_segments ────────────────────────────────────────────────────
class TestCountValidSegments:
def test_all_valid(self):
segs = [
{"video_path": "/a.mp4"},
{"video_path": "/b.mp4"},
]
assert count_valid_segments(segs) == 2
def test_some_invalid(self):
segs = [
{"video_path": "/a.mp4"},
{"video_path": ""},
{"video_path": "/c.mp4"},
]
assert count_valid_segments(segs) == 2
def test_none_valid(self):
segs = [
{"video_path": ""},
{"other_field": "x"},
]
assert count_valid_segments(segs) == 0
def test_empty_list(self):
assert count_valid_segments([]) == 0