xiaoxia
8a2d2df3cd
CI/CD Pipeline / Validate Code Quality And Tests (push) Successful in 30s
CI/CD Pipeline / Frontend Lint (push) Successful in 1m12s
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Integration Tests (push) Successful in 1m20s
CI/CD Pipeline / Unit Tests (push) Successful in 4m17s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Successful in 17m1s
CI/CD Pipeline / Staging E2E Tests (push) Failing after 48s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 3m16s
Squash merge PR #305
861 lines
30 KiB
Python
Executable File
861 lines
30 KiB
Python
Executable File
"""封面生成 + 视频倒放 + 贴纸叠加 单元测试.
|
||
|
||
覆盖三个新渲染能力的核心场景和降级逻辑。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
from video_processing.cover_generator import (
|
||
DEFAULT_COVER_HEIGHT,
|
||
DEFAULT_COVER_WIDTH,
|
||
CoverGenerator,
|
||
generate_cover_from_plan,
|
||
)
|
||
from video_processing.reverse_engine import ReverseConfig, ReverseEngine
|
||
from video_processing.sticker_engine import (
|
||
POSITION_PRESETS,
|
||
STICKER_CATEGORIES,
|
||
ImageStickerConfig,
|
||
StickerEngine,
|
||
TextStickerConfig,
|
||
get_sticker_categories,
|
||
parse_stickers_from_config,
|
||
)
|
||
from video_processing.unified_render_service import (
|
||
ResolvedClip,
|
||
UnifiedRenderService,
|
||
)
|
||
|
||
# ── Fixtures ──────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@dataclass
|
||
class FakePlan:
|
||
"""模拟 EditPlan."""
|
||
|
||
id: str = "plan_001"
|
||
name: str = "测试计划"
|
||
config: dict[str, Any] = field(default_factory=dict)
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_video(tmp_path):
|
||
"""创建一个测试视频文件（空文件，仅用于路径测试）."""
|
||
video_path = tmp_path / "test_video.mp4"
|
||
video_path.write_bytes(b"fake video data")
|
||
return video_path
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_image(tmp_path):
|
||
"""创建一个测试图片文件."""
|
||
img_path = tmp_path / "sticker.png"
|
||
img_path.write_bytes(b"fake png data")
|
||
return img_path
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# 一、视频倒放引擎测试
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestReverseConfig:
|
||
"""ReverseConfig 配置解析测试."""
|
||
|
||
def test_default_disabled(self):
|
||
"""默认配置为关闭."""
|
||
config = ReverseConfig.from_dict(None)
|
||
assert config.enabled is False
|
||
assert config.reverse_video is True
|
||
assert config.reverse_audio is True
|
||
|
||
def test_empty_dict(self):
|
||
"""空字典视为关闭."""
|
||
config = ReverseConfig.from_dict({})
|
||
assert config.enabled is False
|
||
|
||
def test_enabled(self):
|
||
"""启用倒放."""
|
||
config = ReverseConfig.from_dict({"enabled": True})
|
||
assert config.enabled is True
|
||
assert config.reverse_video is True
|
||
assert config.reverse_audio is True
|
||
|
||
def test_video_only(self):
|
||
"""只倒放视频."""
|
||
config = ReverseConfig.from_dict(
|
||
{
|
||
"enabled": True,
|
||
"reverse_video": True,
|
||
"reverse_audio": False,
|
||
}
|
||
)
|
||
assert config.enabled is True
|
||
assert config.reverse_video is True
|
||
assert config.reverse_audio is False
|
||
|
||
def test_audio_only(self):
|
||
"""只倒放音频."""
|
||
config = ReverseConfig.from_dict(
|
||
{
|
||
"enabled": True,
|
||
"reverse_video": False,
|
||
"reverse_audio": True,
|
||
}
|
||
)
|
||
assert config.reverse_video is False
|
||
assert config.reverse_audio is True
|
||
|
||
def test_invalid_config_fallback(self):
|
||
"""无效配置降级为默认."""
|
||
config = ReverseConfig.from_dict("invalid") # type: ignore
|
||
assert config.enabled is False
|
||
|
||
def test_none_config(self):
|
||
"""None 配置."""
|
||
config = ReverseConfig.from_dict(None)
|
||
assert config.enabled is False
|
||
|
||
|
||
class TestReverseEngine:
|
||
"""ReverseEngine 滤镜生成测试."""
|
||
|
||
def test_video_reverse_filter(self):
|
||
"""视频倒放滤镜生成."""
|
||
config = ReverseConfig(enabled=True, reverse_video=True)
|
||
f = ReverseEngine.build_video_filter(config, duration=10.0)
|
||
assert f == "reverse"
|
||
|
||
def test_video_disabled(self):
|
||
"""视频倒放关闭时返回空."""
|
||
config = ReverseConfig(enabled=False)
|
||
f = ReverseEngine.build_video_filter(config, duration=10.0)
|
||
assert f == ""
|
||
|
||
def test_video_disabled_flag(self):
|
||
"""启用但 reverse_video=False."""
|
||
config = ReverseConfig(enabled=True, reverse_video=False)
|
||
f = ReverseEngine.build_video_filter(config, duration=10.0)
|
||
assert f == ""
|
||
|
||
def test_audio_reverse_filter(self):
|
||
"""音频倒放滤镜生成."""
|
||
config = ReverseConfig(enabled=True, reverse_audio=True)
|
||
f = ReverseEngine.build_audio_filter(config, duration=10.0)
|
||
assert f == "areverse"
|
||
|
||
def test_audio_disabled(self):
|
||
"""音频倒放关闭."""
|
||
config = ReverseConfig(enabled=False)
|
||
f = ReverseEngine.build_audio_filter(config, duration=10.0)
|
||
assert f == ""
|
||
|
||
def test_long_video_safety_limit(self):
|
||
"""超长视频安全限制：跳过倒放."""
|
||
config = ReverseConfig(enabled=True)
|
||
f = ReverseEngine.build_video_filter(config, duration=200.0)
|
||
assert f == "" # 超过 MAX_SAFE_DURATION
|
||
|
||
def test_long_audio_safety_limit(self):
|
||
"""超长音频安全限制."""
|
||
config = ReverseConfig(enabled=True)
|
||
f = ReverseEngine.build_audio_filter(config, duration=200.0)
|
||
assert f == ""
|
||
|
||
def test_duration_zero(self):
|
||
"""时长为0时正常返回."""
|
||
config = ReverseConfig(enabled=True)
|
||
f = ReverseEngine.build_video_filter(config, duration=0.0)
|
||
assert f == "reverse"
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# 二、贴纸引擎测试
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestStickerPosition:
|
||
"""贴纸位置计算测试."""
|
||
|
||
def test_presets_exist(self):
|
||
"""9宫格预设存在."""
|
||
assert "top_left" in POSITION_PRESETS
|
||
assert "center" in POSITION_PRESETS
|
||
assert "bottom_right" in POSITION_PRESETS
|
||
assert len(POSITION_PRESETS) == 9
|
||
|
||
def test_resolve_position_center(self):
|
||
"""居中位置计算."""
|
||
sticker = ImageStickerConfig(position="center")
|
||
x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 200, 200)
|
||
assert abs(x - 400) < 1 # (1000-200)/2 = 400
|
||
assert abs(y - 400) < 1
|
||
|
||
def test_resolve_position_top_left(self):
|
||
"""左上角位置."""
|
||
sticker = ImageStickerConfig(position="top_left")
|
||
x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 100, 100)
|
||
assert x == 0 # 0.05*1000 - 50 = 0 (clamped)
|
||
assert y == 0
|
||
|
||
def test_custom_position_percent(self):
|
||
"""自定义百分比位置."""
|
||
sticker = ImageStickerConfig(
|
||
position="center",
|
||
x=30.0,
|
||
y=70.0,
|
||
x_unit="percent",
|
||
y_unit="percent",
|
||
)
|
||
x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 100, 100)
|
||
assert abs(x - 250) < 1 # 300 - 50 = 250
|
||
assert abs(y - 650) < 1 # 700 - 50 = 650
|
||
|
||
def test_custom_position_pixel(self):
|
||
"""自定义像素位置."""
|
||
sticker = ImageStickerConfig(
|
||
position="center",
|
||
x=100.0,
|
||
y=200.0,
|
||
x_unit="pixel",
|
||
y_unit="pixel",
|
||
)
|
||
x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 50, 50)
|
||
assert abs(x - 75) < 1 # 100 - 25 = 75
|
||
assert abs(y - 175) < 1 # 200 - 25 = 175
|
||
|
||
def test_position_clamped(self):
|
||
"""位置钳制在画布内."""
|
||
sticker = ImageStickerConfig(
|
||
position="center",
|
||
x=-10.0,
|
||
y=-10.0,
|
||
x_unit="pixel",
|
||
y_unit="pixel",
|
||
)
|
||
x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 50, 50)
|
||
assert x >= 0
|
||
assert y >= 0
|
||
|
||
|
||
class TestTextSticker:
|
||
"""文字贴纸测试."""
|
||
|
||
def test_drawtext_filter_basic(self):
|
||
"""基础文字贴纸滤镜生成."""
|
||
sticker = TextStickerConfig(
|
||
enabled=True,
|
||
text="Hello World",
|
||
font_size=36,
|
||
font_color="#FFFFFF",
|
||
position="center",
|
||
)
|
||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||
assert "drawtext" in f
|
||
assert "Hello World" in f
|
||
assert "fontsize=36" in f
|
||
assert "[in]" in f
|
||
assert "[out]" in f
|
||
|
||
def test_drawtext_with_stroke(self):
|
||
"""带描边的文字贴纸."""
|
||
sticker = TextStickerConfig(
|
||
enabled=True,
|
||
text="Test",
|
||
stroke_width=3,
|
||
stroke_color="#FF0000",
|
||
)
|
||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||
assert "borderw=3" in f
|
||
assert "bordercolor=#FF0000" in f
|
||
|
||
def test_drawtext_with_shadow(self):
|
||
"""带阴影的文字贴纸."""
|
||
sticker = TextStickerConfig(
|
||
enabled=True,
|
||
text="Shadow",
|
||
shadow_x=4,
|
||
shadow_y=4,
|
||
shadow_alpha=0.5,
|
||
)
|
||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||
assert "shadowx=4" in f
|
||
assert "shadowy=4" in f
|
||
|
||
def test_drawtext_time_range(self):
|
||
"""带时间范围的文字贴纸."""
|
||
sticker = TextStickerConfig(
|
||
enabled=True,
|
||
text="Timed",
|
||
start_time=2.0,
|
||
duration=3.0,
|
||
)
|
||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||
assert "enable='between(t,2.0,5.0)'" in f
|
||
|
||
def test_drawtext_empty_text(self):
|
||
"""空文字直通."""
|
||
sticker = TextStickerConfig(enabled=True, text="")
|
||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||
assert "[in]copy[out]" in f
|
||
|
||
def test_drawtext_with_fade(self):
|
||
"""带淡入淡出的文字贴纸."""
|
||
sticker = TextStickerConfig(
|
||
enabled=True,
|
||
text="Fade",
|
||
start_time=1.0,
|
||
duration=5.0,
|
||
fade_in=0.5,
|
||
fade_out=0.5,
|
||
)
|
||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||
assert "alpha=" in f
|
||
|
||
|
||
class TestImageSticker:
|
||
"""图片贴纸测试."""
|
||
|
||
def test_image_sticker_overlay(self, sample_image):
|
||
"""图片贴纸 overlay 滤镜生成."""
|
||
result = StickerEngine.build_sticker_chain(
|
||
stickers=[
|
||
{
|
||
"type": "image",
|
||
"image_path": str(sample_image),
|
||
"position": "top_right",
|
||
"scale": 0.5,
|
||
"opacity": 0.8,
|
||
"z_index": 10,
|
||
}
|
||
],
|
||
input_label="[base]",
|
||
output_label="[final]",
|
||
canvas_w=1080,
|
||
canvas_h=1920,
|
||
)
|
||
assert result.filter_str != ""
|
||
assert "overlay" in result.filter_str
|
||
assert len(result.extra_inputs) == 1
|
||
assert result.extra_inputs[0] == str(sample_image)
|
||
|
||
def test_image_sticker_missing_file(self):
|
||
"""图片贴纸素材不存在时跳过."""
|
||
result = StickerEngine.build_sticker_chain(
|
||
stickers=[
|
||
{
|
||
"type": "image",
|
||
"image_path": "/nonexistent/image.png",
|
||
"position": "center",
|
||
}
|
||
],
|
||
input_label="[in]",
|
||
output_label="[out]",
|
||
canvas_w=1080,
|
||
canvas_h=1920,
|
||
)
|
||
# 素材不存在，跳过，返回直通
|
||
assert "[in]copy[out]" in result.filter_str
|
||
assert len(result.extra_inputs) == 0
|
||
|
||
def test_mixed_stickers(self, sample_image):
|
||
"""混合贴纸：图片 + 文字."""
|
||
result = StickerEngine.build_sticker_chain(
|
||
stickers=[
|
||
{
|
||
"type": "image",
|
||
"image_path": str(sample_image),
|
||
"position": "top_left",
|
||
"z_index": 5,
|
||
},
|
||
{
|
||
"type": "text",
|
||
"text": "Hello",
|
||
"position": "bottom_center",
|
||
"z_index": 10,
|
||
},
|
||
],
|
||
input_label="[in]",
|
||
output_label="[out]",
|
||
canvas_w=1080,
|
||
canvas_h=1920,
|
||
)
|
||
assert "overlay" in result.filter_str
|
||
assert "drawtext" in result.filter_str
|
||
assert len(result.extra_inputs) == 1
|
||
|
||
def test_sticker_z_index_order(self, sample_image):
|
||
"""贴纸按 z_index 排序."""
|
||
result = StickerEngine.build_sticker_chain(
|
||
stickers=[
|
||
{"type": "text", "text": "Top", "z_index": 20, "position": "center"},
|
||
{"type": "text", "text": "Bottom", "z_index": 5, "position": "center"},
|
||
],
|
||
input_label="[in]",
|
||
output_label="[out]",
|
||
canvas_w=1080,
|
||
canvas_h=1920,
|
||
)
|
||
# z_index 小的先叠加，大的后叠加（在上面）
|
||
assert result.filter_str.count("drawtext") == 2
|
||
|
||
def test_empty_stickers(self):
|
||
"""空贴纸列表."""
|
||
result = StickerEngine.build_sticker_chain(
|
||
stickers=[],
|
||
input_label="[in]",
|
||
output_label="[out]",
|
||
canvas_w=1080,
|
||
canvas_h=1920,
|
||
)
|
||
assert "[in]copy[out]" in result.filter_str
|
||
assert result.extra_inputs == []
|
||
|
||
def test_invalid_sticker_skipped(self):
|
||
"""无效贴纸配置跳过."""
|
||
result = StickerEngine.build_sticker_chain(
|
||
stickers=[{"invalid": "data"}],
|
||
input_label="[in]",
|
||
output_label="[out]",
|
||
canvas_w=1080,
|
||
canvas_h=1920,
|
||
)
|
||
# 解析失败，跳过，直通
|
||
assert "[in]copy[out]" in result.filter_str
|
||
|
||
|
||
class TestStickerHelpers:
|
||
"""贴纸辅助函数测试."""
|
||
|
||
def test_parse_stickers_empty(self):
|
||
"""空配置解析."""
|
||
assert parse_stickers_from_config(None) == []
|
||
assert parse_stickers_from_config({}) == []
|
||
|
||
def test_parse_stickers_list(self):
|
||
"""正常贴纸列表解析."""
|
||
config = {"stickers": [{"type": "text", "text": "A"}, {"type": "text", "text": "B"}]}
|
||
result = parse_stickers_from_config(config)
|
||
assert len(result) == 2
|
||
|
||
def test_parse_stickers_not_list(self):
|
||
"""非列表类型返回空."""
|
||
config = {"stickers": "not a list"}
|
||
assert parse_stickers_from_config(config) == []
|
||
|
||
def test_get_categories(self):
|
||
"""贴纸分类列表."""
|
||
cats = get_sticker_categories()
|
||
assert len(cats) == len(STICKER_CATEGORIES)
|
||
assert cats[0][0] == "emoji"
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# 三、封面生成器测试
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
|
||
class TestCoverGenerator:
|
||
"""CoverGenerator 测试."""
|
||
|
||
def test_default_dimensions(self):
|
||
"""默认封面尺寸."""
|
||
assert DEFAULT_COVER_WIDTH == 1080
|
||
assert DEFAULT_COVER_HEIGHT == 1920
|
||
|
||
@patch("video_processing.cover_generator.run_ffmpeg")
|
||
@patch("video_processing.cover_generator.probe_video_info")
|
||
def test_extract_frame_basic(self, mock_probe, mock_run, sample_video, tmp_path):
|
||
"""基础抽帧测试."""
|
||
mock_probe.return_value = {"duration": 30.0}
|
||
|
||
# mock run_ffmpeg 实际创建输出文件
|
||
def fake_run_ffmpeg(cmd):
|
||
# 找到输出路径并创建文件
|
||
output_path = Path(cmd[-1])
|
||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||
output_path.write_bytes(b"fake jpeg data")
|
||
|
||
mock_run.side_effect = fake_run_ffmpeg
|
||
|
||
output = tmp_path / "cover.jpg"
|
||
result = CoverGenerator.extract_frame(
|
||
sample_video,
|
||
output,
|
||
time_sec=2.0,
|
||
)
|
||
|
||
assert result == output
|
||
mock_run.assert_called_once()
|
||
# 检查命令参数
|
||
cmd = mock_run.call_args[0][0]
|
||
assert "-ss" in cmd
|
||
assert "2.000" in cmd
|
||
assert "-vframes" in cmd
|
||
assert "1" in cmd
|
||
|
||
@patch("video_processing.cover_generator.run_ffmpeg")
|
||
@patch("video_processing.cover_generator.probe_video_info")
|
||
def test_extract_frame_time_clamped(self, mock_probe, mock_run, sample_video, tmp_path):
|
||
"""抽帧时间超过视频长度时钳制."""
|
||
mock_probe.return_value = {"duration": 10.0}
|
||
|
||
def fake_run_ffmpeg(cmd):
|
||
output_path = Path(cmd[-1])
|
||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||
output_path.write_bytes(b"fake jpeg data")
|
||
|
||
mock_run.side_effect = fake_run_ffmpeg
|
||
|
||
output = tmp_path / "cover.jpg"
|
||
CoverGenerator.extract_frame(
|
||
sample_video,
|
||
output,
|
||
time_sec=100.0, # 超过视频时长
|
||
)
|
||
|
||
cmd = mock_run.call_args[0][0]
|
||
ss_idx = cmd.index("-ss")
|
||
time_val = float(cmd[ss_idx + 1])
|
||
# 应该被钳制到中间帧（5秒左右）
|
||
assert time_val <= 10.0
|
||
|
||
@patch("video_processing.cover_generator.run_ffmpeg")
|
||
@patch("video_processing.cover_generator.probe_video_info")
|
||
def test_extract_frame_negative_time(self, mock_probe, mock_run, sample_video, tmp_path):
|
||
"""负时间钳制到0."""
|
||
mock_probe.return_value = {"duration": 30.0}
|
||
|
||
def fake_run_ffmpeg(cmd):
|
||
output_path = Path(cmd[-1])
|
||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||
output_path.write_bytes(b"fake jpeg data")
|
||
|
||
mock_run.side_effect = fake_run_ffmpeg
|
||
|
||
output = tmp_path / "cover.jpg"
|
||
CoverGenerator.extract_frame(
|
||
sample_video,
|
||
output,
|
||
time_sec=-5.0,
|
||
)
|
||
|
||
cmd = mock_run.call_args[0][0]
|
||
ss_idx = cmd.index("-ss")
|
||
time_val = float(cmd[ss_idx + 1])
|
||
assert time_val >= 0
|
||
|
||
def test_extract_frame_file_not_found(self, tmp_path):
|
||
"""视频文件不存在抛异常."""
|
||
with pytest.raises(FileNotFoundError):
|
||
CoverGenerator.extract_frame(
|
||
"/nonexistent/video.mp4",
|
||
tmp_path / "cover.jpg",
|
||
)
|
||
|
||
@patch("video_processing.cover_generator.CoverGenerator.extract_frame")
|
||
@patch("video_processing.cover_generator.probe_video_info")
|
||
def test_smart_cover_3_frames(self, mock_probe, mock_extract, sample_video, tmp_path):
|
||
"""智能封面抽取3帧选最佳."""
|
||
mock_probe.return_value = {"duration": 30.0}
|
||
|
||
# 创建三个大小不同的临时文件（模拟清晰度不同）
|
||
def create_frame(video_path, output_path, **kwargs):
|
||
# 第二帧最大（最清晰）
|
||
p = Path(output_path)
|
||
p.parent.mkdir(parents=True, exist_ok=True)
|
||
if "candidate_1" in str(p):
|
||
p.write_bytes(b"x" * 10000) # 最大 = 最清晰
|
||
elif "candidate_0" in str(p):
|
||
p.write_bytes(b"x" * 1000)
|
||
else:
|
||
p.write_bytes(b"x" * 5000)
|
||
return p
|
||
|
||
mock_extract.side_effect = create_frame
|
||
|
||
output = tmp_path / "smart_cover.jpg"
|
||
result = CoverGenerator.extract_smart_cover(
|
||
sample_video,
|
||
output,
|
||
frame_count=3,
|
||
)
|
||
|
||
assert result == output
|
||
assert output.exists()
|
||
# 应该选最大的那个文件（candidate_1）
|
||
assert output.stat().st_size == 10000
|
||
|
||
@patch("video_processing.cover_generator.CoverGenerator.extract_frame")
|
||
@patch("video_processing.cover_generator.probe_video_info")
|
||
def test_smart_cover_fallback(self, mock_probe, mock_extract, sample_video, tmp_path):
|
||
"""智能封面全部失败时降级."""
|
||
mock_probe.return_value = {"duration": 0.0} # 时长为0
|
||
|
||
output = tmp_path / "cover.jpg"
|
||
output.write_bytes(b"x" * 100)
|
||
mock_extract.return_value = output
|
||
|
||
result = CoverGenerator.extract_smart_cover(sample_video, output, frame_count=3)
|
||
assert result == output
|
||
|
||
@patch("video_processing.cover_generator.run_ffmpeg")
|
||
def test_custom_cover(self, mock_run, sample_image, tmp_path):
|
||
"""自定义封面处理."""
|
||
output = tmp_path / "custom_cover.jpg"
|
||
|
||
result = CoverGenerator.process_custom_cover(
|
||
sample_image,
|
||
output,
|
||
)
|
||
|
||
assert result == output
|
||
mock_run.assert_called_once()
|
||
cmd = mock_run.call_args[0][0]
|
||
assert str(sample_image) in cmd
|
||
|
||
def test_custom_cover_not_found(self, tmp_path):
|
||
"""自定义封面文件不存在."""
|
||
with pytest.raises(FileNotFoundError):
|
||
CoverGenerator.process_custom_cover(
|
||
"/nonexistent/img.png",
|
||
tmp_path / "cover.jpg",
|
||
)
|
||
|
||
@patch("video_processing.cover_generator.CoverGenerator.extract_frame")
|
||
def test_generate_cover_time_mode(self, mock_extract, sample_video, tmp_path):
|
||
"""统一入口 - time 模式."""
|
||
output = tmp_path / "cover.jpg"
|
||
mock_extract.return_value = output
|
||
|
||
result = CoverGenerator.generate_cover(
|
||
sample_video,
|
||
output,
|
||
mode="time",
|
||
time_sec=3.0,
|
||
)
|
||
|
||
assert result == output
|
||
mock_extract.assert_called_once()
|
||
|
||
@patch("video_processing.cover_generator.CoverGenerator.extract_smart_cover")
|
||
def test_generate_cover_smart_mode(self, mock_smart, sample_video, tmp_path):
|
||
"""统一入口 - smart 模式."""
|
||
output = tmp_path / "cover.jpg"
|
||
mock_smart.return_value = output
|
||
|
||
result = CoverGenerator.generate_cover(
|
||
sample_video,
|
||
output,
|
||
mode="smart",
|
||
)
|
||
|
||
assert result == output
|
||
mock_smart.assert_called_once()
|
||
|
||
@patch("video_processing.cover_generator.CoverGenerator.process_custom_cover")
|
||
def test_generate_cover_custom_mode(self, mock_custom, sample_video, sample_image, tmp_path):
|
||
"""统一入口 - custom 模式."""
|
||
output = tmp_path / "cover.jpg"
|
||
mock_custom.return_value = output
|
||
|
||
result = CoverGenerator.generate_cover(
|
||
sample_video,
|
||
output,
|
||
mode="custom",
|
||
custom_image=sample_image,
|
||
)
|
||
|
||
assert result == output
|
||
mock_custom.assert_called_once()
|
||
|
||
|
||
class TestGenerateCoverFromPlan:
|
||
"""从 plan 配置生成封面测试."""
|
||
|
||
@patch("video_processing.cover_generator.CoverGenerator.extract_smart_cover")
|
||
def test_smart_mode_from_plan(self, mock_smart, sample_video, tmp_path):
|
||
"""plan 配置 smart 模式."""
|
||
plan = FakePlan(id="plan_001", config={"cover_config": {"mode": "smart"}})
|
||
mock_smart.return_value = tmp_path / "cover.jpg"
|
||
(tmp_path / "cover.jpg").write_bytes(b"test")
|
||
|
||
result = generate_cover_from_plan(plan, sample_video, tmp_path)
|
||
assert result is not None
|
||
|
||
def test_no_cover_config(self, sample_video, tmp_path):
|
||
"""没有封面配置时返回 None."""
|
||
plan = FakePlan(id="plan_001", config={})
|
||
result = generate_cover_from_plan(plan, sample_video, tmp_path)
|
||
assert result is None
|
||
|
||
def test_none_config(self, sample_video, tmp_path):
|
||
"""config 为 None."""
|
||
plan = FakePlan(id="plan_001", config=None) # type: ignore
|
||
result = generate_cover_from_plan(plan, sample_video, tmp_path)
|
||
assert result is None
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# 四、UnifiedRenderService 集成测试
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
|
||
def _make_clip(clip_id="c1", asset_id="a1", path=Path("/fake/video.mp4"), clip_type="main", config=None):
|
||
"""创建测试用 ResolvedClip."""
|
||
return ResolvedClip(
|
||
clip_id=clip_id,
|
||
asset_id=asset_id,
|
||
local_path=path,
|
||
clip_type=clip_type,
|
||
order=0,
|
||
start_time=0.0,
|
||
duration=0.0,
|
||
transition_effect="cut",
|
||
config=config or {},
|
||
actual_duration=10.0,
|
||
)
|
||
|
||
|
||
def _make_service(plan, clips, asset_path_map=None, work_dir=None, tmp_path=None):
|
||
"""创建测试用 UnifiedRenderService."""
|
||
from pathlib import Path as P
|
||
|
||
work_dir = work_dir or (tmp_path or P("/tmp")) / "render_test"
|
||
work_dir.mkdir(exist_ok=True, parents=True)
|
||
return UnifiedRenderService(
|
||
plan=plan,
|
||
clips=clips,
|
||
asset_path_map=asset_path_map or {},
|
||
work_dir=work_dir,
|
||
output_width=1080,
|
||
output_height=1920,
|
||
output_fps=30,
|
||
transition_duration=0.5,
|
||
)
|
||
|
||
|
||
class TestReverseIntegration:
|
||
"""倒放功能集成测试."""
|
||
|
||
@patch("video_processing.unified_render_service.probe_video_info")
|
||
@patch("video_processing.unified_render_service.run_ffmpeg")
|
||
def test_reverse_in_filter_complex(self, mock_run, mock_probe, tmp_path):
|
||
"""filter_complex 路径中包含倒放滤镜."""
|
||
mock_probe.return_value = {"duration": 10.0, "has_audio": True, "width": 1920, "height": 1080}
|
||
mock_run.return_value = None
|
||
|
||
plan = FakePlan(id="p1")
|
||
clip = _make_clip(config={"reverse": {"enabled": True}})
|
||
clip.actual_duration = 5.0
|
||
# 两个 clip 触发 filter_complex 路径
|
||
clip2 = _make_clip(clip_id="c2", config={})
|
||
clip2.actual_duration = 5.0
|
||
clip2.order = 1
|
||
|
||
service = _make_service(plan, [clip, clip2], tmp_path=tmp_path)
|
||
|
||
# 直接测 _build_filter_complex
|
||
from video_processing.unified_render_service import RenderLayer
|
||
|
||
layer = RenderLayer(role="main", clips=[clip, clip2])
|
||
filter_str, inputs = service._build_filter_complex([layer])
|
||
|
||
assert "reverse" in filter_str
|
||
|
||
def test_can_use_pass_through_with_reverse(self, tmp_path):
|
||
"""倒放不影响直通模式判断（只有贴纸才禁用）."""
|
||
plan = FakePlan(id="p1")
|
||
clip = _make_clip(config={"reverse": {"enabled": True}})
|
||
clip.actual_duration = 5.0
|
||
|
||
service = _make_service(plan, [clip], tmp_path=tmp_path)
|
||
from video_processing.unified_render_service import RenderLayer
|
||
|
||
layer = RenderLayer(role="main", clips=[clip])
|
||
layers = [layer]
|
||
|
||
assert service._can_use_pass_through(layers) is True
|
||
|
||
|
||
class TestStickerIntegration:
|
||
"""贴纸功能集成测试."""
|
||
|
||
def test_can_use_pass_through_with_stickers(self, tmp_path):
|
||
"""有贴纸时禁用直通模式."""
|
||
plan = FakePlan(id="p1", config={"stickers": [{"type": "text", "text": "Hello", "position": "center"}]})
|
||
clip = _make_clip()
|
||
clip.actual_duration = 5.0
|
||
|
||
service = _make_service(plan, [clip], tmp_path=tmp_path)
|
||
from video_processing.unified_render_service import RenderLayer
|
||
|
||
layer = RenderLayer(role="main", clips=[clip])
|
||
layers = [layer]
|
||
|
||
assert service._can_use_pass_through(layers) is False
|
||
|
||
def test_can_use_pass_through_no_stickers(self, tmp_path):
|
||
"""无贴纸时直通模式正常."""
|
||
plan = FakePlan(id="p1", config={})
|
||
clip = _make_clip()
|
||
clip.actual_duration = 5.0
|
||
|
||
service = _make_service(plan, [clip], tmp_path=tmp_path)
|
||
from video_processing.unified_render_service import RenderLayer
|
||
|
||
layer = RenderLayer(role="main", clips=[clip])
|
||
layers = [layer]
|
||
|
||
assert service._can_use_pass_through(layers) is True
|
||
|
||
def test_build_sticker_filters_text(self, tmp_path):
|
||
"""文字贴纸滤镜构建."""
|
||
plan = FakePlan(
|
||
id="p1", config={"stickers": [{"type": "text", "text": "Hello", "position": "top_center", "z_index": 10}]}
|
||
)
|
||
service = _make_service(plan, [], tmp_path=tmp_path)
|
||
|
||
filter_str, extra_inputs = service._build_sticker_filters("in", "out")
|
||
|
||
assert "drawtext" in filter_str
|
||
assert len(extra_inputs) == 0
|
||
|
||
def test_build_sticker_filters_empty(self, tmp_path):
|
||
"""无贴纸返回空."""
|
||
plan = FakePlan(id="p1", config={})
|
||
service = _make_service(plan, [], tmp_path=tmp_path)
|
||
|
||
filter_str, extra_inputs = service._build_sticker_filters("in", "out")
|
||
|
||
assert filter_str == ""
|
||
assert extra_inputs == []
|
||
|
||
def test_build_sticker_filters_image(self, sample_image, tmp_path):
|
||
"""图片贴纸滤镜构建 + 额外输入."""
|
||
plan = FakePlan(
|
||
id="p1",
|
||
config={
|
||
"stickers": [
|
||
{
|
||
"type": "image",
|
||
"image_path": str(sample_image),
|
||
"position": "bottom_right",
|
||
"z_index": 5,
|
||
}
|
||
]
|
||
},
|
||
)
|
||
service = _make_service(plan, [], tmp_path=tmp_path)
|
||
|
||
filter_str, extra_inputs = service._build_sticker_filters("in", "out")
|
||
|
||
assert "overlay" in filter_str
|
||
assert len(extra_inputs) == 1
|