Compare commits

...

1 Commits

Author SHA1 Message Date
CI Bot 28745c401d test(wave104): extract template_clip_converter domain module + 62 unit tests
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 18s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m15s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m6s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 55s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 16s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 11s
CI/CD Pipeline / Validate - Code Quality (pull_request) Failing after 1m49s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 56s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m5s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 2m23s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
AI Code Review / AI Code Review (pull_request) Successful in 2m2s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m12s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 2m27s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m2s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Has been cancelled
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 11s
从 edit_template_service.py (959→813行, -15%) 抽出模板片段转换纯逻辑:
- safe_parse_transition_effect / safe_parse_clip_type 安全枚举解析
- filter_clip_config / filter_plan_config_to_template 字段过滤
- clip_to_template_clip_config / clips_to_template_clip_configs 片段转换
- clip_config_to_snapshot / clip_configs_to_snapshots 快照序列化
- snapshot_to_template_clip_config / snapshots_to_template_clip_configs 快照反序列化
- validate_template_name 名称校验

消除 3 处重复代码 (save_plan_as_template / publish_template_from_draft / rollback_to_version)
验证: 114 passed (52原有 + 62新增),全套模板相关 442 passed
2026-07-26 22:10:38 +08:00
3 changed files with 788 additions and 161 deletions
+15 -161
View File
@@ -23,6 +23,13 @@ from packages.domain.template_clip_config import (
TemplateClipConfig,
TransitionEffect,
)
from packages.domain.template_clip_converter import (
clip_configs_to_snapshots,
clips_to_template_clip_configs,
filter_plan_config_to_template,
snapshots_to_template_clip_configs,
validate_template_name,
)
logger = logging.getLogger(__name__)
@@ -121,9 +128,7 @@ class EditTemplateService:
ValueError: 名称为空或重复
"""
# 名称校验
clean_name = name.strip()
if not clean_name:
raise ValueError("模板名称不能为空")
clean_name = validate_template_name(name)
# 名称重复检查
existing = self._template_repo.list_all(skip=0, limit=1000)
@@ -471,12 +476,7 @@ class EditTemplateService:
raise ValueError(f"模板名称已存在: {clean_name}")
# 从计划 config 中提取模板级配置,去掉运行时/素材相关字段
plan_config = plan.config or {}
template_config: dict[str, Any] = {}
for key, value in plan_config.items():
# 跳过明显的运行时/实例字段,保留风格/模式类配置
if key not in {"asset_ids", "source_edit_plan_id", "generation_task_id"}:
template_config[key] = value
template_config = filter_plan_config_to_template(plan.config)
template = EditTemplate.create(
name=clean_name,
@@ -497,40 +497,7 @@ class EditTemplateService:
# 5. 转换每个片段为模板片段配置
created_configs: List[TemplateClipConfig] = []
for clip in clips:
clip_config: dict[str, Any] = {}
# 播放速度存入 config
if clip.playback_speed and clip.playback_speed != 1.0:
clip_config["playback_speed"] = clip.playback_speed
# 片段自有 config 合并(优先级:clip.config 覆盖上面的)
if clip.config:
clip_config.update(clip.config)
# 去掉素材相关字段
clip_config.pop("asset_info", None)
clip_config.pop("source_asset_id", None)
# 转场效果兼容校验
try:
transition = TransitionEffect(clip.transition_effect)
except ValueError:
transition = TransitionEffect.CUT
# 片段类型兼容校验
try:
clip_type = ClipType(clip.clip_type)
except ValueError:
clip_type = ClipType.MAIN
clip_config_obj = TemplateClipConfig.create(
template_id=created_template.id,
clip_type=clip_type,
order=clip.order,
min_duration=clip.duration,
max_duration=clip.duration,
text_template=clip.text_content or "",
transition_effect=transition,
config=clip_config,
)
for clip_config_obj in clips_to_template_clip_configs(created_template.id, clips):
created = self._clip_config_repo.create(clip_config_obj)
created_configs.append(created)
@@ -679,8 +646,6 @@ class EditTemplateService:
Raises:
ValueError: 模板/草稿不存在,或草稿不属于该模板
"""
from packages.domain.template_clip_config import TemplateClipConfig
# 1. 校验模板和草稿
template = self.get_template_or_raise(template_id)
draft = self._plan_repo.get(draft_plan_id)
@@ -700,39 +665,14 @@ class EditTemplateService:
editing_mode = config.get("editing_mode", "one_take")
# 4. 提取模板配置(去掉草稿/运行时字段)
draft_config = draft.config or {}
template_config: dict[str, Any] = {}
skip_keys = {
"is_template_draft",
"asset_ids",
"source_edit_plan_id",
"generation_task_id",
}
for key, value in draft_config.items():
if key not in skip_keys:
template_config[key] = value
template_config = filter_plan_config_to_template(draft.config)
# 5. 事务更新
try:
# 5.0 先保存旧版快照(发布前的状态),用于回滚
old_version = template.version or 1
old_clip_configs = self._clip_config_repo.list_by_template(template_id)
old_clip_snapshots = [
{
"clip_type": cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type,
"order": cfg.order,
"min_duration": cfg.min_duration,
"max_duration": cfg.max_duration,
"text_template": cfg.text_template or "",
"transition_effect": (
cfg.transition_effect.value
if hasattr(cfg.transition_effect, "value")
else cfg.transition_effect
),
"config": cfg.config or {},
}
for cfg in old_clip_configs
]
old_clip_snapshots = clip_configs_to_snapshots(old_clip_configs)
from packages.domain.template_version import EditTemplateVersion
@@ -759,46 +699,7 @@ class EditTemplateService:
# 创建新的片段配置
created_configs: list[TemplateClipConfig] = []
for clip in draft_clips:
clip_config: dict[str, Any] = {}
# 播放速度存入 config
if clip.playback_speed and clip.playback_speed != 1.0:
clip_config["playback_speed"] = clip.playback_speed
# 片段自有 config 合并
if clip.config:
clip_config.update(clip.config)
# 去掉素材相关字段
clip_config.pop("asset_info", None)
clip_config.pop("source_asset_id", None)
# 转场效果兼容校验
try:
from packages.domain.template_clip_config import (
TransitionEffect,
)
transition = TransitionEffect(clip.transition_effect)
except (ValueError, ImportError):
transition = TransitionEffect.CUT # type: ignore
# 片段类型兼容校验
try:
from packages.domain.template_clip_config import ClipType
clip_type = ClipType(clip.clip_type)
except (ValueError, ImportError):
clip_type = ClipType.MAIN # type: ignore
config_obj = TemplateClipConfig.create(
template_id=template_id,
clip_type=clip_type,
order=clip.order,
min_duration=clip.duration,
max_duration=clip.duration,
text_template=clip.text_content or "",
transition_effect=transition,
config=clip_config,
)
for config_obj in clips_to_template_clip_configs(template_id, draft_clips):
created = self._clip_config_repo.create(config_obj)
created_configs.append(created)
@@ -843,8 +744,6 @@ class EditTemplateService:
Raises:
ValueError: 模板/版本不存在
"""
from packages.domain.template_clip_config import TemplateClipConfig
template = self.get_template_or_raise(template_id)
# 1. 读取目标版本快照
@@ -857,22 +756,7 @@ class EditTemplateService:
try:
# 2. 先保存当前状态快照(当前版本号),确保回滚可撤销
old_clip_configs = self._clip_config_repo.list_by_template(template_id)
old_clip_snapshots = [
{
"clip_type": cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type,
"order": cfg.order,
"min_duration": cfg.min_duration,
"max_duration": cfg.max_duration,
"text_template": cfg.text_template or "",
"transition_effect": (
cfg.transition_effect.value
if hasattr(cfg.transition_effect, "value")
else cfg.transition_effect
),
"config": cfg.config or {},
}
for cfg in old_clip_configs
]
old_clip_snapshots = clip_configs_to_snapshots(old_clip_configs)
from packages.domain.template_version import EditTemplateVersion
@@ -905,37 +789,7 @@ class EditTemplateService:
synchronize_session=False
)
for clip_snap in target_version.clip_configs:
# 转场效果兼容校验
try:
from packages.domain.template_clip_config import TransitionEffect
transition = TransitionEffect(clip_snap.get("transition_effect", "cut"))
except (ValueError, ImportError):
from packages.domain.template_clip_config import TransitionEffect
transition = TransitionEffect.CUT
# 片段类型兼容校验
try:
from packages.domain.template_clip_config import ClipType
clip_type = ClipType(clip_snap.get("clip_type", "main"))
except (ValueError, ImportError):
from packages.domain.template_clip_config import ClipType
clip_type = ClipType.MAIN
config_obj = TemplateClipConfig.create(
template_id=template_id,
clip_type=clip_type,
order=clip_snap.get("order", 0),
min_duration=clip_snap.get("min_duration", 0.0),
max_duration=clip_snap.get("max_duration", 0.0),
text_template=clip_snap.get("text_template", ""),
transition_effect=transition,
config=clip_snap.get("config", {}) or {},
)
for config_obj in snapshots_to_template_clip_configs(template_id, target_version.clip_configs):
self._clip_config_repo.create(config_obj)
self._db.commit()
+293
View File
@@ -0,0 +1,293 @@
"""模板片段转换器 — 纯函数集合.
从 edit_template_service.py 抽离的纯逻辑,负责在不同数据形态间转换:
- 剪辑计划片段 (EditPlanClip) → 模板片段配置 (TemplateClipConfig)
- 模板片段配置 → 版本快照 dict
- 版本快照 dict → 模板片段配置
- 计划 config → 模板 config(过滤运行时字段)
所有函数均为纯函数,不依赖数据库或外部 IO。
"""
from __future__ import annotations
from typing import Any
from packages.domain.template_clip_config import (
ClipType,
TemplateClipConfig,
TransitionEffect,
)
# ── 安全枚举解析 ────────────────────────────────────────────────────────────
def safe_parse_transition_effect(value: Any, default: TransitionEffect = TransitionEffect.CUT) -> TransitionEffect:
"""安全解析转场效果枚举,解析失败返回默认值。
Args:
value: 待解析的值(枚举、字符串或其他)
default: 解析失败时的默认值
Returns:
TransitionEffect 枚举值
"""
if isinstance(value, TransitionEffect):
return value
try:
return TransitionEffect(value)
except (ValueError, TypeError):
return default
def safe_parse_clip_type(value: Any, default: ClipType = ClipType.MAIN) -> ClipType:
"""安全解析片段类型枚举,解析失败返回默认值。
Args:
value: 待解析的值(枚举、字符串或其他)
default: 解析失败时的默认值
Returns:
ClipType 枚举值
"""
if isinstance(value, ClipType):
return value
try:
return ClipType(value)
except (ValueError, TypeError):
return default
# ── Config 字段过滤 ─────────────────────────────────────────────────────────
# 默认需要从 clip config 中移除的素材/运行时字段
_DEFAULT_CLIP_CONFIG_SKIP_KEYS = frozenset(
{
"asset_info",
"source_asset_id",
}
)
# 默认需要从 plan config 中移除的运行时/实例字段
_DEFAULT_PLAN_CONFIG_SKIP_KEYS = frozenset(
{
"is_template_draft",
"asset_ids",
"source_edit_plan_id",
"generation_task_id",
}
)
def filter_clip_config(
clip_config: dict[str, Any] | None,
playback_speed: float | None = None,
skip_keys: frozenset[str] | None = None,
) -> dict[str, Any]:
"""构建模板片段的 config 字典。
处理逻辑:
1. 如果 playback_speed 存在且不等于 1.0,加入 config
2. 合并 clip 自身的 config
3. 移除素材相关字段
Args:
clip_config: 原始片段 config(可为 None
playback_speed: 播放速度(可选,1.0 时不写入)
skip_keys: 需要跳过的字段集合(None 时用默认)
Returns:
过滤后的 config 字典
"""
skip = skip_keys if skip_keys is not None else _DEFAULT_CLIP_CONFIG_SKIP_KEYS
result: dict[str, Any] = {}
if playback_speed is not None and playback_speed != 1.0:
result["playback_speed"] = playback_speed
if clip_config:
result.update(clip_config)
for key in skip:
result.pop(key, None)
return result
def filter_plan_config_to_template(
plan_config: dict[str, Any] | None,
skip_keys: frozenset[str] | None = None,
) -> dict[str, Any]:
"""从计划 config 中提取模板 config(过滤运行时/实例字段)。
Args:
plan_config: 原始计划 config(可为 None
skip_keys: 需要跳过的字段集合(None 时用默认)
Returns:
过滤后的模板 config
"""
skip = skip_keys if skip_keys is not None else _DEFAULT_PLAN_CONFIG_SKIP_KEYS
if not plan_config:
return {}
return {k: v for k, v in plan_config.items() if k not in skip}
# ── Clip → TemplateClipConfig 转换 ────────────────────────────────────────
def clip_to_template_clip_config(
template_id: str,
clip: Any,
) -> TemplateClipConfig:
"""将剪辑计划片段转换为模板片段配置。
转换规则:
- clip_type → 安全解析后映射
- order → 保持不变
- duration → min_duration = max_duration = duration(固定时长)
- text_content → text_template
- transition_effect → 安全解析后映射
- playback_speed → 存入 config(非 1.0 时)
- clip.config → 合并入 config(过滤素材字段)
Args:
template_id: 目标模板 ID
clip: 源片段对象(需有 clip_type/order/duration/text_content/
transition_effect/playback_speed/config 属性)
Returns:
新创建的 TemplateClipConfig 实例
"""
clip_type = safe_parse_clip_type(getattr(clip, "clip_type", None))
transition = safe_parse_transition_effect(getattr(clip, "transition_effect", None))
config = filter_clip_config(
getattr(clip, "config", None),
playback_speed=getattr(clip, "playback_speed", None),
)
duration = getattr(clip, "duration", 0.0) or 0.0
return TemplateClipConfig.create(
template_id=template_id,
clip_type=clip_type,
order=getattr(clip, "order", 0),
min_duration=duration,
max_duration=duration,
text_template=getattr(clip, "text_content", "") or "",
transition_effect=transition,
config=config,
)
def clips_to_template_clip_configs(
template_id: str,
clips: list[Any],
) -> list[TemplateClipConfig]:
"""批量将剪辑计划片段转换为模板片段配置列表。
Args:
template_id: 目标模板 ID
clips: 源片段对象列表
Returns:
TemplateClipConfig 实例列表
"""
return [clip_to_template_clip_config(template_id, c) for c in clips]
# ── TemplateClipConfig → Snapshot 转换 ────────────────────────────────────
def _enum_value(value: Any) -> Any:
"""获取枚举的 value 值(兼容枚举和字符串)。"""
if hasattr(value, "value"):
return value.value
return value
def clip_config_to_snapshot(cfg: Any) -> dict[str, Any]:
"""将模板片段配置转换为版本快照 dict。
Args:
cfg: TemplateClipConfig 对象(或有对应属性的对象)
Returns:
快照字典,包含 clip_type/order/min_duration/max_duration/
text_template/transition_effect/config
"""
return {
"clip_type": _enum_value(getattr(cfg, "clip_type", None)),
"order": getattr(cfg, "order", 0),
"min_duration": getattr(cfg, "min_duration", 0.0),
"max_duration": getattr(cfg, "max_duration", 0.0),
"text_template": getattr(cfg, "text_template", "") or "",
"transition_effect": _enum_value(getattr(cfg, "transition_effect", None)),
"config": dict(getattr(cfg, "config", {}) or {}),
}
def clip_configs_to_snapshots(configs: list[Any]) -> list[dict[str, Any]]:
"""批量将模板片段配置转换为版本快照列表。"""
return [clip_config_to_snapshot(c) for c in configs]
# ── Snapshot → TemplateClipConfig 转换 ────────────────────────────────────
def snapshot_to_template_clip_config(
template_id: str,
snapshot: dict[str, Any],
) -> TemplateClipConfig:
"""将版本快照 dict 转换为模板片段配置。
Args:
template_id: 目标模板 ID
snapshot: 快照字典
Returns:
新创建的 TemplateClipConfig 实例
"""
clip_type = safe_parse_clip_type(snapshot.get("clip_type", "main"))
transition = safe_parse_transition_effect(snapshot.get("transition_effect", "cut"))
return TemplateClipConfig.create(
template_id=template_id,
clip_type=clip_type,
order=snapshot.get("order", 0),
min_duration=snapshot.get("min_duration", 0.0),
max_duration=snapshot.get("max_duration", 0.0),
text_template=snapshot.get("text_template", ""),
transition_effect=transition,
config=dict(snapshot.get("config", {}) or {}),
)
def snapshots_to_template_clip_configs(
template_id: str,
snapshots: list[dict[str, Any]],
) -> list[TemplateClipConfig]:
"""批量将版本快照转换为模板片段配置列表。"""
return [snapshot_to_template_clip_config(template_id, s) for s in snapshots]
# ── 名称校验工具 ──────────────────────────────────────────────────────────
def validate_template_name(name: str | None) -> str:
"""校验并清洗模板名称。
Args:
name: 原始名称
Returns:
清洗后的名称(去除首尾空格)
Raises:
ValueError: 名称为空
"""
clean_name = name.strip() if name else ""
if not clean_name:
raise ValueError("模板名称不能为空")
return clean_name
+480
View File
@@ -0,0 +1,480 @@
"""template_clip_converter 模块单元测试."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import pytest
from packages.domain.template_clip_config import ClipType, TransitionEffect
from packages.domain.template_clip_converter import (
clip_config_to_snapshot,
clip_configs_to_snapshots,
clip_to_template_clip_config,
clips_to_template_clip_configs,
filter_clip_config,
filter_plan_config_to_template,
safe_parse_clip_type,
safe_parse_transition_effect,
snapshot_to_template_clip_config,
snapshots_to_template_clip_configs,
validate_template_name,
)
# ── 辅助数据类 ──────────────────────────────────────────────────────────────
@dataclass
class FakeClip:
"""模拟剪辑计划片段对象."""
clip_type: Any = "main"
order: int = 0
duration: float = 5.0
text_content: str = ""
transition_effect: Any = "cut"
playback_speed: float | None = None
config: dict[str, Any] | None = None
@dataclass
class FakeClipConfig:
"""模拟模板片段配置对象."""
clip_type: Any = "main"
order: int = 0
min_duration: float = 0.0
max_duration: float = 0.0
text_template: str = ""
transition_effect: Any = "cut"
config: dict[str, Any] | None = None
# ── safe_parse_transition_effect ────────────────────────────────────────────
class TestSafeParseTransitionEffect:
def test_enum_value_passthrough(self):
assert safe_parse_transition_effect(TransitionEffect.DISSOLVE) == TransitionEffect.DISSOLVE
def test_valid_string(self):
assert safe_parse_transition_effect("dissolve") == TransitionEffect.DISSOLVE
def test_invalid_string_defaults_to_cut(self):
assert safe_parse_transition_effect("invalid_effect") == TransitionEffect.CUT
def test_none_defaults_to_cut(self):
assert safe_parse_transition_effect(None) == TransitionEffect.CUT
def test_custom_default(self):
assert safe_parse_transition_effect("bad", default=TransitionEffect.FADE) == TransitionEffect.FADE
def test_int_value(self):
assert safe_parse_transition_effect(123) == TransitionEffect.CUT
# ── safe_parse_clip_type ────────────────────────────────────────────────────
class TestSafeParseClipType:
def test_enum_value_passthrough(self):
assert safe_parse_clip_type(ClipType.INTRO) == ClipType.INTRO
def test_valid_string(self):
assert safe_parse_clip_type("intro") == ClipType.INTRO
def test_invalid_string_defaults_to_main(self):
assert safe_parse_clip_type("invalid_type") == ClipType.MAIN
def test_none_defaults_to_main(self):
assert safe_parse_clip_type(None) == ClipType.MAIN
def test_custom_default(self):
assert safe_parse_clip_type("bad", default=ClipType.OUTRO) == ClipType.OUTRO
def test_int_value(self):
assert safe_parse_clip_type(42) == ClipType.MAIN
# ── filter_clip_config ──────────────────────────────────────────────────────
class TestFilterClipConfig:
def test_none_config_no_speed(self):
result = filter_clip_config(None)
assert result == {}
def test_empty_config_no_speed(self):
result = filter_clip_config({})
assert result == {}
def test_playback_speed_added_when_not_default(self):
result = filter_clip_config(None, playback_speed=1.5)
assert result == {"playback_speed": 1.5}
def test_playback_speed_skipped_when_default(self):
result = filter_clip_config(None, playback_speed=1.0)
assert result == {}
def test_playback_speed_none_skipped(self):
result = filter_clip_config(None, playback_speed=None)
assert result == {}
def test_config_merged(self):
result = filter_clip_config({"filter": "vintage", "intensity": 0.5})
assert result == {"filter": "vintage", "intensity": 0.5}
def test_asset_info_removed(self):
result = filter_clip_config({"asset_info": {"name": "test.mp4"}, "filter": "vintage"})
assert "asset_info" not in result
assert result["filter"] == "vintage"
def test_source_asset_id_removed(self):
result = filter_clip_config({"source_asset_id": "abc123", "filter": "vintage"})
assert "source_asset_id" not in result
assert result["filter"] == "vintage"
def test_speed_overrides_config_playback_speed(self):
result = filter_clip_config({"playback_speed": 2.0}, playback_speed=0.5)
assert result["playback_speed"] == 2.0 # config 优先级更高
def test_custom_skip_keys(self):
skip = frozenset({"custom_field"})
result = filter_clip_config(
{"custom_field": "x", "asset_info": "keep_it"},
skip_keys=skip,
)
assert "custom_field" not in result
assert "asset_info" in result # 自定义 skip 覆盖默认
# ── filter_plan_config_to_template ──────────────────────────────────────────
class TestFilterPlanConfigToTemplate:
def test_none_config(self):
assert filter_plan_config_to_template(None) == {}
def test_empty_config(self):
assert filter_plan_config_to_template({}) == {}
def test_draft_flag_removed(self):
result = filter_plan_config_to_template({"is_template_draft": True, "editing_mode": "one_take"})
assert "is_template_draft" not in result
assert result["editing_mode"] == "one_take"
def test_asset_ids_removed(self):
result = filter_plan_config_to_template({"asset_ids": ["a", "b"], "resolution": "1080p"})
assert "asset_ids" not in result
assert result["resolution"] == "1080p"
def test_source_edit_plan_id_removed(self):
result = filter_plan_config_to_template({"source_edit_plan_id": "plan123", "theme": "dark"})
assert "source_edit_plan_id" not in result
assert result["theme"] == "dark"
def test_generation_task_id_removed(self):
result = filter_plan_config_to_template({"generation_task_id": "task123", "bgm": "on"})
assert "generation_task_id" not in result
assert result["bgm"] == "on"
def test_normal_fields_preserved(self):
config = {
"editing_mode": "pip",
"resolution": "720p",
"duration": 30,
"style": "cinematic",
}
result = filter_plan_config_to_template(config)
assert result == config
def test_custom_skip_keys(self):
skip = frozenset({"secret_field"})
result = filter_plan_config_to_template(
{"secret_field": "x", "is_template_draft": "keep"},
skip_keys=skip,
)
assert "secret_field" not in result
assert "is_template_draft" in result
# ── clip_to_template_clip_config ────────────────────────────────────────────
class TestClipToTemplateClipConfig:
def test_basic_conversion(self):
clip = FakeClip(
clip_type="main",
order=2,
duration=3.5,
text_content="Hello world",
transition_effect="dissolve",
)
result = clip_to_template_clip_config("tmpl_001", clip)
assert result.template_id == "tmpl_001"
assert result.clip_type == ClipType.MAIN
assert result.order == 2
assert result.min_duration == 3.5
assert result.max_duration == 3.5
assert result.text_template == "Hello world"
assert result.transition_effect == TransitionEffect.DISSOLVE
def test_playback_speed_in_config(self):
clip = FakeClip(playback_speed=2.0)
result = clip_to_template_clip_config("tmpl_001", clip)
assert result.config["playback_speed"] == 2.0
def test_default_speed_not_in_config(self):
clip = FakeClip(playback_speed=1.0)
result = clip_to_template_clip_config("tmpl_001", clip)
assert "playback_speed" not in result.config
def test_config_preserved_and_filtered(self):
clip = FakeClip(config={"filter": "vintage", "asset_info": {"id": "x"}})
result = clip_to_template_clip_config("tmpl_001", clip)
assert result.config["filter"] == "vintage"
assert "asset_info" not in result.config
def test_text_content_none_becomes_empty(self):
clip = FakeClip(text_content=None)
result = clip_to_template_clip_config("tmpl_001", clip)
assert result.text_template == ""
def test_invalid_type_falls_back(self):
clip = FakeClip(clip_type="nonexistent")
result = clip_to_template_clip_config("tmpl_001", clip)
assert result.clip_type == ClipType.MAIN
def test_invalid_transition_falls_back(self):
clip = FakeClip(transition_effect="nonexistent")
result = clip_to_template_clip_config("tmpl_001", clip)
assert result.transition_effect == TransitionEffect.CUT
def test_enum_type_input(self):
clip = FakeClip(clip_type=ClipType.INTRO, transition_effect=TransitionEffect.FADE)
result = clip_to_template_clip_config("tmpl_001", clip)
assert result.clip_type == ClipType.INTRO
assert result.transition_effect == TransitionEffect.FADE
def test_duration_none_defaults_zero(self):
clip = FakeClip(duration=None)
result = clip_to_template_clip_config("tmpl_001", clip)
assert result.min_duration == 0.0
assert result.max_duration == 0.0
class TestClipsToTemplateClipConfigs:
def test_empty_list(self):
result = clips_to_template_clip_configs("tmpl_001", [])
assert result == []
def test_multiple_clips(self):
clips = [
FakeClip(clip_type="intro", order=0, duration=2.0),
FakeClip(clip_type="main", order=1, duration=5.0),
FakeClip(clip_type="outro", order=2, duration=3.0),
]
result = clips_to_template_clip_configs("tmpl_001", clips)
assert len(result) == 3
assert result[0].clip_type == ClipType.INTRO
assert result[1].clip_type == ClipType.MAIN
assert result[2].clip_type == ClipType.OUTRO
assert all(r.template_id == "tmpl_001" for r in result)
# ── clip_config_to_snapshot ────────────────────────────────────────────────
class TestClipConfigToSnapshot:
def test_basic_snapshot(self):
cfg = FakeClipConfig(
clip_type="main",
order=1,
min_duration=2.0,
max_duration=5.0,
text_template="hello",
transition_effect="dissolve",
config={"filter": "vintage"},
)
snap = clip_config_to_snapshot(cfg)
assert snap["clip_type"] == "main"
assert snap["order"] == 1
assert snap["min_duration"] == 2.0
assert snap["max_duration"] == 5.0
assert snap["text_template"] == "hello"
assert snap["transition_effect"] == "dissolve"
assert snap["config"] == {"filter": "vintage"}
def test_enum_values_converted_to_strings(self):
cfg = FakeClipConfig(
clip_type=ClipType.INTRO,
transition_effect=TransitionEffect.FADE,
)
snap = clip_config_to_snapshot(cfg)
assert snap["clip_type"] == "intro"
assert snap["transition_effect"] == "fade"
def test_none_text_template_becomes_empty(self):
cfg = FakeClipConfig(text_template=None)
snap = clip_config_to_snapshot(cfg)
assert snap["text_template"] == ""
def test_none_config_becomes_empty_dict(self):
cfg = FakeClipConfig(config=None)
snap = clip_config_to_snapshot(cfg)
assert snap["config"] == {}
def test_config_is_copy_not_reference(self):
original = {"key": "value"}
cfg = FakeClipConfig(config=original)
snap = clip_config_to_snapshot(cfg)
snap["config"]["key"] = "modified"
assert original["key"] == "value"
class TestClipConfigsToSnapshots:
def test_empty_list(self):
assert clip_configs_to_snapshots([]) == []
def test_multiple_configs(self):
configs = [
FakeClipConfig(clip_type="intro", order=0),
FakeClipConfig(clip_type="main", order=1),
]
result = clip_configs_to_snapshots(configs)
assert len(result) == 2
assert result[0]["clip_type"] == "intro"
assert result[1]["order"] == 1
# ── snapshot_to_template_clip_config ────────────────────────────────────────
class TestSnapshotToTemplateClipConfig:
def test_basic_conversion(self):
snap = {
"clip_type": "intro",
"order": 2,
"min_duration": 1.0,
"max_duration": 3.0,
"text_template": "hi",
"transition_effect": "dissolve",
"config": {"filter": "bw"},
}
result = snapshot_to_template_clip_config("tmpl_001", snap)
assert result.template_id == "tmpl_001"
assert result.clip_type == ClipType.INTRO
assert result.order == 2
assert result.min_duration == 1.0
assert result.max_duration == 3.0
assert result.text_template == "hi"
assert result.transition_effect == TransitionEffect.DISSOLVE
assert result.config == {"filter": "bw"}
def test_missing_fields_get_defaults(self):
result = snapshot_to_template_clip_config("tmpl_001", {})
assert result.clip_type == ClipType.MAIN
assert result.order == 0
assert result.min_duration == 0.0
assert result.max_duration == 0.0
assert result.text_template == ""
assert result.transition_effect == TransitionEffect.CUT
assert result.config == {}
def test_invalid_type_falls_back(self):
snap = {"clip_type": "invalid"}
result = snapshot_to_template_clip_config("tmpl_001", snap)
assert result.clip_type == ClipType.MAIN
def test_invalid_transition_falls_back(self):
snap = {"transition_effect": "invalid"}
result = snapshot_to_template_clip_config("tmpl_001", snap)
assert result.transition_effect == TransitionEffect.CUT
def test_none_config_becomes_empty_dict(self):
snap = {"config": None}
result = snapshot_to_template_clip_config("tmpl_001", snap)
assert result.config == {}
class TestSnapshotsToTemplateClipConfigs:
def test_empty_list(self):
assert snapshots_to_template_clip_configs("tmpl_001", []) == []
def test_multiple_snapshots(self):
snaps = [
{"clip_type": "intro", "order": 0},
{"clip_type": "outro", "order": 2},
]
result = snapshots_to_template_clip_configs("tmpl_001", snaps)
assert len(result) == 2
assert result[0].clip_type == ClipType.INTRO
assert result[1].clip_type == ClipType.OUTRO
assert all(r.template_id == "tmpl_001" for r in result)
# ── 往返一致性测试 ──────────────────────────────────────────────────────────
class TestRoundTrip:
def test_snapshot_clip_config_roundtrip(self):
"""snapshot → TemplateClipConfig → snapshot 应保持一致."""
original = {
"clip_type": "intro",
"order": 3,
"min_duration": 1.5,
"max_duration": 4.0,
"text_template": "test text",
"transition_effect": "dissolve",
"config": {"key": "value", "nested": {"a": 1}},
}
cfg = snapshot_to_template_clip_config("tmpl_test", original)
result = clip_config_to_snapshot(cfg)
assert result == original
def test_clip_to_config_to_snapshot(self):
"""clip → TemplateClipConfig → snapshot 的预期结果."""
clip = FakeClip(
clip_type="main",
order=1,
duration=5.0,
text_content="hello",
transition_effect="fade",
playback_speed=1.5,
config={"filter": "vintage", "asset_info": "should_remove"},
)
cfg = clip_to_template_clip_config("tmpl_001", clip)
snap = clip_config_to_snapshot(cfg)
assert snap["clip_type"] == "main"
assert snap["order"] == 1
assert snap["min_duration"] == 5.0
assert snap["max_duration"] == 5.0
assert snap["text_template"] == "hello"
assert snap["transition_effect"] == "fade"
assert snap["config"]["playback_speed"] == 1.5
assert snap["config"]["filter"] == "vintage"
assert "asset_info" not in snap["config"]
# ── validate_template_name ──────────────────────────────────────────────────
class TestValidateTemplateName:
def test_valid_name(self):
assert validate_template_name("My Template") == "My Template"
def test_strips_whitespace(self):
assert validate_template_name(" Hello ") == "Hello"
def test_empty_string_raises(self):
with pytest.raises(ValueError, match="名称不能为空"):
validate_template_name("")
def test_whitespace_only_raises(self):
with pytest.raises(ValueError, match="名称不能为空"):
validate_template_name(" ")
def test_none_raises(self):
with pytest.raises(ValueError, match="名称不能为空"):
validate_template_name(None)