xiaoxia
981d965d89
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (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 Staging Web Image (push) Successful in 48s
CI/CD Pipeline / Frontend Lint (push) Successful in 3m18s
CI/CD Pipeline / Validate Code Quality And Tests (push) Successful in 5m42s
CI/CD Pipeline / Unit Tests (push) Successful in 6m2s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 6m13s
CI/CD Pipeline / Integration Tests (push) Successful in 2m14s
CI/CD Pipeline / Build Staging API Image (push) Successful in 16m6s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 17m8s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 59s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 36s
CI/CD Pipeline / Staging E2E Tests (push) Failing after 59s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 3m11s
Co-authored-by: xiaoxia <dev@xiaoxiajianji.com> Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
1023 lines
36 KiB
Python
Executable File
1023 lines
36 KiB
Python
Executable File
"""
|
||
EditTemplateService 单元测试
|
||
|
||
覆盖（30+ 测试用例）：
|
||
- 模板 CRUD：创建、查询、更新、软删除
|
||
- 名称去重校验
|
||
- 片段配置 CRUD
|
||
- 片段配置重排序
|
||
- 复合查询 get_template_with_configs
|
||
- 异常处理：不存在、参数校验
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import List, Optional
|
||
from unittest.mock import MagicMock
|
||
|
||
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_template import EditTemplate, EditTemplateStatus
|
||
from packages.domain.template_clip_config import ClipType, TemplateClipConfig
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Stub Repositories
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class StubEditTemplateRepository:
|
||
"""内存中的 EditTemplate 仓储 stub"""
|
||
|
||
def __init__(self) -> None:
|
||
self._templates: dict[str, EditTemplate] = {}
|
||
self._counter = 0
|
||
|
||
def _next_id(self) -> str:
|
||
self._counter += 1
|
||
return f"tpl-{self._counter:03d}"
|
||
|
||
def list_all(
|
||
self,
|
||
*,
|
||
template_type: Optional[str] = None,
|
||
status: Optional[EditTemplateStatus] = None,
|
||
skip: int = 0,
|
||
limit: int = 50,
|
||
) -> List[EditTemplate]:
|
||
items = list(self._templates.values())
|
||
if template_type:
|
||
items = [t for t in items if t.template_type == template_type]
|
||
if status:
|
||
items = [t for t in items if t.status == status]
|
||
return items[skip : skip + limit]
|
||
|
||
def list_active(
|
||
self,
|
||
*,
|
||
template_type: Optional[str] = None,
|
||
skip: int = 0,
|
||
limit: int = 50,
|
||
) -> List[EditTemplate]:
|
||
items = [t for t in self._templates.values() if t.status == EditTemplateStatus.ACTIVE]
|
||
if template_type:
|
||
items = [t for t in items if t.template_type == template_type]
|
||
return items[skip : skip + limit]
|
||
|
||
def get(self, template_id: str) -> Optional[EditTemplate]:
|
||
return self._templates.get(template_id)
|
||
|
||
def create(self, template: EditTemplate) -> EditTemplate:
|
||
if not template.id:
|
||
template = EditTemplate(
|
||
id=self._next_id(),
|
||
name=template.name,
|
||
description=template.description,
|
||
template_type=template.template_type,
|
||
config=template.config,
|
||
preview_url=template.preview_url,
|
||
sort_weight=template.sort_weight,
|
||
status=template.status,
|
||
created_at=datetime.now(timezone.utc),
|
||
updated_at=datetime.now(timezone.utc),
|
||
)
|
||
self._templates[template.id] = template
|
||
return template
|
||
|
||
def update(self, template: EditTemplate) -> EditTemplate:
|
||
self._templates[template.id] = template
|
||
return template
|
||
|
||
def delete(self, template_id: str) -> bool:
|
||
return self._templates.pop(template_id, None) is not None
|
||
|
||
def count(
|
||
self,
|
||
*,
|
||
template_type: Optional[str] = None,
|
||
status: Optional[EditTemplateStatus] = None,
|
||
) -> int:
|
||
items = list(self._templates.values())
|
||
if template_type:
|
||
items = [t for t in items if t.template_type == template_type]
|
||
if status:
|
||
items = [t for t in items if t.status == status]
|
||
return len(items)
|
||
|
||
|
||
class StubTemplateClipConfigRepository:
|
||
"""内存中的 TemplateClipConfig 仓储 stub"""
|
||
|
||
def __init__(self) -> None:
|
||
self._configs: dict[str, TemplateClipConfig] = {}
|
||
self._counter = 0
|
||
|
||
def _next_id(self) -> str:
|
||
self._counter += 1
|
||
return f"cfg-{self._counter:03d}"
|
||
|
||
def list_by_template(
|
||
self,
|
||
template_id: str,
|
||
*,
|
||
clip_type: Optional[ClipType] = None,
|
||
skip: int = 0,
|
||
limit: int = 100,
|
||
) -> List[TemplateClipConfig]:
|
||
items = [c for c in self._configs.values() if c.template_id == template_id]
|
||
if clip_type:
|
||
items = [c for c in items if c.clip_type == clip_type]
|
||
items.sort(key=lambda c: c.order)
|
||
return items[skip : skip + limit]
|
||
|
||
def get(self, config_id: str) -> Optional[TemplateClipConfig]:
|
||
return self._configs.get(config_id)
|
||
|
||
def create(self, config: TemplateClipConfig) -> TemplateClipConfig:
|
||
if not config.id:
|
||
config = TemplateClipConfig(
|
||
id=self._next_id(),
|
||
template_id=config.template_id,
|
||
clip_type=config.clip_type,
|
||
order=config.order,
|
||
min_duration=config.min_duration,
|
||
max_duration=config.max_duration,
|
||
text_template=config.text_template,
|
||
material_requirements=config.material_requirements,
|
||
transition_effect=config.transition_effect,
|
||
config=config.config,
|
||
created_at=datetime.now(timezone.utc),
|
||
updated_at=datetime.now(timezone.utc),
|
||
)
|
||
self._configs[config.id] = config
|
||
return config
|
||
|
||
def update(self, config: TemplateClipConfig) -> TemplateClipConfig:
|
||
self._configs[config.id] = config
|
||
return config
|
||
|
||
def delete(self, config_id: str) -> bool:
|
||
return self._configs.pop(config_id, None) is not None
|
||
|
||
def delete_by_template(self, template_id: str, *, commit: bool = True) -> int:
|
||
ids = [cid for cid, c in self._configs.items() if c.template_id == template_id]
|
||
for cid in ids:
|
||
del self._configs[cid]
|
||
return len(ids)
|
||
|
||
def count(self, template_id: str) -> int:
|
||
return len([c for c in self._configs.values() if c.template_id == template_id])
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Service under test (inject stub repos)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _make_service():
|
||
"""创建使用 stub 仓储的 EditTemplateService"""
|
||
from app.services.edit_template_service import EditTemplateService
|
||
|
||
db = MagicMock()
|
||
svc = EditTemplateService(db)
|
||
svc._template_repo = StubEditTemplateRepository()
|
||
svc._clip_config_repo = StubTemplateClipConfigRepository()
|
||
return svc
|
||
|
||
|
||
# ===========================================================================
|
||
# 模板 CRUD 测试
|
||
# ===========================================================================
|
||
|
||
|
||
class TestEditTemplateServiceCRUD:
|
||
"""模板 CRUD 测试"""
|
||
|
||
def test_create_template_success(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="测试模板", description="描述")
|
||
assert t.name == "测试模板"
|
||
assert t.description == "描述"
|
||
assert t.status == EditTemplateStatus.ACTIVE
|
||
assert t.id
|
||
|
||
def test_create_template_strips_name(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name=" 测试 ")
|
||
assert t.name == "测试"
|
||
|
||
def test_create_template_empty_name_raises(self):
|
||
svc = _make_service()
|
||
with pytest.raises(ValueError, match="模板名称不能为空"):
|
||
svc.create_template(name=" ")
|
||
|
||
def test_create_template_duplicate_name_raises(self):
|
||
svc = _make_service()
|
||
svc.create_template(name="重复名称")
|
||
with pytest.raises(ValueError, match="模板名称已存在"):
|
||
svc.create_template(name="重复名称")
|
||
|
||
def test_create_template_inactive_name_can_reuse(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="可复用")
|
||
svc.deactivate_template(t.id)
|
||
# inactive 的名称可以复用
|
||
t2 = svc.create_template(name="可复用")
|
||
assert t2.name == "可复用"
|
||
assert t2.id != t.id
|
||
|
||
def test_get_template(self):
|
||
svc = _make_service()
|
||
created = svc.create_template(name="查询测试")
|
||
fetched = svc.get_template(created.id)
|
||
assert fetched is not None
|
||
assert fetched.id == created.id
|
||
|
||
def test_get_template_returns_none(self):
|
||
svc = _make_service()
|
||
assert svc.get_template("nonexistent") is None
|
||
|
||
def test_get_template_or_raise(self):
|
||
svc = _make_service()
|
||
created = svc.create_template(name="查询测试")
|
||
fetched = svc.get_template_or_raise(created.id)
|
||
assert fetched.id == created.id
|
||
|
||
def test_get_template_or_raise_not_found(self):
|
||
svc = _make_service()
|
||
with pytest.raises(ValueError, match="模板不存在"):
|
||
svc.get_template_or_raise("nonexistent")
|
||
|
||
def test_list_templates(self):
|
||
svc = _make_service()
|
||
svc.create_template(name="模板1")
|
||
svc.create_template(name="模板2")
|
||
result = svc.list_templates()
|
||
assert len(result) == 2
|
||
|
||
def test_list_templates_with_type_filter(self):
|
||
svc = _make_service()
|
||
svc.create_template(name="默认", template_type="default")
|
||
svc.create_template(name="Vlog", template_type="vlog")
|
||
result = svc.list_templates(template_type="vlog")
|
||
assert len(result) == 1
|
||
assert result[0].name == "Vlog"
|
||
|
||
def test_list_templates_active_only(self):
|
||
svc = _make_service()
|
||
svc.create_template(name="活跃")
|
||
t2 = svc.create_template(name="停用")
|
||
svc.deactivate_template(t2.id)
|
||
result = svc.list_templates(active_only=True)
|
||
assert len(result) == 1
|
||
assert result[0].name == "活跃"
|
||
|
||
def test_count_templates(self):
|
||
svc = _make_service()
|
||
svc.create_template(name="模板1")
|
||
svc.create_template(name="模板2")
|
||
assert svc.count_templates() == 2
|
||
|
||
def test_update_template_name(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="原名")
|
||
updated = svc.update_template(t.id, name="新名")
|
||
assert updated.name == "新名"
|
||
|
||
def test_update_template_duplicate_name_raises(self):
|
||
svc = _make_service()
|
||
svc.create_template(name="已存在")
|
||
t2 = svc.create_template(name="另一个")
|
||
with pytest.raises(ValueError, match="模板名称已存在"):
|
||
svc.update_template(t2.id, name="已存在")
|
||
|
||
def test_update_template_not_found_raises(self):
|
||
svc = _make_service()
|
||
with pytest.raises(ValueError, match="模板不存在"):
|
||
svc.update_template("nonexistent", name="新名")
|
||
|
||
def test_deactivate_template(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="要停用的")
|
||
result = svc.deactivate_template(t.id)
|
||
assert result.status == EditTemplateStatus.INACTIVE
|
||
|
||
def test_deactivate_template_not_found_raises(self):
|
||
svc = _make_service()
|
||
with pytest.raises(ValueError, match="模板不存在"):
|
||
svc.deactivate_template("nonexistent")
|
||
|
||
|
||
# ===========================================================================
|
||
# 片段配置管理测试
|
||
# ===========================================================================
|
||
|
||
|
||
class TestClipConfigManagement:
|
||
"""片段配置管理测试"""
|
||
|
||
def test_create_clip_config(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="模板")
|
||
cfg = svc.create_clip_config(
|
||
template_id=t.id,
|
||
clip_type=ClipType.INTRO,
|
||
order=0,
|
||
min_duration=1.0,
|
||
max_duration=5.0,
|
||
)
|
||
assert cfg.template_id == t.id
|
||
assert cfg.clip_type == ClipType.INTRO
|
||
assert cfg.order == 0
|
||
assert cfg.min_duration == 1.0
|
||
assert cfg.max_duration == 5.0
|
||
|
||
def test_create_clip_config_with_string_clip_type(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="模板")
|
||
cfg = svc.create_clip_config(
|
||
template_id=t.id,
|
||
clip_type="main",
|
||
order=1,
|
||
)
|
||
assert cfg.clip_type == ClipType.MAIN
|
||
|
||
def test_list_clip_configs(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="模板")
|
||
svc.create_clip_config(t.id, ClipType.INTRO, 0)
|
||
svc.create_clip_config(t.id, ClipType.MAIN, 1)
|
||
svc.create_clip_config(t.id, ClipType.OUTRO, 2)
|
||
result = svc.list_clip_configs(t.id)
|
||
assert len(result) == 3
|
||
# 按 order 排序
|
||
assert result[0].order == 0
|
||
assert result[1].order == 1
|
||
assert result[2].order == 2
|
||
|
||
def test_list_clip_configs_by_type(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="模板")
|
||
svc.create_clip_config(t.id, ClipType.INTRO, 0)
|
||
svc.create_clip_config(t.id, ClipType.MAIN, 1)
|
||
result = svc.list_clip_configs(t.id, clip_type=ClipType.MAIN)
|
||
assert len(result) == 1
|
||
assert result[0].clip_type == ClipType.MAIN
|
||
|
||
def test_get_clip_config(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="模板")
|
||
cfg = svc.create_clip_config(t.id, ClipType.INTRO, 0)
|
||
fetched = svc.get_clip_config(cfg.id)
|
||
assert fetched is not None
|
||
assert fetched.id == cfg.id
|
||
|
||
def test_get_clip_config_not_found(self):
|
||
svc = _make_service()
|
||
assert svc.get_clip_config("nonexistent") is None
|
||
|
||
def test_get_clip_config_or_raise(self):
|
||
svc = _make_service()
|
||
with pytest.raises(ValueError, match="片段配置不存在"):
|
||
svc.get_clip_config_or_raise("nonexistent")
|
||
|
||
def test_update_clip_config(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="模板")
|
||
cfg = svc.create_clip_config(t.id, ClipType.INTRO, 0, min_duration=1.0)
|
||
updated = svc.update_clip_config(cfg.id, min_duration=2.0, max_duration=10.0)
|
||
assert updated.min_duration == 2.0
|
||
assert updated.max_duration == 10.0
|
||
|
||
def test_delete_clip_config(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="模板")
|
||
cfg = svc.create_clip_config(t.id, ClipType.INTRO, 0)
|
||
assert svc.delete_clip_config(cfg.id) is True
|
||
assert svc.get_clip_config(cfg.id) is None
|
||
|
||
def test_delete_clip_config_not_found(self):
|
||
svc = _make_service()
|
||
assert svc.delete_clip_config("nonexistent") is False
|
||
|
||
def test_reorder_clip_configs(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="模板")
|
||
c1 = svc.create_clip_config(t.id, ClipType.INTRO, 0)
|
||
c2 = svc.create_clip_config(t.id, ClipType.MAIN, 1)
|
||
c3 = svc.create_clip_config(t.id, ClipType.OUTRO, 2)
|
||
|
||
# 反转顺序
|
||
reordered = svc.reorder_clip_configs(t.id, [c3.id, c2.id, c1.id])
|
||
assert len(reordered) == 3
|
||
assert reordered[0].id == c3.id
|
||
assert reordered[0].order == 0
|
||
assert reordered[1].id == c2.id
|
||
assert reordered[1].order == 1
|
||
assert reordered[2].id == c1.id
|
||
assert reordered[2].order == 2
|
||
|
||
def test_reorder_clip_configs_mismatch_raises(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="模板")
|
||
c1 = svc.create_clip_config(t.id, ClipType.INTRO, 0)
|
||
svc.create_clip_config(t.id, ClipType.MAIN, 1)
|
||
|
||
with pytest.raises(ValueError, match="配置 ID 列表与模板下的配置不匹配"):
|
||
svc.reorder_clip_configs(t.id, [c1.id]) # 缺少一个
|
||
|
||
|
||
# ===========================================================================
|
||
# 复合查询测试
|
||
# ===========================================================================
|
||
|
||
|
||
class TestCompositeQueries:
|
||
"""复合查询测试"""
|
||
|
||
def test_get_template_with_configs(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="模板")
|
||
svc.create_clip_config(t.id, ClipType.INTRO, 0)
|
||
svc.create_clip_config(t.id, ClipType.MAIN, 1)
|
||
|
||
result = svc.get_template_with_configs(t.id)
|
||
assert result["template"].id == t.id
|
||
assert len(result["clip_configs"]) == 2
|
||
|
||
def test_get_template_with_configs_not_found(self):
|
||
svc = _make_service()
|
||
with pytest.raises(ValueError, match="模板不存在"):
|
||
svc.get_template_with_configs("nonexistent")
|
||
|
||
def test_get_template_with_configs_empty(self):
|
||
svc = _make_service()
|
||
t = svc.create_template(name="空模板")
|
||
result = svc.get_template_with_configs(t.id)
|
||
assert len(result["clip_configs"]) == 0
|
||
|
||
|
||
# ===========================================================================
|
||
# Stub Repositories for EditPlan (save_as_template 测试用)
|
||
# ===========================================================================
|
||
|
||
|
||
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.id = self._next_id()
|
||
self._plans[plan.id] = plan
|
||
return plan
|
||
|
||
def list_by_template(
|
||
self,
|
||
template_id: str,
|
||
*,
|
||
status: Optional[str] = None,
|
||
skip: int = 0,
|
||
limit: int = 50,
|
||
) -> List[EditPlan]:
|
||
items = [p for p in self._plans.values() if p.template_id == template_id]
|
||
if status:
|
||
items = [p for p in items if p.status.value == status]
|
||
items.sort(key=lambda p: p.created_at, reverse=True)
|
||
return items[skip : skip + limit]
|
||
|
||
def update(self, plan: EditPlan) -> EditPlan:
|
||
self._plans[plan.id] = plan
|
||
return plan
|
||
|
||
|
||
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 list_by_plan(
|
||
self,
|
||
plan_id: str,
|
||
*,
|
||
status: Optional[str] = None,
|
||
skip: int = 0,
|
||
limit: int = 100,
|
||
) -> List[EditPlanClip]:
|
||
items = [c for c in self._clips.values() if c.plan_id == plan_id]
|
||
if status:
|
||
items = [c for c in items if c.status.value == status]
|
||
items.sort(key=lambda c: c.order)
|
||
return items[skip : skip + limit]
|
||
|
||
def create(self, clip: EditPlanClip) -> EditPlanClip:
|
||
if not clip.id:
|
||
clip.id = self._next_id()
|
||
self._clips[clip.id] = clip
|
||
return clip
|
||
|
||
|
||
def _make_service_with_plan_stubs():
|
||
"""创建使用 stub 仓储的 EditTemplateService（含 plan 相关 stub）"""
|
||
from app.services.edit_template_service import EditTemplateService
|
||
|
||
db = MagicMock()
|
||
svc = EditTemplateService(db)
|
||
svc._template_repo = StubEditTemplateRepository()
|
||
svc._clip_config_repo = StubTemplateClipConfigRepository()
|
||
svc._plan_repo = StubEditPlanRepository()
|
||
svc._plan_clip_repo = StubEditPlanClipRepository()
|
||
return svc
|
||
|
||
|
||
def _make_test_plan_with_clips(svc, *, clip_count: int = 3, plan_config=None):
|
||
"""辅助方法：创建一个带片段的测试计划，返回 plan 对象"""
|
||
from packages.domain.edit_plan import EditPlan
|
||
from packages.domain.edit_plan_clip import EditPlanClip
|
||
|
||
plan = EditPlan.create(
|
||
template_id="tpl-source",
|
||
name="我的剪辑计划",
|
||
config=plan_config or {"editing_mode": "one_take", "theme": "minimal"},
|
||
project_id="proj-001",
|
||
created_by_user_id="user-001",
|
||
)
|
||
plan.id = "plan-test-001"
|
||
svc._plan_repo.create(plan)
|
||
|
||
for i in range(clip_count):
|
||
clip = EditPlanClip.create(
|
||
plan_id=plan.id,
|
||
clip_type=ClipType.MAIN.value,
|
||
order=i,
|
||
asset_id=f"asset-{i:03d}",
|
||
text_content=f"片段{i}的文案",
|
||
duration=10.0 + i * 5,
|
||
transition_effect="cut" if i == 0 else "fade",
|
||
playback_speed=1.0 if i == 0 else 1.5,
|
||
config={"filter": "vivid"} if i == 1 else {},
|
||
)
|
||
svc._plan_clip_repo.create(clip)
|
||
|
||
return plan
|
||
|
||
|
||
# ===========================================================================
|
||
# 保存为模板测试
|
||
# ===========================================================================
|
||
|
||
|
||
class TestSavePlanAsTemplate:
|
||
"""从剪辑计划保存为模板测试"""
|
||
|
||
def test_basic_save_as_template(self):
|
||
"""基础场景：将有3个片段的计划保存为模板"""
|
||
svc = _make_service_with_plan_stubs()
|
||
plan = _make_test_plan_with_clips(svc, clip_count=3)
|
||
|
||
result = svc.save_plan_as_template(plan.id, name="我的自定义模板")
|
||
|
||
assert result["template"].name == "我的自定义模板"
|
||
assert result["template"].template_type == "custom"
|
||
assert result["template"].editing_mode == "one_take"
|
||
assert result["template"].status == EditTemplateStatus.ACTIVE
|
||
assert len(result["clip_configs"]) == 3
|
||
|
||
def test_clip_configs_correctly_converted(self):
|
||
"""片段正确转换为模板片段配置"""
|
||
svc = _make_service_with_plan_stubs()
|
||
plan = _make_test_plan_with_clips(svc, clip_count=2)
|
||
|
||
result = svc.save_plan_as_template(plan.id, name="转换测试模板")
|
||
configs = result["clip_configs"]
|
||
configs.sort(key=lambda c: c.order)
|
||
|
||
# 第0个片段
|
||
assert configs[0].clip_type == ClipType.MAIN
|
||
assert configs[0].order == 0
|
||
assert configs[0].min_duration == 10.0
|
||
assert configs[0].max_duration == 10.0
|
||
assert configs[0].text_template == "片段0的文案"
|
||
assert configs[0].transition_effect.value == "cut"
|
||
# playback_speed=1.0 不存
|
||
assert "playback_speed" not in configs[0].config
|
||
|
||
# 第1个片段
|
||
assert configs[1].order == 1
|
||
assert configs[1].min_duration == 15.0
|
||
assert configs[1].max_duration == 15.0
|
||
assert configs[1].transition_effect.value == "fade"
|
||
# playback_speed=1.5 存入config
|
||
assert configs[1].config.get("playback_speed") == 1.5
|
||
# config 中的 filter 保留
|
||
assert configs[1].config.get("filter") == "vivid"
|
||
|
||
def test_no_asset_id_in_template(self):
|
||
"""模板不保留具体素材ID"""
|
||
svc = _make_service_with_plan_stubs()
|
||
plan = _make_test_plan_with_clips(svc, clip_count=2)
|
||
|
||
result = svc.save_plan_as_template(plan.id, name="素材剥离测试")
|
||
|
||
for cfg in result["clip_configs"]:
|
||
# 模板片段配置没有 asset_id 字段
|
||
assert not hasattr(cfg, "asset_id") or not getattr(cfg, "asset_id", "")
|
||
# config 中也不应有素材相关字段
|
||
assert "asset_info" not in cfg.config
|
||
assert "source_asset_id" not in cfg.config
|
||
|
||
def test_template_config_stripped_of_runtime_fields(self):
|
||
"""模板config剥离运行时字段"""
|
||
svc = _make_service_with_plan_stubs()
|
||
plan_config = {
|
||
"editing_mode": "one_take",
|
||
"theme": "cinematic",
|
||
"asset_ids": ["a1", "a2"],
|
||
"source_edit_plan_id": "old-plan",
|
||
"generation_task_id": "task-123",
|
||
}
|
||
plan = _make_test_plan_with_clips(svc, clip_count=1, plan_config=plan_config)
|
||
|
||
result = svc.save_plan_as_template(plan.id, name="配置剥离测试")
|
||
|
||
tpl_config = result["template"].config
|
||
assert tpl_config.get("theme") == "cinematic"
|
||
assert "asset_ids" not in tpl_config
|
||
assert "source_edit_plan_id" not in tpl_config
|
||
assert "generation_task_id" not in tpl_config
|
||
|
||
def test_plan_not_found_raises_error(self):
|
||
"""计划不存在时报错"""
|
||
svc = _make_service_with_plan_stubs()
|
||
|
||
with pytest.raises(ValueError, match="剪辑计划不存在"):
|
||
svc.save_plan_as_template("nonexistent-plan", name="不存在的计划")
|
||
|
||
def test_empty_name_raises_error(self):
|
||
"""模板名称为空时报错"""
|
||
svc = _make_service_with_plan_stubs()
|
||
plan = _make_test_plan_with_clips(svc, clip_count=1)
|
||
|
||
with pytest.raises(ValueError, match="模板名称不能为空"):
|
||
svc.save_plan_as_template(plan.id, name=" ")
|
||
|
||
def test_duplicate_name_raises_error(self):
|
||
"""模板名称重复时报错"""
|
||
svc = _make_service_with_plan_stubs()
|
||
svc.create_template(name="重名模板")
|
||
plan = _make_test_plan_with_clips(svc, clip_count=1)
|
||
|
||
with pytest.raises(ValueError, match="模板名称已存在"):
|
||
svc.save_plan_as_template(plan.id, name="重名模板")
|
||
|
||
def test_save_zero_clip_plan(self):
|
||
"""零片段计划也能保存为模板"""
|
||
svc = _make_service_with_plan_stubs()
|
||
from packages.domain.edit_plan import EditPlan
|
||
|
||
plan = EditPlan.create(
|
||
template_id="tpl-source",
|
||
name="空计划",
|
||
config={"editing_mode": "one_take"},
|
||
)
|
||
plan.id = "plan-empty"
|
||
svc._plan_repo.create(plan)
|
||
|
||
result = svc.save_plan_as_template(plan.id, name="空模板")
|
||
|
||
assert result["template"].name == "空模板"
|
||
assert len(result["clip_configs"]) == 0
|
||
|
||
def test_custom_description_and_type(self):
|
||
"""自定义描述和模板类型"""
|
||
svc = _make_service_with_plan_stubs()
|
||
plan = _make_test_plan_with_clips(svc, clip_count=1)
|
||
|
||
result = svc.save_plan_as_template(
|
||
plan.id,
|
||
name="自定义模板",
|
||
description="这是一个测试模板",
|
||
template_type="vlog",
|
||
)
|
||
|
||
assert result["template"].description == "这是一个测试模板"
|
||
assert result["template"].template_type == "vlog"
|
||
|
||
def test_unknown_transition_effect_falls_back_to_cut(self):
|
||
"""未知转场效果回退到cut"""
|
||
svc = _make_service_with_plan_stubs()
|
||
from packages.domain.edit_plan import EditPlan
|
||
from packages.domain.edit_plan_clip import EditPlanClip
|
||
|
||
plan = EditPlan.create(template_id="tpl-src", name="转场测试计划")
|
||
plan.id = "plan-transition-test"
|
||
svc._plan_repo.create(plan)
|
||
|
||
clip = EditPlanClip.create(
|
||
plan_id=plan.id,
|
||
clip_type="main",
|
||
order=0,
|
||
duration=10.0,
|
||
transition_effect="weird_effect_that_does_not_exist",
|
||
)
|
||
svc._plan_clip_repo.create(clip)
|
||
|
||
result = svc.save_plan_as_template(plan.id, name="转场兼容模板")
|
||
assert result["clip_configs"][0].transition_effect.value == "cut"
|
||
|
||
def test_unknown_clip_type_falls_back_to_main(self):
|
||
"""未知片段类型回退到main"""
|
||
svc = _make_service_with_plan_stubs()
|
||
from packages.domain.edit_plan import EditPlan
|
||
from packages.domain.edit_plan_clip import EditPlanClip
|
||
|
||
plan = EditPlan.create(template_id="tpl-src", name="类型测试计划")
|
||
plan.id = "plan-type-test"
|
||
svc._plan_repo.create(plan)
|
||
|
||
clip = EditPlanClip.create(
|
||
plan_id=plan.id,
|
||
clip_type="unknown_clip_type",
|
||
order=0,
|
||
duration=10.0,
|
||
)
|
||
svc._plan_clip_repo.create(clip)
|
||
|
||
result = svc.save_plan_as_template(plan.id, name="类型兼容模板")
|
||
assert result["clip_configs"][0].clip_type == ClipType.MAIN
|
||
|
||
|
||
class TestTemplateDraft:
|
||
"""模板草稿相关功能测试"""
|
||
|
||
def test_get_template_draft_not_found(self):
|
||
"""没有草稿时返回None"""
|
||
svc = _make_service_with_plan_stubs()
|
||
# 先创建一个模板
|
||
tpl = svc.create_template(name="测试模板", editing_mode="one_take")
|
||
result = svc.get_template_draft(tpl.id)
|
||
assert result is None
|
||
|
||
def test_get_template_draft_found(self):
|
||
"""能正确找到标记了is_template_draft的草稿"""
|
||
svc = _make_service_with_plan_stubs()
|
||
from packages.domain.edit_plan import EditPlan
|
||
|
||
tpl = svc.create_template(name="测试模板", editing_mode="one_take")
|
||
|
||
# 普通计划（不是草稿）
|
||
normal_plan = EditPlan.create(template_id=tpl.id, name="普通计划", config={"key": "value"})
|
||
normal_plan.id = "plan-normal"
|
||
svc._plan_repo.create(normal_plan)
|
||
|
||
# 草稿计划
|
||
draft_plan = EditPlan.create(
|
||
template_id=tpl.id,
|
||
name="草稿计划",
|
||
config={"is_template_draft": True, "other": "data"},
|
||
)
|
||
draft_plan.id = "plan-draft"
|
||
svc._plan_repo.create(draft_plan)
|
||
|
||
result = svc.get_template_draft(tpl.id)
|
||
assert result is not None
|
||
assert result.id == "plan-draft"
|
||
assert result.config["is_template_draft"] is True
|
||
|
||
def test_create_template_draft_success(self, monkeypatch):
|
||
"""创建草稿成功，标记is_template_draft=True"""
|
||
svc = _make_service_with_plan_stubs()
|
||
from packages.domain.edit_plan import EditPlan
|
||
|
||
tpl = svc.create_template(name="草稿测试模板", editing_mode="one_take")
|
||
|
||
# mock PlanGeneratorService.generate_from_template
|
||
called_with = {}
|
||
|
||
def fake_generate(self, template, clip_configs, asset_ids, **kwargs):
|
||
called_with["template"] = template
|
||
called_with["clip_configs"] = clip_configs
|
||
called_with["asset_ids"] = asset_ids
|
||
called_with["kwargs"] = kwargs
|
||
plan = EditPlan.create(
|
||
template_id=template.id,
|
||
name=kwargs.get("name", "测试草稿"),
|
||
config={"editing_mode": "one_take"},
|
||
)
|
||
plan.id = "plan-new-draft"
|
||
svc._plan_repo.create(plan)
|
||
return {"plan": plan, "clips": []}
|
||
|
||
from app.services import plan_generator_service
|
||
|
||
monkeypatch.setattr(
|
||
plan_generator_service.PlanGeneratorService,
|
||
"generate_from_template",
|
||
fake_generate,
|
||
)
|
||
|
||
draft = svc.create_template_draft(tpl.id, "user-001")
|
||
|
||
assert draft is not None
|
||
assert draft.id == "plan-new-draft"
|
||
assert draft.config.get("is_template_draft") is True
|
||
assert called_with["asset_ids"] == []
|
||
assert called_with["kwargs"]["created_by_user_id"] == "user-001"
|
||
|
||
def test_create_template_draft_duplicate_raises(self):
|
||
"""草稿已存在时抛ValueError"""
|
||
svc = _make_service_with_plan_stubs()
|
||
from packages.domain.edit_plan import EditPlan
|
||
|
||
tpl = svc.create_template(name="重复草稿模板", editing_mode="one_take")
|
||
|
||
# 预先创建一个草稿
|
||
draft = EditPlan.create(
|
||
template_id=tpl.id,
|
||
name="已存在草稿",
|
||
config={"is_template_draft": True},
|
||
)
|
||
draft.id = "plan-existing"
|
||
svc._plan_repo.create(draft)
|
||
|
||
import pytest
|
||
|
||
with pytest.raises(ValueError, match="草稿已存在"):
|
||
svc.create_template_draft(tpl.id, "user-001")
|
||
|
||
def test_get_or_create_draft_creates_when_missing(self, monkeypatch):
|
||
"""草稿不存在时自动创建"""
|
||
svc = _make_service_with_plan_stubs()
|
||
from packages.domain.edit_plan import EditPlan
|
||
|
||
tpl = svc.create_template(name="自动创建模板", editing_mode="one_take")
|
||
|
||
def fake_generate(self, template, clip_configs, asset_ids, **kwargs):
|
||
plan = EditPlan.create(
|
||
template_id=template.id,
|
||
name="自动创建草稿",
|
||
config={"editing_mode": "one_take"},
|
||
)
|
||
plan.id = "plan-auto"
|
||
svc._plan_repo.create(plan)
|
||
return {"plan": plan, "clips": []}
|
||
|
||
from app.services import plan_generator_service
|
||
|
||
monkeypatch.setattr(
|
||
plan_generator_service.PlanGeneratorService,
|
||
"generate_from_template",
|
||
fake_generate,
|
||
)
|
||
|
||
# 第一次调用：创建
|
||
draft1 = svc.get_or_create_draft(tpl.id, "user-001")
|
||
assert draft1 is not None
|
||
assert draft1.config.get("is_template_draft") is True
|
||
|
||
# 第二次调用：返回已存在的
|
||
draft2 = svc.get_or_create_draft(tpl.id, "user-001")
|
||
assert draft2.id == draft1.id
|
||
|
||
def test_publish_template_from_draft_success(self):
|
||
"""草稿发布成功，同步config和clips到模板"""
|
||
svc = _make_service_with_plan_stubs()
|
||
from packages.domain.edit_plan import EditPlan
|
||
from packages.domain.edit_plan_clip import EditPlanClip
|
||
|
||
# 创建模板和初始片段配置
|
||
tpl = svc.create_template(
|
||
name="发布测试模板",
|
||
editing_mode="one_take",
|
||
config={"original": "value"},
|
||
)
|
||
svc.create_clip_config(
|
||
tpl.id,
|
||
clip_type="main",
|
||
order=0,
|
||
min_duration=5.0,
|
||
max_duration=5.0,
|
||
)
|
||
|
||
# 创建草稿
|
||
draft = EditPlan.create(
|
||
template_id=tpl.id,
|
||
name="发布草稿",
|
||
config={
|
||
"is_template_draft": True,
|
||
"editing_mode": "pip",
|
||
"bgm": "song.mp3",
|
||
"title": "新标题",
|
||
},
|
||
)
|
||
draft.id = "plan-publish"
|
||
svc._plan_repo.create(draft)
|
||
|
||
# 草稿的片段
|
||
clip1 = EditPlanClip.create(
|
||
plan_id=draft.id,
|
||
clip_type="main",
|
||
order=0,
|
||
duration=8.0,
|
||
text_content="第一段",
|
||
transition_effect="fade",
|
||
config={"playback_speed": 1.5},
|
||
)
|
||
clip1.id = "clip-pub-1"
|
||
svc._plan_clip_repo.create(clip1)
|
||
|
||
clip2 = EditPlanClip.create(
|
||
plan_id=draft.id,
|
||
clip_type="main",
|
||
order=1,
|
||
duration=12.0,
|
||
text_content="第二段",
|
||
)
|
||
clip2.id = "clip-pub-2"
|
||
svc._plan_clip_repo.create(clip2)
|
||
|
||
# 发布
|
||
result = svc.publish_template_from_draft(tpl.id, draft.id)
|
||
|
||
assert result is not None
|
||
assert result.editing_mode == "pip"
|
||
assert result.config.get("original") is None # 旧配置被覆盖
|
||
assert result.config.get("bgm") == "song.mp3"
|
||
assert result.config.get("is_template_draft") is None # 草稿标记不带过去
|
||
|
||
# 检查片段配置已更新
|
||
new_configs = svc.list_clip_configs(tpl.id)
|
||
assert len(new_configs) == 2
|
||
assert new_configs[0].order == 0
|
||
assert new_configs[0].min_duration == 8.0
|
||
assert new_configs[0].text_template == "第一段"
|
||
assert new_configs[0].config.get("playback_speed") == 1.5
|
||
assert new_configs[1].order == 1
|
||
assert new_configs[1].max_duration == 12.0
|
||
|
||
def test_publish_template_wrong_template_raises(self):
|
||
"""草稿不属于指定模板时抛错"""
|
||
svc = _make_service_with_plan_stubs()
|
||
from packages.domain.edit_plan import EditPlan
|
||
|
||
tpl1 = svc.create_template(name="模板1", editing_mode="one_take")
|
||
tpl2 = svc.create_template(name="模板2", editing_mode="one_take")
|
||
|
||
# 草稿属于tpl1
|
||
draft = EditPlan.create(
|
||
template_id=tpl1.id,
|
||
name="草稿",
|
||
config={"is_template_draft": True},
|
||
)
|
||
draft.id = "plan-wrong-tpl"
|
||
svc._plan_repo.create(draft)
|
||
|
||
import pytest
|
||
|
||
with pytest.raises(ValueError, match="不属于该模板"):
|
||
svc.publish_template_from_draft(tpl2.id, draft.id)
|
||
|
||
def test_publish_template_not_draft_raises(self):
|
||
"""不是草稿的计划不能发布"""
|
||
svc = _make_service_with_plan_stubs()
|
||
from packages.domain.edit_plan import EditPlan
|
||
|
||
tpl = svc.create_template(name="非草稿模板", editing_mode="one_take")
|
||
|
||
normal_plan = EditPlan.create(
|
||
template_id=tpl.id,
|
||
name="普通计划",
|
||
config={"is_template_draft": False},
|
||
)
|
||
normal_plan.id = "plan-not-draft"
|
||
svc._plan_repo.create(normal_plan)
|
||
|
||
import pytest
|
||
|
||
with pytest.raises(ValueError, match="不是模板草稿"):
|
||
svc.publish_template_from_draft(tpl.id, normal_plan.id)
|