diff --git a/tests/unit/test_edit_plan_service.py b/tests/unit/test_edit_plan_service.py
new file mode 100644
index 000000000..25ae58095
--- /dev/null
+++ b/tests/unit/test_edit_plan_service.py
@@ -0,0 +1,569 @@
+"""
+EditPlanService 单元测试
+
+覆盖（35+ 测试用例）：
+- 计划 CRUD：创建、查询、更新、删除
+- 状态机流转：合法流转、非法流转、幂等流转
+- 片段管理：创建、更新、删除、分配素材
+- 渲染生成流程：can_generate、mark_clips_ready、get_generation_status
+- 异常处理：不存在、参数校验
+"""
+
+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
+
+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
+
+
+# ---------------------------------------------------------------------------
+# 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 list_all(
+ self,
+ *,
+ status: Optional[EditPlanStatus] = None,
+ skip: int = 0,
+ limit: int = 50,
+ ) -> List[EditPlan]:
+ items = list(self._plans.values())
+ if status:
+ items = [p for p in items if p.status == status]
+ return items[skip : skip + limit]
+
+ def list_by_template(
+ self,
+ template_id: str,
+ *,
+ status: Optional[EditPlanStatus] = 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 == status]
+ return items[skip : skip + limit]
+
+ 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,
+ config=plan.config,
+ created_at=datetime.now(timezone.utc),
+ updated_at=datetime.now(timezone.utc),
+ )
+ self._plans[plan.id] = plan
+ return plan
+
+ def update(self, plan: EditPlan) -> EditPlan:
+ self._plans[plan.id] = plan
+ return plan
+
+ def delete(self, plan_id: str) -> bool:
+ return self._plans.pop(plan_id, None) is not None
+
+ def count(
+ self,
+ *,
+ status: Optional[EditPlanStatus] = None,
+ ) -> int:
+ items = list(self._plans.values())
+ if status:
+ items = [p for p in items if p.status == status]
+ return len(items)
+
+
+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[EditPlanClipStatus] = 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 == status]
+ items.sort(key=lambda c: c.order)
+ return items[skip : skip + limit]
+
+ 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,
+ clip_type=clip.clip_type,
+ order=clip.order,
+ template_clip_config_id=clip.template_clip_config_id,
+ 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=datetime.now(timezone.utc),
+ updated_at=datetime.now(timezone.utc),
+ )
+ self._clips[clip.id] = clip
+ return clip
+
+ def update(self, clip: EditPlanClip) -> EditPlanClip:
+ self._clips[clip.id] = clip
+ return clip
+
+ def delete(self, clip_id: str) -> bool:
+ return self._clips.pop(clip_id, None) is not None
+
+ def delete_by_plan(self, plan_id: str) -> int:
+ ids = [cid for cid, c in self._clips.items() if c.plan_id == plan_id]
+ for cid in ids:
+ del self._clips[cid]
+ return len(ids)
+
+ def count(
+ self,
+ *,
+ plan_id: Optional[str] = None,
+ status: Optional[EditPlanClipStatus] = None,
+ ) -> int:
+ items = list(self._clips.values())
+ if plan_id:
+ items = [c for c in items if c.plan_id == plan_id]
+ if status:
+ items = [c for c in items if c.status == status]
+ return len(items)
+
+
+class StubGenerationTaskRepository:
+ """内存中的 GenerationTask 仓储 stub"""
+
+ def __init__(self) -> None:
+ self._tasks: dict[str, Any] = {}
+
+ def get(self, task_id: str) -> Optional[Any]:
+ return self._tasks.get(task_id)
+
+ def create(self, task: Any) -> Any:
+ self._tasks[task.id] = task
+ return task
+
+
+# ---------------------------------------------------------------------------
+# Service factory
+# ---------------------------------------------------------------------------
+
+
+def _make_service():
+ """创建使用 stub 仓储的 EditPlanService"""
+ from app.services.edit_plan_service import EditPlanService
+
+ db = MagicMock()
+ svc = EditPlanService(db)
+ svc._plan_repo = StubEditPlanRepository()
+ svc._clip_repo = StubEditPlanClipRepository()
+ svc._generation_task_repo = StubGenerationTaskRepository()
+ return svc
+
+
+# ===========================================================================
+# 计划 CRUD 测试
+# ===========================================================================
+
+
+class TestEditPlanServiceCRUD:
+ """计划 CRUD 测试"""
+
+ def test_create_plan_success(self):
+ svc = _make_service()
+ plan = svc.create_plan(template_id="tpl-001", name="测试计划")
+ assert plan.name == "测试计划"
+ assert plan.template_id == "tpl-001"
+ assert plan.status == EditPlanStatus.DRAFT
+ assert plan.id
+
+ def test_get_plan(self):
+ svc = _make_service()
+ created = svc.create_plan("tpl-001", "查询测试")
+ fetched = svc.get_plan(created.id)
+ assert fetched is not None
+ assert fetched.id == created.id
+
+ def test_get_plan_returns_none(self):
+ svc = _make_service()
+ assert svc.get_plan("nonexistent") is None
+
+ def test_get_plan_or_raise(self):
+ svc = _make_service()
+ created = svc.create_plan("tpl-001", "查询测试")
+ fetched = svc.get_plan_or_raise(created.id)
+ assert fetched.id == created.id
+
+ def test_get_plan_or_raise_not_found(self):
+ svc = _make_service()
+ with pytest.raises(ValueError, match="剪辑计划不存在"):
+ svc.get_plan_or_raise("nonexistent")
+
+ def test_list_plans(self):
+ svc = _make_service()
+ svc.create_plan("tpl-001", "计划1")
+ svc.create_plan("tpl-001", "计划2")
+ result = svc.list_plans()
+ assert len(result) == 2
+
+ def test_list_plans_by_template(self):
+ svc = _make_service()
+ svc.create_plan("tpl-001", "计划1")
+ svc.create_plan("tpl-002", "计划2")
+ result = svc.list_plans(template_id="tpl-001")
+ assert len(result) == 1
+ assert result[0].name == "计划1"
+
+ def test_list_plans_by_status(self):
+ svc = _make_service()
+ p1 = svc.create_plan("tpl-001", "计划1")
+ svc.create_plan("tpl-001", "计划2")
+ svc.transition_status(p1.id, EditPlanStatus.EDITING)
+ result = svc.list_plans(status=EditPlanStatus.EDITING)
+ assert len(result) == 1
+
+ def test_count_plans(self):
+ svc = _make_service()
+ svc.create_plan("tpl-001", "计划1")
+ svc.create_plan("tpl-001", "计划2")
+ assert svc.count_plans() == 2
+
+ def test_update_plan_name(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "原名")
+ updated = svc.update_plan(p.id, name="新名")
+ assert updated.name == "新名"
+
+ def test_update_plan_not_found_raises(self):
+ svc = _make_service()
+ with pytest.raises(ValueError, match="剪辑计划不存在"):
+ svc.update_plan("nonexistent", name="新名")
+
+ def test_delete_plan(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "要删除")
+ assert svc.delete_plan(p.id) is True
+ assert svc.get_plan(p.id) is None
+
+ def test_delete_plan_not_found(self):
+ svc = _make_service()
+ assert svc.delete_plan("nonexistent") is False
+
+ def test_delete_plan_also_deletes_clips(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "带片段")
+ svc.create_clip(p.id, "intro", 0)
+ svc.create_clip(p.id, "main", 1)
+ assert svc.count_clips(p.id) == 2
+ svc.delete_plan(p.id)
+ # 片段应被一并删除
+ assert svc.count_clips(p.id) == 0
+
+
+# ===========================================================================
+# 状态机流转测试
+# ===========================================================================
+
+
+class TestStatusTransitions:
+ """状态机流转测试"""
+
+ def test_transition_draft_to_editing(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ result = svc.transition_status(p.id, EditPlanStatus.EDITING)
+ assert result.status == EditPlanStatus.EDITING
+
+ def test_transition_editing_to_rendering(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ svc.transition_status(p.id, EditPlanStatus.EDITING)
+ result = svc.transition_status(p.id, EditPlanStatus.RENDERING)
+ assert result.status == EditPlanStatus.RENDERING
+
+ def test_transition_rendering_to_completed(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ svc.transition_status(p.id, EditPlanStatus.EDITING)
+ svc.transition_status(p.id, EditPlanStatus.RENDERING)
+ result = svc.transition_status(p.id, EditPlanStatus.COMPLETED)
+ assert result.status == EditPlanStatus.COMPLETED
+
+ def test_transition_rendering_to_failed(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ svc.transition_status(p.id, EditPlanStatus.EDITING)
+ svc.transition_status(p.id, EditPlanStatus.RENDERING)
+ result = svc.transition_status(p.id, EditPlanStatus.FAILED)
+ assert result.status == EditPlanStatus.FAILED
+
+ def test_transition_failed_to_draft(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ svc.transition_status(p.id, EditPlanStatus.EDITING)
+ svc.transition_status(p.id, EditPlanStatus.RENDERING)
+ svc.transition_status(p.id, EditPlanStatus.FAILED)
+ result = svc.transition_status(p.id, EditPlanStatus.DRAFT)
+ assert result.status == EditPlanStatus.DRAFT
+
+ def test_transition_idempotent(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ result = svc.transition_status(p.id, EditPlanStatus.DRAFT)
+ assert result.status == EditPlanStatus.DRAFT
+
+ def test_transition_illegal_raises(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ # draft → completed 是非法的
+ with pytest.raises(ValueError):
+ svc.transition_status(p.id, EditPlanStatus.COMPLETED)
+
+ def test_transition_not_found_raises(self):
+ svc = _make_service()
+ with pytest.raises(ValueError, match="剪辑计划不存在"):
+ svc.transition_status("nonexistent", EditPlanStatus.EDITING)
+
+
+# ===========================================================================
+# 片段管理测试
+# ===========================================================================
+
+
+class TestClipManagement:
+ """片段管理测试"""
+
+ def test_create_clip(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ clip = svc.create_clip(p.id, "intro", 0)
+ assert clip.plan_id == p.id
+ assert clip.clip_type == "intro"
+ assert clip.order == 0
+ assert clip.status == EditPlanClipStatus.PENDING
+
+ def test_create_clip_plan_not_found_raises(self):
+ svc = _make_service()
+ with pytest.raises(ValueError, match="剪辑计划不存在"):
+ svc.create_clip("nonexistent", "intro", 0)
+
+ def test_list_clips(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ svc.create_clip(p.id, "intro", 0)
+ svc.create_clip(p.id, "main", 1)
+ svc.create_clip(p.id, "outro", 2)
+ result = svc.list_clips(p.id)
+ assert len(result) == 3
+
+ def test_list_clips_by_status(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ c1 = svc.create_clip(p.id, "intro", 0)
+ svc.create_clip(p.id, "main", 1)
+ # assign_asset 只设置 asset_id，需要额外 mark_ready 才变 ready
+ svc.assign_asset(c1.id, "asset-001")
+ # 手动 mark_ready
+ clip_obj = svc._clip_repo.get(c1.id)
+ clip_obj.mark_ready()
+ svc._clip_repo.update(clip_obj)
+ result = svc.list_clips(p.id, status=EditPlanClipStatus.READY)
+ assert len(result) == 1
+
+ def test_count_clips(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ svc.create_clip(p.id, "intro", 0)
+ svc.create_clip(p.id, "main", 1)
+ assert svc.count_clips(p.id) == 2
+
+ def test_get_clip(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ clip = svc.create_clip(p.id, "intro", 0)
+ fetched = svc.get_clip(clip.id)
+ assert fetched is not None
+ assert fetched.id == clip.id
+
+ def test_get_clip_or_raise_not_found(self):
+ svc = _make_service()
+ with pytest.raises(ValueError, match="片段不存在"):
+ svc.get_clip_or_raise("nonexistent")
+
+ def test_update_clip(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ clip = svc.create_clip(p.id, "intro", 0, duration=3.0)
+ updated = svc.update_clip(clip.id, duration=5.0)
+ assert updated.duration == 5.0
+
+ def test_update_clip_not_found_raises(self):
+ svc = _make_service()
+ with pytest.raises(ValueError, match="片段不存在"):
+ svc.update_clip("nonexistent", duration=5.0)
+
+ def test_assign_asset(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ clip = svc.create_clip(p.id, "intro", 0)
+ result = svc.assign_asset(clip.id, "asset-001")
+ assert result.asset_id == "asset-001"
+ # assign_asset 只设置 asset_id，不改变状态（状态需 mark_ready 流转）
+ assert result.status == EditPlanClipStatus.PENDING
+
+ def test_assign_asset_empty_raises(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ clip = svc.create_clip(p.id, "intro", 0)
+ with pytest.raises(ValueError):
+ svc.assign_asset(clip.id, "")
+
+ def test_delete_clip(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ clip = svc.create_clip(p.id, "intro", 0)
+ assert svc.delete_clip(clip.id) is True
+ assert svc.get_clip(clip.id) is None
+
+ def test_delete_all_clips(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ svc.create_clip(p.id, "intro", 0)
+ svc.create_clip(p.id, "main", 1)
+ count = svc.delete_all_clips(p.id)
+ assert count == 2
+ assert svc.count_clips(p.id) == 0
+
+
+# ===========================================================================
+# 渲染生成流程测试
+# ===========================================================================
+
+
+class TestGenerationWorkflow:
+ """渲染生成流程测试"""
+
+ def test_can_generate_editing_with_clips(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ svc.transition_status(p.id, EditPlanStatus.EDITING)
+ svc.create_clip(p.id, "intro", 0)
+ can, reason = svc.can_generate(p.id)
+ assert can is True
+ assert reason == ""
+
+ def test_can_generate_draft_fails(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ svc.create_clip(p.id, "intro", 0)
+ can, reason = svc.can_generate(p.id)
+ assert can is False
+ assert "editing" in reason
+
+ def test_can_generate_no_clips_fails(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ svc.transition_status(p.id, EditPlanStatus.EDITING)
+ can, reason = svc.can_generate(p.id)
+ assert can is False
+ assert "没有片段" in reason
+
+ def test_mark_clips_ready(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ svc.create_clip(p.id, "intro", 0)
+ svc.create_clip(p.id, "main", 1)
+ count = svc.mark_clips_ready(p.id)
+ assert count == 2
+ # 验证所有片段都是 ready 状态
+ clips = svc.list_clips(p.id)
+ for c in clips:
+ assert c.status == EditPlanClipStatus.READY
+
+ def test_get_plan_with_clips(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ svc.create_clip(p.id, "intro", 0)
+ svc.create_clip(p.id, "main", 1)
+ result = svc.get_plan_with_clips(p.id)
+ assert result["plan"].id == p.id
+ assert len(result["clips"]) == 2
+
+ def test_get_plan_with_clips_not_found(self):
+ svc = _make_service()
+ with pytest.raises(ValueError, match="剪辑计划不存在"):
+ svc.get_plan_with_clips("nonexistent")
+
+ def test_get_generation_status(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试")
+ svc.create_clip(p.id, "intro", 0)
+ result = svc.get_generation_status(p.id)
+ assert result["plan"].id == p.id
+ assert len(result["clips"]) == 1
+ assert result["generation_task_id"] is None
+
+ def test_update_plan_config(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试", config={"key1": "val1"})
+ updated = svc.update_plan_config(p.id, {"key2": "val2"})
+ assert updated.config["key1"] == "val1"
+ assert updated.config["key2"] == "val2"
+
+ def test_update_plan_config_overwrites(self):
+ svc = _make_service()
+ p = svc.create_plan("tpl-001", "测试", config={"key1": "val1"})
+ updated = svc.update_plan_config(p.id, {"key1": "new_val"})
+ assert updated.config["key1"] == "new_val"
diff --git a/tests/unit/test_edit_template_service.py b/tests/unit/test_edit_template_service.py
new file mode 100644
index 000000000..99692083f
--- /dev/null
+++ b/tests/unit/test_edit_template_service.py
@@ -0,0 +1,466 @@
+"""
+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 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_template import EditTemplate, EditTemplateStatus
+from packages.domain.template_clip_config import ClipType, TemplateClipConfig, TransitionEffect
+
+
+# ---------------------------------------------------------------------------
+# 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) -> 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()
+ t1 = 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