ed972a230c
CI/CD Pipeline / Validate Code Quality And Tests (push) Successful in 1m10s
CI/CD Pipeline / Frontend Lint (push) Successful in 2m34s
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 / Build & Push Staging (Watchtower auto-deploy) (push) Successful in 3m40s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 1m51s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 5m1s
599 lines
20 KiB
Python
599 lines
20 KiB
Python
"""
|
||
PlanGeneratorService 单元测试
|
||
|
||
覆盖(6 组测试):
|
||
- ONE_TAKE 模式:素材顺序分配给 main clips
|
||
- PIP 模式:第1个素材→main,其余→overlay
|
||
- VOICE_OVER 模式:素材→main clips (B-roll)
|
||
- VOICE_PIP 模式:第1个→background, 第2个→corner_voice, 其余→b_roll
|
||
- 无 clip_configs 时自动生成默认结构
|
||
- 空素材列表时 clips 创建但无素材分配
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, List, Optional
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
|
||
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
|
||
|
||
import pytest
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
|
||
|
||
from packages.domain.edit_plan import EditPlan, EditPlanStatus
|
||
from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
|
||
from packages.domain.edit_template import EditTemplate, EditTemplateStatus
|
||
from packages.domain.editing_mode import EditingMode
|
||
from packages.domain.template_clip_config import ClipType, TemplateClipConfig, TransitionEffect
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Stub Repositories
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class StubEditPlanRepository:
|
||
"""内存中的 EditPlan 仓储 stub"""
|
||
|
||
def __init__(self) -> None:
|
||
self._plans: dict[str, EditPlan] = {}
|
||
self._counter = 0
|
||
|
||
def _next_id(self) -> str:
|
||
self._counter += 1
|
||
return f"plan-{self._counter:03d}"
|
||
|
||
def get(self, plan_id: str) -> Optional[EditPlan]:
|
||
return self._plans.get(plan_id)
|
||
|
||
def create(self, plan: EditPlan) -> EditPlan:
|
||
if not plan.id:
|
||
plan = EditPlan(
|
||
id=self._next_id(),
|
||
template_id=plan.template_id,
|
||
name=plan.name,
|
||
status=plan.status,
|
||
total_duration=plan.total_duration,
|
||
source_edit_plan_id=plan.source_edit_plan_id,
|
||
project_id=plan.project_id,
|
||
created_by_user_id=plan.created_by_user_id,
|
||
config=plan.config,
|
||
created_at=plan.created_at,
|
||
updated_at=plan.updated_at,
|
||
)
|
||
self._plans[plan.id] = plan
|
||
return plan
|
||
|
||
def update(self, plan: EditPlan) -> EditPlan:
|
||
self._plans[plan.id] = plan
|
||
return plan
|
||
|
||
def list_all(self, **kwargs) -> List[EditPlan]:
|
||
return list(self._plans.values())
|
||
|
||
def count(self, **kwargs) -> int:
|
||
return len(self._plans)
|
||
|
||
def delete(self, plan_id: str) -> bool:
|
||
return self._plans.pop(plan_id, None) is not None
|
||
|
||
|
||
class StubEditPlanClipRepository:
|
||
"""内存中的 EditPlanClip 仓储 stub"""
|
||
|
||
def __init__(self) -> None:
|
||
self._clips: dict[str, EditPlanClip] = {}
|
||
self._counter = 0
|
||
|
||
def _next_id(self) -> str:
|
||
self._counter += 1
|
||
return f"clip-{self._counter:03d}"
|
||
|
||
def get(self, clip_id: str) -> Optional[EditPlanClip]:
|
||
return self._clips.get(clip_id)
|
||
|
||
def create(self, clip: EditPlanClip) -> EditPlanClip:
|
||
if not clip.id:
|
||
clip = EditPlanClip(
|
||
id=self._next_id(),
|
||
plan_id=clip.plan_id,
|
||
template_clip_config_id=clip.template_clip_config_id,
|
||
clip_type=clip.clip_type,
|
||
order=clip.order,
|
||
asset_id=clip.asset_id,
|
||
text_content=clip.text_content,
|
||
start_time=clip.start_time,
|
||
duration=clip.duration,
|
||
transition_effect=clip.transition_effect,
|
||
status=clip.status,
|
||
config=clip.config,
|
||
created_at=clip.created_at,
|
||
updated_at=clip.updated_at,
|
||
)
|
||
self._clips[clip.id] = clip
|
||
return clip
|
||
|
||
def update(self, clip: EditPlanClip) -> EditPlanClip:
|
||
self._clips[clip.id] = clip
|
||
return clip
|
||
|
||
def list_by_plan(
|
||
self,
|
||
plan_id: str,
|
||
*,
|
||
status: Optional[EditPlanClipStatus] = None,
|
||
skip: int = 0,
|
||
limit: int = 100,
|
||
) -> List[EditPlanClip]:
|
||
items = [c for c in self._clips.values() if c.plan_id == plan_id]
|
||
items.sort(key=lambda c: c.order)
|
||
if status:
|
||
items = [c for c in items if c.status == status]
|
||
return items[skip : skip + limit]
|
||
|
||
def delete(self, clip_id: str) -> bool:
|
||
return self._clips.pop(clip_id, None) is not None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helper: 构建 PlanGeneratorService(patch 仓储)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _make_generator():
|
||
"""创建使用 stub 仓储的 PlanGeneratorService"""
|
||
from apps.api.app.services.plan_generator_service import PlanGeneratorService
|
||
|
||
plan_repo = StubEditPlanRepository()
|
||
clip_repo = StubEditPlanClipRepository()
|
||
|
||
with (
|
||
patch(
|
||
"apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanRepository",
|
||
return_value=plan_repo,
|
||
),
|
||
patch(
|
||
"apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanClipRepository",
|
||
return_value=clip_repo,
|
||
),
|
||
):
|
||
db = MagicMock()
|
||
svc = PlanGeneratorService(db)
|
||
# 替换为 stub
|
||
svc._plan_repo = plan_repo
|
||
svc._clip_repo = clip_repo
|
||
|
||
return svc, plan_repo, clip_repo
|
||
|
||
|
||
def _make_template(
|
||
editing_mode: str = "one_take",
|
||
config: Optional[dict] = None,
|
||
) -> EditTemplate:
|
||
"""创建测试用 EditTemplate"""
|
||
return EditTemplate(
|
||
id="tpl-001",
|
||
name="测试模板",
|
||
description="",
|
||
template_type="default",
|
||
editing_mode=editing_mode,
|
||
config=config or {},
|
||
preview_url="",
|
||
sort_weight=0,
|
||
status=EditTemplateStatus.ACTIVE,
|
||
created_at=datetime.now(timezone.utc),
|
||
updated_at=datetime.now(timezone.utc),
|
||
)
|
||
|
||
|
||
def _make_clip_configs(
|
||
template_id: str = "tpl-001",
|
||
specs: Optional[List[dict]] = None,
|
||
) -> List[TemplateClipConfig]:
|
||
"""创建测试用 TemplateClipConfig 列表
|
||
|
||
specs 示例: [{"clip_type": ClipType.INTRO, "order": 0}, ...]
|
||
"""
|
||
if specs is None:
|
||
specs = [
|
||
{"clip_type": ClipType.INTRO, "order": 0, "min_duration": 2.0, "max_duration": 4.0},
|
||
{"clip_type": ClipType.MAIN, "order": 1, "min_duration": 3.0, "max_duration": 7.0},
|
||
{"clip_type": ClipType.OUTRO, "order": 2, "min_duration": 2.0, "max_duration": 4.0},
|
||
]
|
||
configs = []
|
||
for i, spec in enumerate(specs):
|
||
cfg = TemplateClipConfig(
|
||
id=f"cfg-{i:03d}",
|
||
template_id=template_id,
|
||
clip_type=spec.get("clip_type", ClipType.MAIN),
|
||
order=spec.get("order", i),
|
||
min_duration=spec.get("min_duration", 0.0),
|
||
max_duration=spec.get("max_duration", 0.0),
|
||
text_template=spec.get("text_template", ""),
|
||
material_requirements=spec.get("material_requirements"),
|
||
transition_effect=spec.get("transition_effect", TransitionEffect.CUT),
|
||
config=spec.get("config"),
|
||
created_at=datetime.now(timezone.utc),
|
||
updated_at=datetime.now(timezone.utc),
|
||
)
|
||
configs.append(cfg)
|
||
return configs
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 测试:ONE_TAKE 模式
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGenerateOneTakePlan:
|
||
"""ONE_TAKE 模式:素材顺序分配给 main clips"""
|
||
|
||
def test_generate_one_take_plan(self):
|
||
"""3个clip_configs + 3个asset_ids → 按顺序分配"""
|
||
svc, plan_repo, clip_repo = _make_generator()
|
||
template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
|
||
clip_configs = _make_clip_configs()
|
||
asset_ids = ["asset-1", "asset-2", "asset-3"]
|
||
|
||
result = svc.generate_from_template(
|
||
template=template,
|
||
clip_configs=clip_configs,
|
||
asset_ids=asset_ids,
|
||
project_id="proj-001",
|
||
created_by_user_id="user-001",
|
||
)
|
||
|
||
plan = result["plan"]
|
||
clips = result["clips"]
|
||
|
||
assert plan.template_id == "tpl-001"
|
||
assert plan.config["editing_mode"] == "one_take"
|
||
assert plan.status == EditPlanStatus.EDITING
|
||
assert len(clips) == 3
|
||
|
||
# 按 order 排序后检查素材分配
|
||
sorted_clips = sorted(clips, key=lambda c: c.order)
|
||
# intro clip (order=0) 不是 main 类型,不分配素材
|
||
# main clip (order=1) → asset-2(ONE_TAKE 只分配给 main clips)
|
||
# outro clip (order=2) 不是 main 类型
|
||
main_clips = [c for c in sorted_clips if c.clip_type == ClipType.MAIN.value]
|
||
assert len(main_clips) == 1
|
||
assert main_clips[0].asset_id == "asset-1"
|
||
|
||
def test_one_take_plan_name_from_template(self):
|
||
"""name 为空时自动取模板名"""
|
||
svc, _, _ = _make_generator()
|
||
template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
|
||
|
||
result = svc.generate_from_template(
|
||
template=template,
|
||
clip_configs=[],
|
||
asset_ids=["asset-1"],
|
||
)
|
||
|
||
assert "测试模板" in result["plan"].name
|
||
|
||
def test_one_take_custom_name(self):
|
||
"""指定 name 时使用自定义名称"""
|
||
svc, _, _ = _make_generator()
|
||
template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
|
||
|
||
result = svc.generate_from_template(
|
||
template=template,
|
||
clip_configs=[],
|
||
asset_ids=["asset-1"],
|
||
name="我的剪辑",
|
||
)
|
||
|
||
assert result["plan"].name == "我的剪辑"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 测试:PIP 模式
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGeneratePipPlan:
|
||
"""PIP 模式:第1个素材→main,其余→overlay"""
|
||
|
||
def test_generate_pip_plan(self):
|
||
"""4个素材 → 第1个→main,其余→overlay"""
|
||
svc, _, _ = _make_generator()
|
||
template = _make_template(editing_mode=EditingMode.PIP.value)
|
||
|
||
# 无 clip_configs,自动生成默认结构
|
||
asset_ids = ["bg-asset", "overlay-1", "overlay-2", "overlay-3"]
|
||
|
||
result = svc.generate_from_template(
|
||
template=template,
|
||
clip_configs=[],
|
||
asset_ids=asset_ids,
|
||
)
|
||
|
||
plan = result["plan"]
|
||
clips = result["clips"]
|
||
|
||
assert plan.config["editing_mode"] == "pip"
|
||
# 自动生成: 1 main + 3 overlay
|
||
assert len(clips) == 4
|
||
|
||
main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
|
||
overlay_clips = [c for c in clips if c.clip_type == "overlay"]
|
||
|
||
assert len(main_clips) == 1
|
||
assert len(overlay_clips) == 3
|
||
|
||
# 第1个素材 → main
|
||
assert main_clips[0].asset_id == "bg-asset"
|
||
# 其余 → overlay
|
||
assert overlay_clips[0].asset_id == "overlay-1"
|
||
assert overlay_clips[1].asset_id == "overlay-2"
|
||
assert overlay_clips[2].asset_id == "overlay-3"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 测试:VOICE_OVER 模式
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGenerateVoiceOverPlan:
|
||
"""VOICE_OVER 模式:素材→main clips (B-roll)"""
|
||
|
||
def test_generate_voice_over_plan(self):
|
||
"""3个素材 → 3个 main clips,每个标记为 b_roll"""
|
||
svc, _, _ = _make_generator()
|
||
template = _make_template(editing_mode=EditingMode.VOICE_OVER.value)
|
||
asset_ids = ["video-1", "video-2", "video-3"]
|
||
|
||
result = svc.generate_from_template(
|
||
template=template,
|
||
clip_configs=[],
|
||
asset_ids=asset_ids,
|
||
)
|
||
|
||
plan = result["plan"]
|
||
clips = result["clips"]
|
||
|
||
assert plan.config["editing_mode"] == "voice_over"
|
||
assert len(clips) == 3
|
||
|
||
# 所有 clips 都是 main 类型
|
||
for clip in clips:
|
||
assert clip.clip_type == ClipType.MAIN.value
|
||
|
||
# 素材按顺序分配
|
||
assert clips[0].asset_id == "video-1"
|
||
assert clips[1].asset_id == "video-2"
|
||
assert clips[2].asset_id == "video-3"
|
||
|
||
# 每个 clip 的 config 标记为 b_roll
|
||
for clip in clips:
|
||
assert clip.config.get("role") == "b_roll"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 测试:VOICE_PIP 模式
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGenerateVoicePipPlan:
|
||
"""VOICE_PIP 模式:第1个→background, 第2个→corner_voice, 其余→b_roll"""
|
||
|
||
def test_generate_voice_pip_plan(self):
|
||
"""4个素材 → background + corner_voice + 2 b_roll"""
|
||
svc, _, _ = _make_generator()
|
||
template = _make_template(editing_mode=EditingMode.VOICE_PIP.value)
|
||
asset_ids = ["bg-video", "corner-video", "broll-1", "broll-2"]
|
||
|
||
result = svc.generate_from_template(
|
||
template=template,
|
||
clip_configs=[],
|
||
asset_ids=asset_ids,
|
||
)
|
||
|
||
plan = result["plan"]
|
||
clips = result["clips"]
|
||
|
||
assert plan.config["editing_mode"] == "voice_pip"
|
||
assert len(clips) == 4
|
||
|
||
bg_clips = [c for c in clips if c.clip_type == "background"]
|
||
corner_clips = [c for c in clips if c.clip_type == "corner_voice"]
|
||
broll_clips = [c for c in clips if c.clip_type == "b_roll"]
|
||
|
||
assert len(bg_clips) == 1
|
||
assert len(corner_clips) == 1
|
||
assert len(broll_clips) == 2
|
||
|
||
# 素材分配
|
||
assert bg_clips[0].asset_id == "bg-video"
|
||
assert corner_clips[0].asset_id == "corner-video"
|
||
assert broll_clips[0].asset_id == "broll-1"
|
||
assert broll_clips[1].asset_id == "broll-2"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 测试:无 clip_configs 时自动生成默认结构
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGenerateWithoutClipConfigs:
|
||
"""无 clip_configs 时根据 editing_mode 生成默认 clip 结构"""
|
||
|
||
def test_one_take_default_clips(self):
|
||
"""ONE_TAKE + 3个素材 → 3个 main clips"""
|
||
svc, _, _ = _make_generator()
|
||
template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
|
||
|
||
result = svc.generate_from_template(
|
||
template=template,
|
||
clip_configs=[],
|
||
asset_ids=["a1", "a2", "a3"],
|
||
)
|
||
|
||
clips = result["clips"]
|
||
assert len(clips) == 3
|
||
for clip in clips:
|
||
assert clip.clip_type == ClipType.MAIN.value
|
||
|
||
def test_pip_default_clips(self):
|
||
"""PIP + 3个素材 → 1 main + 2 overlay"""
|
||
svc, _, _ = _make_generator()
|
||
template = _make_template(editing_mode=EditingMode.PIP.value)
|
||
|
||
result = svc.generate_from_template(
|
||
template=template,
|
||
clip_configs=[],
|
||
asset_ids=["a1", "a2", "a3"],
|
||
)
|
||
|
||
clips = result["clips"]
|
||
assert len(clips) == 3
|
||
main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
|
||
overlay_clips = [c for c in clips if c.clip_type == "overlay"]
|
||
assert len(main_clips) == 1
|
||
assert len(overlay_clips) == 2
|
||
|
||
def test_voice_pip_default_clips(self):
|
||
"""VOICE_PIP + 4个素材 → 1 background + 1 corner_voice + 2 b_roll"""
|
||
svc, _, _ = _make_generator()
|
||
template = _make_template(editing_mode=EditingMode.VOICE_PIP.value)
|
||
|
||
result = svc.generate_from_template(
|
||
template=template,
|
||
clip_configs=[],
|
||
asset_ids=["a1", "a2", "a3", "a4"],
|
||
)
|
||
|
||
clips = result["clips"]
|
||
assert len(clips) == 4
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 测试:空素材列表
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGenerateEmptyAssets:
|
||
"""空素材列表时 clips 创建但无素材分配"""
|
||
|
||
def test_empty_assets(self):
|
||
"""空 asset_ids → clips 创建但 asset_id 为空"""
|
||
svc, _, _ = _make_generator()
|
||
template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
|
||
clip_configs = _make_clip_configs()
|
||
|
||
result = svc.generate_from_template(
|
||
template=template,
|
||
clip_configs=clip_configs,
|
||
asset_ids=[],
|
||
)
|
||
|
||
plan = result["plan"]
|
||
clips = result["clips"]
|
||
|
||
assert plan.total_duration > 0 # clips 有默认时长
|
||
assert len(clips) == 3
|
||
for clip in clips:
|
||
assert clip.asset_id == ""
|
||
|
||
def test_empty_assets_pip(self):
|
||
"""PIP 模式空素材 → 1个 main clip(至少1个)"""
|
||
svc, _, _ = _make_generator()
|
||
template = _make_template(editing_mode=EditingMode.PIP.value)
|
||
|
||
result = svc.generate_from_template(
|
||
template=template,
|
||
clip_configs=[],
|
||
asset_ids=[],
|
||
)
|
||
|
||
clips = result["clips"]
|
||
# 至少1个 main clip(n = max(asset_count, 1) = 1)
|
||
assert len(clips) == 1
|
||
assert clips[0].clip_type == ClipType.MAIN.value
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 测试:plan config 继承模板配置
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPlanConfigInheritance:
|
||
"""plan config 继承模板的 cover/title/subtitle/bgm"""
|
||
|
||
def test_inherit_template_config(self):
|
||
"""模板有 cover/title/bgm 配置 → plan 继承"""
|
||
svc, _, _ = _make_generator()
|
||
template_config = {
|
||
"editing_mode": "one_take",
|
||
"cover": {"type": "ai_frame"},
|
||
"title": {"text": "测试标题", "font": "思源黑体"},
|
||
"bgm": {"url": "https://example.com/bgm.mp3"},
|
||
}
|
||
template = _make_template(
|
||
editing_mode=EditingMode.ONE_TAKE.value,
|
||
config=template_config,
|
||
)
|
||
|
||
result = svc.generate_from_template(
|
||
template=template,
|
||
clip_configs=[],
|
||
asset_ids=["a1"],
|
||
)
|
||
|
||
plan_config = result["plan"].config
|
||
assert plan_config["editing_mode"] == "one_take"
|
||
assert plan_config["cover"]["type"] == "ai_frame"
|
||
assert plan_config["title"]["text"] == "测试标题"
|
||
assert plan_config["bgm"]["url"] == "https://example.com/bgm.mp3"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 测试:total_duration 计算
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestTotalDuration:
|
||
"""total_duration 正确计算"""
|
||
|
||
def test_duration_from_clip_configs(self):
|
||
"""有 clip_configs 时,duration 取 min/max 中间值"""
|
||
svc, _, _ = _make_generator()
|
||
template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
|
||
clip_configs = _make_clip_configs(
|
||
specs=[
|
||
{"clip_type": ClipType.INTRO, "order": 0, "min_duration": 2.0, "max_duration": 4.0},
|
||
{"clip_type": ClipType.MAIN, "order": 1, "min_duration": 4.0, "max_duration": 6.0},
|
||
{"clip_type": ClipType.OUTRO, "order": 2, "min_duration": 2.0, "max_duration": 4.0},
|
||
]
|
||
)
|
||
|
||
result = svc.generate_from_template(
|
||
template=template,
|
||
clip_configs=clip_configs,
|
||
asset_ids=["a1"],
|
||
)
|
||
|
||
# intro: (2+4)/2=3, main: (4+6)/2=5, outro: (2+4)/2=3 → total=11
|
||
assert result["plan"].total_duration == 11.0
|
||
|
||
def test_duration_from_default_clips(self):
|
||
"""无 clip_configs 时,每个 clip 默认 5 秒"""
|
||
svc, _, _ = _make_generator()
|
||
template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
|
||
|
||
result = svc.generate_from_template(
|
||
template=template,
|
||
clip_configs=[],
|
||
asset_ids=["a1", "a2", "a3"],
|
||
)
|
||
|
||
# 3 个 clips × 5 秒 = 15 秒
|
||
assert result["plan"].total_duration == 15.0
|