fix(#1774): 模板不存在明确返回404并收敛双表异常降级逻辑 #1779

Merged
auto-approve-bot merged 1 commits from fix/issue-1774-template-404 into develop 2026-09-08 00:30:09 +08:00
9 changed files with 485 additions and 354 deletions
@@ -36,17 +36,11 @@ from app.services.asset_segment_tracker import (
remove_used_segment,
)
from app.services.edit_plan_service import EditPlanService
from app.services.edit_template_service import EditTemplateService
from app.services.edit_template_service import EditTemplateService, TemplateNotFoundError
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
from packages.adapters.sqlalchemy_impl.template_clip_config_repository import (
SQLAlchemyTemplateClipConfigRepository,
)
from packages.adapters.sqlalchemy_impl.template_repository import (
SQLAlchemyTemplateRepository,
)
from packages.domain.plan_generator_utils import (
_calc_random_start_time,
build_scene_segments,
@@ -399,73 +393,42 @@ def _safe_segment_duration(value, default: float) -> float:
def _get_template_segments(
template_id: str,
user_id: str,
tpl_svc: EditTemplateService,
db: Session,
) -> list[tuple[int, float, float]]:
"""获取模板的片段配置(顺序、最短时长、最长时长).
优先从新模板系统(template_clip_configs)查询,
若不存在则回退到旧模板系统(template_segments)。
单一数据源:模板主表为 ``templates``(用户自建,归属 user_id/
``edit_templates``(全局模板库),片段配置主表为 ``template_clip_configs``
(由 ``EditTemplateService.list_clip_configs_for_editor`` 统一读取)。
不再使用"新表抛异常 → 降级直查配置表 → 再降级查 segments"的异常控制流,
也不在正常请求中打印 ``ValueError: 模板不存在`` 堆栈。
Args:
template_id: 模板 ID
user_id: 当前登录用户 ID(用于归属校验)
tpl_svc: 模板编辑器服务
Returns:
[(segment_order, duration_min, duration_max), ...] 按 order 排序
[(segment_order, duration_min, duration_max), ...] 按 order 排序
模板存在但未配置片段时返回空列表。
Raises:
TemplateNotFoundError: 模板不存在、已删除或不归属于当前用户。
"""
# 优先查新模板系统
try:
clip_configs = tpl_svc.list_clip_configs(template_id)
if clip_configs:
result = []
for cc in clip_configs:
dur_min = _safe_segment_duration(cc.min_duration, _DEFAULT_EDITOR_CLIP_DURATION)
dur_max = _safe_segment_duration(
cc.max_duration or cc.min_duration,
_DEFAULT_EDITOR_CLIP_DURATION,
)
dur_min, dur_max = min(dur_min, dur_max), max(dur_min, dur_max)
result.append((cc.order, dur_min, dur_max))
return sorted(result, key=lambda x: x[0])
except Exception:
logger.warning("新模板系统查询clip_configs失败(主表可能不存在),直接查clip_configs表", exc_info=True)
clip_configs = tpl_svc.list_clip_configs_for_editor(template_id, user_id)
# 兜底:直接查 template_clip_configs 表(片段表有 template_id 外键,不依赖模板主表)
try:
direct_repo = SQLAlchemyTemplateClipConfigRepository(db)
direct_configs = direct_repo.list_by_template(template_id)
if direct_configs:
result = []
for cc in direct_configs:
dur_min = _safe_segment_duration(cc.min_duration, _DEFAULT_EDITOR_CLIP_DURATION)
dur_max = _safe_segment_duration(
cc.max_duration or cc.min_duration,
_DEFAULT_EDITOR_CLIP_DURATION,
)
dur_min, dur_max = min(dur_min, dur_max), max(dur_min, dur_max)
result.append((cc.order, dur_min, dur_max))
return sorted(result, key=lambda x: x[0])
except Exception:
logger.warning("直接查clip_configs表也失败,继续回退旧系统", exc_info=True)
# 回退到旧模板系统(template_segments表)
try:
old_repo = SQLAlchemyTemplateRepository(db)
segments = old_repo.list_segments(template_id)
if segments:
result = []
for s in segments:
dur_min = _safe_segment_duration(s.duration_min, _DEFAULT_EDITOR_CLIP_DURATION)
dur_max = _safe_segment_duration(s.duration_max, _DEFAULT_EDITOR_CLIP_DURATION)
dur_min, dur_max = min(dur_min, dur_max), max(dur_min, dur_max)
result.append((s.segment_order, dur_min, dur_max))
return sorted(result, key=lambda x: x[0])
except Exception:
logger.warning("旧模板系统查询segments失败", exc_info=True)
# 所有途径都失败:模板没有片段配置(可能是无效测试模板)
logger.error(
"模板无片段配置:template_id=%s(可能是 is_active=false 的无效模板)",
template_id,
)
return []
result = []
for cc in clip_configs:
dur_min = _safe_segment_duration(cc.min_duration, _DEFAULT_EDITOR_CLIP_DURATION)
dur_max = _safe_segment_duration(
cc.max_duration or cc.min_duration,
_DEFAULT_EDITOR_CLIP_DURATION,
)
dur_min, dur_max = min(dur_min, dur_max), max(dur_min, dur_max)
result.append((cc.order, dur_min, dur_max))
return sorted(result, key=lambda x: x[0])
def _recommended_time_conflicts(
@@ -668,13 +631,21 @@ def create_clips_from_assets_editor(
7. 素材时长为 0 或缺失时报 400,不创建无效片段
"""
tpl_svc, plan_svc = services
user_id = str(current_user.user.id)
# 1. 查询模板 segments
segments = _get_template_segments(template_id, tpl_svc, db)
# 1. 查询模板片段配置。模板不存在/已删除/无权限 → 404;
# 模板存在但确实未配置片段 → 422(配置错误,与 404 区分)。
try:
segments = _get_template_segments(template_id, user_id, tpl_svc)
except TemplateNotFoundError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="模板不存在或无权访问",
) from exc
if not segments:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="模板没有片段配置,无法创建片段",
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="模板未配置片段",
)
# 防御:schema validator 已过滤 null/空串,这里再归一化一次,
@@ -41,29 +41,33 @@ def get_draft_plan_id(
这是模板编辑器路由的核心依赖——所有编辑器端点都先经过这里,
确保 template_id → plan_id 的映射始终存在。
兼容策略:优先从新模板系统(edit_templates 表)查找,
若不存在则回退到旧模板系统(templates 表),确保用户自建模板可用。
模板读取遵循单一数据源、显式判定(不使用异常降级):
- 用户自建模板在旧表 ``templates``(归属 user_idis_active=True);
- 全局模板在新表 ``edit_templates``(无 user_id,全局可读)。
模板不存在、已删除或不归属于当前用户时,一律返回 404。
"""
tpl_svc, plan_svc = services
user_id = str(current_user.user.id)
# 0. 门禁:校验模板存在且可访问(即使草稿已缓存命中也要校验,
# 避免模板被删除/无权访问后仍可通过既有草稿 plan 继续操作)。
old_repo = SQLAlchemyTemplateRepository(db)
old_template = old_repo.get_active(template_id, user_id)
is_global_template = tpl_svc.get_template(template_id) is not None
if old_template is None and not is_global_template:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="模板不存在")
# 1. 草稿已存在 → 直接返回
draft = tpl_svc.get_template_draft(template_id)
if draft is not None:
return draft.id
# 2. 新系统有模板 → 用新服务创建草稿
if tpl_svc.get_template(template_id) is not None:
# 2. 全局模板(新系统)→ 用新服务创建草稿
if is_global_template:
draft = tpl_svc.create_template_draft(template_id, user_id=user_id)
return draft.id
# 3. 回退到旧模板系统templates 表)
old_repo = SQLAlchemyTemplateRepository(db)
old_template = old_repo.get(template_id, user_id=user_id)
if old_template is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="模板不存在")
# 4. 基于旧模板创建草稿计划
# 3. 旧模板(templates 表)→ 基于旧模板创建草稿计划
from app.services.plan_generator_service import PlanGeneratorService
from packages.domain.edit_template import EditTemplate, EditTemplateStatus
@@ -150,7 +150,7 @@ def rollback_template(
try:
tpl = tpl_svc.rollback_to_version(template_id, request.version)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
clip_configs = tpl_svc.list_clip_configs(template_id)
return EditorRollbackResponse(
+65 -1
View File
@@ -34,6 +34,17 @@ from packages.domain.template_clip_converter import (
logger = logging.getLogger(__name__)
class TemplateNotFoundError(Exception):
"""模板不存在、已删除或当前用户无权访问.
"模板存在但无片段配置"区分:路由层应映射为 HTTP 404。
"""
def __init__(self, template_id: str) -> None:
self.template_id = template_id
super().__init__(f"模板不存在: {template_id}")
class EditTemplateService:
"""模板管理服务
@@ -217,7 +228,14 @@ class EditTemplateService:
skip: int = 0,
limit: int = 100,
) -> List[TemplateClipConfig]:
"""列出模板的片段配置"""
"""列出模板的片段配置
注意:本方法要求模板存在于新表 ``edit_templates``(全局模板库),
主要服务于新模板系统的写入/发布路径。用户自建模板存放在旧表
``templates``,不在 ``edit_templates`` 中,读取其片段配置请改用
:meth:`list_clip_configs_for_editor`,后者直接读取片段配置主表
``template_clip_configs``,不依赖新模板主表、也不靠异常降级。
"""
# 确保模板存在
self.get_template_or_raise(template_id)
return self._clip_config_repo.list_by_template(
@@ -227,6 +245,52 @@ class EditTemplateService:
limit=limit,
)
def list_clip_configs_for_editor(
self,
template_id: str,
user_id: str,
*,
clip_type: Optional[ClipType] = None,
skip: int = 0,
limit: int = 100,
) -> List[TemplateClipConfig]:
"""编辑器读取模板片段配置的单一数据源入口.
片段配置主表是 ``template_clip_configs``(直接读取,不抛异常、不降级)。
模板主表按双表现状显式判定,不使用 try/except 控制流:
1. 用户自建模板在旧表 ``templates``(归属 user_id)→ 校验归属与未删除后直接读;
2. 全局模板在新表 ``edit_templates``(无 user_id,全局可读)→ 直接读;
3. 两者都没有 → 模板不存在/无权限,抛 :class:`TemplateNotFoundError`。
Args:
template_id: 模板 ID
user_id: 当前登录用户 ID(用于旧表模板归属校验)
Raises:
TemplateNotFoundError: 模板不存在、已删除或不归属于当前用户。
"""
# 1) 用户自建模板(旧表 templates,归属 user_id
if self._clip_config_repo.template_owned_by(template_id, user_id):
return self._clip_config_repo.list_by_template(
template_id,
clip_type=clip_type,
skip=skip,
limit=limit,
)
# 2) 全局模板(新表 edit_templates,无 user_id,全局可读)
if self._template_repo.get(template_id) is not None:
return self._clip_config_repo.list_by_template(
template_id,
clip_type=clip_type,
skip=skip,
limit=limit,
)
# 3) 两表都没有:不存在 / 已删除 / 无权限
raise TemplateNotFoundError(template_id)
def get_clip_config(self, config_id: str) -> Optional[TemplateClipConfig]:
"""获取片段配置详情"""
return self._clip_config_repo.get(config_id)
@@ -6,7 +6,10 @@ from typing import List, Optional
from sqlalchemy.orm import Session
from packages.adapters.sqlalchemy_impl.models import TemplateClipConfigModel
from packages.adapters.sqlalchemy_impl.models import (
TemplateClipConfigModel,
TemplateModel,
)
from packages.domain.template_clip_config import (
ClipType,
TemplateClipConfig,
@@ -38,6 +41,24 @@ class SQLAlchemyTemplateClipConfigRepository:
models = query.offset(skip).limit(limit).all()
return [self._model_to_entity(m) for m in models]
def template_owned_by(self, template_id: str, user_id: str) -> bool:
"""校验旧模板主表 ``templates`` 中模板归属当前用户且未删除(is_active=True.
片段配置主表 ``template_clip_configs`` 本身没有 user_id 列,
归属关系通过模板主表 ``templates.user_id`` 确定。
新表 ``edit_templates`` 为全局模板库(无 user_id 列),不走此校验。
"""
return (
self.session.query(TemplateModel.id)
.filter(
TemplateModel.id == template_id,
TemplateModel.user_id == user_id,
TemplateModel.is_active.is_(True),
)
.first()
is not None
)
def get(self, config_id: str) -> Optional[TemplateClipConfig]:
"""根据 ID 获取配置"""
model = self.session.query(TemplateClipConfigModel).filter(TemplateClipConfigModel.id == config_id).first()
@@ -120,6 +120,27 @@ class SQLAlchemyTemplateRepository:
template.segments = self.list_segments(template.id)
return template
def get_active(self, template_id: str, user_id: str) -> Optional[Template]:
"""获取归属当前用户且未删除(is_active=True)的模板,否则返回 None.
用于编辑器访问门禁:模板不存在、已软删除或不属于当前用户时返回 None,
由调用方映射为 404。与 :meth:`get` 的区别是额外过滤 is_active。
"""
model = (
self.session.query(TemplateModel)
.filter(
TemplateModel.id == template_id,
TemplateModel.user_id == user_id,
TemplateModel.is_active.is_(True),
)
.first()
)
if model is None:
return None
template = self._model_to_entity(model)
template.segments = self.list_segments(template.id)
return template
def create(self, template: Template) -> Template:
model = TemplateModel(
id=template.id,
+37 -4
View File
@@ -239,8 +239,8 @@ class TestEditorClipsBySegments:
assert not hasattr(mock_plan_svc, "create_clip") or not mock_plan_svc.create_clip.called
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_no_segments_raises_400(self, mock_storage):
"""模板没有 segment 配置时返回 400"""
def test_no_segments_raises_422(self, mock_storage):
"""模板存在但未配置片段时返回 422(配置错误,与 404 区分)"""
from app.api.routes.templates_editor.clips import (
create_clips_from_assets_editor,
)
@@ -264,11 +264,44 @@ class TestEditorClipsBySegments:
current_user=_make_auth_user(),
)
assert exc_info.value.status_code == 400
assert "片段配置" in exc_info.value.detail
assert exc_info.value.status_code == 422
assert "片段" in exc_info.value.detail
# 不应调用替换方法
mock_plan_svc.replace_all_clips_transactional.assert_not_called()
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_template_not_found_raises_404(self, mock_storage):
"""模板不存在/已删除/无权限(服务层抛 TemplateNotFoundError)时返回 404。"""
from app.api.routes.templates_editor.clips import (
create_clips_from_assets_editor,
)
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
from app.services.edit_template_service import TemplateNotFoundError
mock_plan_svc = _make_plan_svc()
mock_asset_repo = MagicMock()
body = ClipsFromAssetsRequest(asset_ids=["a1"])
with patch(
"app.api.routes.templates_editor.clips._get_template_segments",
side_effect=TemplateNotFoundError("tpl-missing"),
):
with pytest.raises(HTTPException) as exc_info:
create_clips_from_assets_editor(
template_id="tpl-missing",
body=body,
background_tasks=MagicMock(),
plan_id=TEST_PLAN_ID,
services=(MagicMock(), mock_plan_svc),
asset_repo=mock_asset_repo,
db=MagicMock(),
current_user=_make_auth_user(),
)
assert exc_info.value.status_code == 404
mock_plan_svc.replace_all_clips_transactional.assert_not_called()
class TestEditorClipsDurationAndStartTime:
"""测试素材时长获取、clip duration 缩短、start_time 传入。"""
+223 -196
View File
@@ -1,13 +1,17 @@
"""_get_template_segments 回退路径测试.
"""模板片段配置读取路径测试(#1774.
验证三级回退链:
1. 新模板系统(tpl_svc.list_clip_configs)正常 → 直接返回
2. 新模板系统主表不存在(ValueError)→ 直接查 template_clip_configs 表兜底
3. 直接查表也失败 → 回退旧模板系统(template_segments
4. 全部失败 → 返回空列表
收敛后模板读取走单一数据源,不再有"新表抛异常→降级查旧表→再降级查 segments"
的异常控制流:
覆盖 P0 修复:自建模板在 edit_templates 主表不存在但在 template_clip_configs 有记录时,
from-assets 流程不再 400。
- ``EditTemplateService.list_clip_configs_for_editor`` 显式判定模板归属/存在性:
1. 用户自建模板在旧表 ``templates``(归属 user_idis_active=True)→ 直接读
``template_clip_configs``
2. 全局模板在新表 ``edit_templates``(无 user_id)→ 直接读 ``template_clip_configs``
3. 两表都没有 → 抛 ``TemplateNotFoundError``(路由层映射 404)。
- ``_get_template_segments`` 仅做配置→(order, min, max) 的映射与排序,
模板存在但无配置返回空列表(路由层映射 422)。
使用真实 SQLite 内存库 + 真实仓储,验证端到端读路径不抛 ``ValueError: 模板不存在``。
"""
from __future__ import annotations
@@ -15,21 +19,198 @@ from __future__ import annotations
import os
import sys
from pathlib import Path
from unittest.mock import MagicMock, PropertyMock
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")
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
from app.api.routes.templates_editor.clips import _get_template_segments
import pytest # noqa: E402
from sqlalchemy import create_engine # noqa: E402
from sqlalchemy.orm import sessionmaker # noqa: E402
from packages.adapters.sqlalchemy_impl.models import ( # noqa: E402
Base,
EditTemplateModel,
TemplateClipConfigModel,
TemplateModel,
)
TEST_TEMPLATE_ID = "tmpl-orphan-001"
DEFAULT_DUR = 5.0 # _DEFAULT_EDITOR_CLIP_DURATION
USER_ID = "user-001"
OTHER_USER_ID = "user-002"
# ---------------------------------------------------------------------------
# 真实内存 DB fixture
# ---------------------------------------------------------------------------
def _make_session():
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
return sessionmaker(bind=engine)()
def _seed_legacy_template(session, template_id: str, user_id: str, *, active: bool = True, clip_count: int = 3):
"""创建旧表 templates 模板(+ template_clip_configs 片段配置)。"""
session.add(
TemplateModel(
id=template_id,
user_id=user_id,
name=f"模板-{template_id}",
mode="one_take",
is_active=active,
)
)
for order in range(clip_count):
session.add(
TemplateClipConfigModel(
id=f"cc-{template_id}-{order}",
template_id=template_id,
clip_type="main",
order=order,
min_duration=5.0,
max_duration=8.0,
)
)
session.commit()
def _seed_global_template(session, template_id: str, *, status: str = "active", clip_count: int = 2):
"""创建新表 edit_templates 全局模板(+ template_clip_configs 片段配置)。"""
session.add(
EditTemplateModel(
id=template_id,
name=f"全局模板-{template_id}",
template_type="default",
editing_mode="one_take",
status=status,
)
)
for order in range(clip_count):
session.add(
TemplateClipConfigModel(
id=f"gcc-{template_id}-{order}",
template_id=template_id,
clip_type="main",
order=order,
min_duration=3.0,
max_duration=6.0,
)
)
session.commit()
# ---------------------------------------------------------------------------
# Service 读路径测试
# ---------------------------------------------------------------------------
class TestListClipConfigsForEditor:
"""list_clip_configs_for_editor 单一数据源 + 归属/存在性判定。"""
def test_legacy_user_template_returns_configs(self):
"""用户自建模板(templates 表 + 3 条 clip_configs)→ 正常返回,不抛异常。"""
from app.services.edit_template_service import EditTemplateService
session = _make_session()
_seed_legacy_template(session, "tmpl-legacy", USER_ID, clip_count=3)
svc = EditTemplateService(session)
configs = svc.list_clip_configs_for_editor("tmpl-legacy", USER_ID)
assert len(configs) == 3
assert [c.order for c in configs] == [0, 1, 2]
assert all(c.min_duration == 5.0 for c in configs)
def test_missing_template_raises_not_found(self):
"""模板不存在(两表都没有)→ TemplateNotFoundError。"""
from app.services.edit_template_service import EditTemplateService, TemplateNotFoundError
session = _make_session()
svc = EditTemplateService(session)
with pytest.raises(TemplateNotFoundError):
svc.list_clip_configs_for_editor("tmpl-not-exist", USER_ID)
def test_other_users_template_raises_not_found(self):
"""他人模板(user_id 不匹配)→ TemplateNotFoundError(归属校验)。"""
from app.services.edit_template_service import EditTemplateService, TemplateNotFoundError
session = _make_session()
_seed_legacy_template(session, "tmpl-owner", OTHER_USER_ID, clip_count=3)
svc = EditTemplateService(session)
with pytest.raises(TemplateNotFoundError):
svc.list_clip_configs_for_editor("tmpl-owner", USER_ID)
def test_deleted_legacy_template_raises_not_found(self):
"""已软删除(is_active=False)的旧表模板 → TemplateNotFoundError。"""
from app.services.edit_template_service import EditTemplateService, TemplateNotFoundError
session = _make_session()
_seed_legacy_template(session, "tmpl-deleted", USER_ID, active=False, clip_count=3)
svc = EditTemplateService(session)
with pytest.raises(TemplateNotFoundError):
svc.list_clip_configs_for_editor("tmpl-deleted", USER_ID)
def test_legacy_template_without_configs_returns_empty(self):
"""模板存在且归属正确但无片段配置 → 返回空列表(不抛异常,路由层映射 422)。"""
from app.services.edit_template_service import EditTemplateService
session = _make_session()
_seed_legacy_template(session, "tmpl-noconfig", USER_ID, clip_count=0)
svc = EditTemplateService(session)
configs = svc.list_clip_configs_for_editor("tmpl-noconfig", USER_ID)
assert configs == []
def test_global_template_returns_configs(self):
"""新表 edit_templates 全局模板(无 user_id)→ 任意用户可读,正常返回。"""
from app.services.edit_template_service import EditTemplateService
session = _make_session()
_seed_global_template(session, "tmpl-global", clip_count=2)
svc = EditTemplateService(session)
configs = svc.list_clip_configs_for_editor("tmpl-global", USER_ID)
assert len(configs) == 2
assert [c.order for c in configs] == [0, 1]
def test_normal_legacy_request_does_not_raise_valueerror(self):
"""正常旧表模板请求绝不在读路径抛 ValueError: 模板不存在(回归保护)。"""
import logging
from app.services.edit_template_service import EditTemplateService
session = _make_session()
_seed_legacy_template(session, "tmpl-ok", USER_ID, clip_count=3)
svc = EditTemplateService(session)
with pytest.MonkeyPatch.context() as mp:
# 若读路径意外抛 ValueError 并被记录为异常堆栈,测试能感知
errors: list[str] = []
mp.setattr(
logging.getLogger("app.services.edit_template_service"),
"exception",
lambda *a, **k: errors.append(str(a)),
)
configs = svc.list_clip_configs_for_editor("tmpl-ok", USER_ID)
assert len(configs) == 3
assert errors == []
# ---------------------------------------------------------------------------
# _get_template_segments 映射测试
# ---------------------------------------------------------------------------
def _make_clip_config(order: int, min_dur: float = 3.0, max_dur: float = 8.0):
"""构造 mock TemplateClipConfig 领域实体."""
cc = MagicMock()
cc.order = order
cc.min_duration = min_dur
@@ -37,207 +218,53 @@ def _make_clip_config(order: int, min_dur: float = 3.0, max_dur: float = 8.0):
return cc
def _make_old_segment(segment_order: int, dur_min: float = 4.0, dur_max: float = 7.0):
"""构造 mock 旧 TemplateSegment."""
s = MagicMock()
s.segment_order = segment_order
s.duration_min = dur_min
s.duration_max = dur_max
return s
class TestGetTemplateSegments:
"""_get_template_segments 仅做映射/排序,异常与空配置语义明确。"""
def test_maps_and_sorts_configs(self):
from app.api.routes.templates_editor.clips import _get_template_segments
# ---------------------------------------------------------------------------
# 测试
# ---------------------------------------------------------------------------
class TestGetTemplateSegmentsFallback:
"""_get_template_segments 三级回退链."""
def test_new_system_works(self):
"""路径1:新模板系统正常返回 → 直接使用."""
configs = [_make_clip_config(0, 2.0, 6.0), _make_clip_config(1, 3.0, 9.0)]
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.return_value = configs
db = MagicMock()
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
assert len(result) == 2
assert result[0] == (0, 2.0, 6.0)
assert result[1] == (1, 3.0, 9.0)
tpl_svc.list_clip_configs.assert_called_once_with(TEST_TEMPLATE_ID)
def test_main_table_missing_direct_query_succeeds(self):
"""路径2(P0修复):主表不存在 ValueError → 直接查表成功.
模拟自建模板在 edit_templates 主表已删除/不存在,
但 template_clip_configs 表有记录。
"""
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.side_effect = ValueError(f"模板不存在: {TEST_TEMPLATE_ID}")
db = MagicMock()
# Mock SQLAlchemyTemplateClipConfigRepository
direct_configs = [
_make_clip_config(0, 2.0, 5.0),
_make_clip_config(1, 3.0, 7.0),
_make_clip_config(2, 4.0, 8.0),
]
with (
__import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
) as mock_repo_cls,
):
mock_repo = MagicMock()
mock_repo.list_by_template.return_value = direct_configs
mock_repo_cls.return_value = mock_repo
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
assert len(result) == 3
assert result[0] == (0, 2.0, 5.0)
assert result[1] == (1, 3.0, 7.0)
assert result[2] == (2, 4.0, 8.0)
mock_repo.list_by_template.assert_called_once_with(TEST_TEMPLATE_ID)
def test_main_table_missing_direct_query_empty_falls_to_old(self):
"""路径2→3:主表不存在 + 直接查表为空 → 回退旧系统."""
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.side_effect = ValueError("模板不存在")
db = MagicMock()
old_segments = [_make_old_segment(0, 3.0, 6.0)]
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
) as mock_repo_cls:
mock_repo = MagicMock()
mock_repo.list_by_template.return_value = [] # 新表也没记录
mock_repo_cls.return_value = mock_repo
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository"
) as mock_old_cls:
mock_old = MagicMock()
mock_old.list_segments.return_value = old_segments
mock_old_cls.return_value = mock_old
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
assert len(result) == 1
assert result[0] == (0, 3.0, 6.0)
def test_all_fail_returns_empty(self):
"""路径4:三级全部失败 → 返回空列表."""
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.side_effect = ValueError("模板不存在")
db = MagicMock()
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
) as mock_repo_cls:
mock_repo = MagicMock()
mock_repo.list_by_template.side_effect = Exception("DB error")
mock_repo_cls.return_value = mock_repo
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository"
) as mock_old_cls:
mock_old = MagicMock()
mock_old.list_segments.return_value = [] # 旧表也空
mock_old_cls.return_value = mock_old
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
assert result == []
def test_direct_query_sorts_by_order(self):
"""直接查表返回的结果按 order 排序."""
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.side_effect = ValueError("模板不存在")
db = MagicMock()
# 故意乱序
configs = [
tpl_svc.list_clip_configs_for_editor.return_value = [
_make_clip_config(2, 5.0, 10.0),
_make_clip_config(0, 2.0, 4.0),
_make_clip_config(1, 3.0, 6.0),
]
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
) as mock_repo_cls:
mock_repo = MagicMock()
mock_repo.list_by_template.return_value = configs
mock_repo_cls.return_value = mock_repo
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
result = _get_template_segments("tmpl-1", USER_ID, tpl_svc)
assert [r[0] for r in result] == [0, 1, 2]
assert result[0] == (0, 2.0, 4.0)
assert result[1] == (1, 3.0, 6.0)
assert result[2] == (2, 5.0, 10.0)
tpl_svc.list_clip_configs_for_editor.assert_called_once_with("tmpl-1", USER_ID)
def test_empty_configs_returns_empty(self):
from app.api.routes.templates_editor.clips import _get_template_segments
def test_direct_query_handles_none_durations(self):
"""直接查表时 min/max_duration 为 None → 使用默认值."""
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.side_effect = ValueError("模板不存在")
db = MagicMock()
tpl_svc.list_clip_configs_for_editor.return_value = []
assert _get_template_segments("tmpl-1", USER_ID, tpl_svc) == []
def test_missing_template_propagates_not_found(self):
from app.api.routes.templates_editor.clips import _get_template_segments
from app.services.edit_template_service import TemplateNotFoundError
tpl_svc = MagicMock()
tpl_svc.list_clip_configs_for_editor.side_effect = TemplateNotFoundError("tmpl-x")
with pytest.raises(TemplateNotFoundError):
_get_template_segments("tmpl-x", USER_ID, tpl_svc)
def test_none_durations_use_default(self):
from app.api.routes.templates_editor.clips import _get_template_segments
tpl_svc = MagicMock()
cc = MagicMock()
cc.order = 0
cc.min_duration = None
cc.max_duration = None
tpl_svc.list_clip_configs_for_editor.return_value = [cc]
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
) as mock_repo_cls:
mock_repo = MagicMock()
mock_repo.list_by_template.return_value = [cc]
mock_repo_cls.return_value = mock_repo
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
assert len(result) == 1
# None → default (5.0), max(None or None) → default (5.0)
assert result[0] == (0, DEFAULT_DUR, DEFAULT_DUR)
def test_new_system_returns_empty_tries_direct(self):
"""新模板系统返回空列表(非异常)→ 继续尝试直接查表."""
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.return_value = [] # 空列表,非异常
db = MagicMock()
direct_configs = [_make_clip_config(0, 3.0, 6.0)]
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
) as mock_repo_cls:
mock_repo = MagicMock()
mock_repo.list_by_template.return_value = direct_configs
mock_repo_cls.return_value = mock_repo
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
# 新系统返回空 → 不走 except → 但也没 return → 继续往下走
# 直接查表有数据 → 返回
assert len(result) == 1
assert result[0] == (0, 3.0, 6.0)
def test_existing_template_unaffected(self):
"""正常模板(主表存在)行为不变."""
configs = [_make_clip_config(0, 2.0, 5.0)]
tpl_svc = MagicMock()
tpl_svc.list_clip_configs.return_value = configs
db = MagicMock()
with __import__("unittest.mock", fromlist=["patch"]).patch(
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
) as mock_repo_cls:
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
# 直接查表不应被调用(新系统已返回)
mock_repo_cls.assert_not_called()
assert len(result) == 1
assert result[0] == (0, 2.0, 5.0)
result = _get_template_segments("tmpl-1", USER_ID, tpl_svc)
assert result == [(0, DEFAULT_DUR, DEFAULT_DUR)]
+60 -70
View File
@@ -226,10 +226,10 @@ class TestGetMediakitRecommendations:
class TestGetTemplateSegments:
"""测试模板片段配置查询。"""
"""测试模板片段配置查询(单一数据源:template_clip_configs"""
def test_returns_segments_from_new_template_system(self):
"""新模板系统clip_configs)有数据时优先使用"""
def test_returns_segments_from_clip_configs(self):
"""片段配置主表clip_configs)有数据时按 order 排序返回"""
from app.api.routes.templates_editor.clips import _get_template_segments
mock_tpl_svc = MagicMock()
@@ -241,66 +241,34 @@ class TestGetTemplateSegments:
cc2.order = 1
cc2.min_duration = 4.0
cc2.max_duration = 8.0
mock_tpl_svc.list_clip_configs.return_value = [cc2, cc1] # 乱序返回
mock_tpl_svc.list_clip_configs_for_editor.return_value = [cc2, cc1] # 乱序返回
result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
result = _get_template_segments("tmpl-1", "user-1", mock_tpl_svc)
assert len(result) == 2
assert result[0] == (0, 3.0, 5.0)
assert result[1] == (1, 4.0, 8.0)
mock_tpl_svc.list_clip_configs_for_editor.assert_called_once_with("tmpl-1", "user-1")
def test_falls_back_to_old_template_segments(self):
"""模板系统无数据时回退到旧系统"""
def test_returns_empty_when_no_configs(self):
"""模板存在但没有片段配置时返回空列表(路由层据此返回 422)"""
from app.api.routes.templates_editor.clips import _get_template_segments
mock_tpl_svc = MagicMock()
mock_tpl_svc.list_clip_configs.return_value = []
mock_tpl_svc.list_clip_configs_for_editor.return_value = []
with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
mock_repo = MagicMock()
seg1 = MagicMock()
seg1.segment_order = 0
seg1.duration_min = 2.0
seg1.duration_max = 4.0
mock_repo.list_segments.return_value = [seg1]
MockRepo.return_value = mock_repo
result = _get_template_segments("tmpl-1", "user-1", mock_tpl_svc)
assert result == []
result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
assert len(result) == 1
assert result[0] == (0, 2.0, 4.0)
def test_returns_empty_when_no_segments(self):
"""两套系统都没有片段配置时返回空列表。"""
def test_missing_template_raises(self):
"""模板不存在/无权限时服务层抛 TemplateNotFoundError(路由层据此返回 404)。"""
from app.api.routes.templates_editor.clips import _get_template_segments
from app.services.edit_template_service import TemplateNotFoundError
mock_tpl_svc = MagicMock()
mock_tpl_svc.list_clip_configs.return_value = []
mock_tpl_svc.list_clip_configs_for_editor.side_effect = TemplateNotFoundError("tmpl-x")
with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
mock_repo = MagicMock()
mock_repo.list_segments.return_value = []
MockRepo.return_value = mock_repo
result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
assert result == []
def test_new_system_exception_falls_back(self):
"""新模板系统异常时回退到旧系统。"""
from app.api.routes.templates_editor.clips import _get_template_segments
mock_tpl_svc = MagicMock()
mock_tpl_svc.list_clip_configs.side_effect = RuntimeError("db error")
with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
mock_repo = MagicMock()
seg = MagicMock()
seg.segment_order = 0
seg.duration_min = 1.0
seg.duration_max = 3.0
mock_repo.list_segments.return_value = [seg]
MockRepo.return_value = mock_repo
result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
assert len(result) == 1
with pytest.raises(TemplateNotFoundError):
_get_template_segments("tmpl-x", "user-1", mock_tpl_svc)
# ── from-assets 端点集成测试 ────────────────────────────────────────────────
@@ -358,7 +326,7 @@ def _make_tpl_svc_with_segments(segments):
"""segments: list of (order, min_dur, max_dur)"""
svc = MagicMock()
clip_configs = [_make_clip_config(o, mn, mx) for o, mn, mx in segments]
svc.list_clip_configs.return_value = clip_configs
svc.list_clip_configs_for_editor.return_value = clip_configs
return svc
@@ -540,34 +508,56 @@ class TestFromAssetsByTemplateSegments:
orders = [c["order"] for c in clips_data]
assert orders == [0, 1, 2]
def test_no_segments_raises_400(self):
"""模板没有 segment 配置时返回 400"""
def test_no_segments_raises_422(self):
"""模板存在但未配置片段时返回 422(与模板不存在的 404 区分)"""
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
from fastapi import HTTPException
mock_tpl_svc = MagicMock()
mock_tpl_svc.list_clip_configs.return_value = []
mock_tpl_svc.list_clip_configs_for_editor.return_value = []
mock_plan_svc = _make_plan_svc()
with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
mock_repo = MagicMock()
mock_repo.list_segments.return_value = []
MockRepo.return_value = mock_repo
body = ClipsFromAssetsRequest(asset_ids=["a1"])
with pytest.raises(HTTPException) as exc_info:
create_clips_from_assets_editor(
template_id="tmpl-1",
body=body,
background_tasks=MagicMock(),
plan_id="plan-1",
services=(mock_tpl_svc, mock_plan_svc),
asset_repo=MagicMock(),
db=MagicMock(),
current_user=_make_auth_user(),
)
assert exc_info.value.status_code == 422
body = ClipsFromAssetsRequest(asset_ids=["a1"])
with pytest.raises(HTTPException) as exc_info:
create_clips_from_assets_editor(
template_id="tmpl-1",
body=body,
background_tasks=MagicMock(),
plan_id="plan-1",
services=(mock_tpl_svc, mock_plan_svc),
asset_repo=MagicMock(),
db=MagicMock(),
current_user=_make_auth_user(),
)
assert exc_info.value.status_code == 400
mock_plan_svc.replace_all_clips_transactional.assert_not_called()
def test_template_not_found_raises_404(self):
"""模板不存在/已删除/无权限时返回 404。"""
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
from app.services.edit_template_service import TemplateNotFoundError
from fastapi import HTTPException
mock_tpl_svc = MagicMock()
mock_tpl_svc.list_clip_configs_for_editor.side_effect = TemplateNotFoundError("tmpl-x")
mock_plan_svc = _make_plan_svc()
body = ClipsFromAssetsRequest(asset_ids=["a1"])
with pytest.raises(HTTPException) as exc_info:
create_clips_from_assets_editor(
template_id="tmpl-x",
body=body,
background_tasks=MagicMock(),
plan_id="plan-1",
services=(mock_tpl_svc, mock_plan_svc),
asset_repo=MagicMock(),
db=MagicMock(),
current_user=_make_auth_user(),
)
assert exc_info.value.status_code == 404
mock_plan_svc.replace_all_clips_transactional.assert_not_called()