fix(p0): #1922 from-assets 空 templateId 双斜杠 404 兜底 (#1926)
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (push) Successful in 2s
CI/CD Pipeline / Check push changed paths (push) Successful in 16s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m0s
CI/CD Pipeline / Frontend Unit Tests (push) Failing after 2m42s
CI/CD Pipeline / Integration Tests (push) Successful in 3m50s
CI/CD Pipeline / Build Staging API Image (push) Successful in 3m35s
CI/CD Pipeline / Validate - Style (push) Successful in 4m42s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 4m29s
CI/CD Pipeline / Validate - Python (mypy + alembic) (push) Successful in 5m0s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 43s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 1m47s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 3m5s
CI/CD Pipeline / Staging E2E Tests (push) Failing after 3m10s
CI/CD Pipeline / Validate - Security (push) Successful in 11m12s
CI/CD Pipeline / Unit Tests (push) Successful in 11m18s
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / CI Gate (push) Failing after 27h18m15s
CI/CD Pipeline / Build Production API Image (push) Failing after 27h18m15s
CI/CD Pipeline / Frontend Lint (push) Failing after 27h29m35s
CI/CD Pipeline / Retag skipped Staging API Image (push) Failing after 27h24m8s
CI/CD Pipeline / PR Build Worker Image (push) Failing after 27h28m58s
CI/CD Pipeline / PR Build API Image (push) Failing after 27h29m0s
CI/CD Pipeline / Deploy Production (push) Failing after 27h17m36s
CI/CD Pipeline / Build Production Worker Image (push) Failing after 27h17m38s
CI/CD Pipeline / Build Production Web Image (push) Failing after 27h17m38s
CI/CD Pipeline / Canary Release to Production (push) Failing after 27h17m36s
CI/CD Pipeline / Retag skipped Staging Worker Image (push) Failing after 27h24m7s
CI/CD Pipeline / Retag skipped Staging Web Image (push) Failing after 27h24m8s
CI/CD Pipeline / PR Build Web Image (push) Failing after 27h28m58s
CI/CD Pipeline / Check if frontend-only change (push) Failing after 27h29m0s

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
This commit was merged in pull request #1926.
This commit is contained in:
2026-09-15 11:54:06 +08:00
committed by auto-approve-bot
parent a25aa7baed
commit 352221f199
7 changed files with 636 additions and 26 deletions
+5
View File
@@ -6,6 +6,7 @@ from app.api.routes.assets import router as assets_router
from app.api.routes.auth import router as auth_router
from app.api.routes.chunked_upload import router as chunked_upload_router
from app.api.routes.classification_jobs import router as classification_jobs_router
from app.api.routes.clips_standalone import router as clips_standalone_router
from app.api.routes.cover_templates import router as cover_templates_router
from app.api.routes.duplication import router as duplication_router
from app.api.routes.feature_flags import router as feature_flags_router
@@ -157,6 +158,10 @@ api_router.include_router(
prefix="/templates",
tags=["Template"],
)
api_router.include_router(
clips_standalone_router,
tags=["Clips"],
)
api_router.include_router(
templates_editor_router,
prefix="/templates/{template_id}/editor",
@@ -0,0 +1,91 @@
"""默认模板兜底共享逻辑(P0 #1922).
提供 get_or_create_default_template_id(db, user_id) 共享函数,
供 templates.py 列表查询、clips_standalone.py 独立端点、dependencies.py
resolve_draft_plan_id 三处复用,避免三处各写一套兜底逻辑产生分叉。
根因:PR#1918 清理模板管理 API 时误删了 GET /templates 自动创建默认模板
兜底,前端 PR#1913 去掉空 tid 拦截后首次进入生成页拼出
/templates//editor/clips/from-assets(双斜杠)→ FastAPI 404,阻断新用户首次
生成。
"""
from __future__ import annotations
import logging
from typing import Optional
from sqlalchemy.orm import Session
logger = logging.getLogger(__name__)
def get_or_create_default_template_id(db: Session, user_id: str) -> Optional[str]:
"""获取或自动创建默认配音模板的 id。
判定逻辑(不做异常降级,只有确实创建失败时才回滚重查):
1. 查用户名下 is_active=True 且有 TemplateClipConfig 的模板 → 返回其 id;
2. 无则调用 CreateTemplateUseCase 创建一条默认 voice_over 模板;
3. 创建异常时 rollback 再重查一次(防并发唯一键冲突),重查仍无返回 None。
"""
from packages.adapters.sqlalchemy_impl.models import (
TemplateClipConfigModel,
TemplateModel,
)
from packages.adapters.sqlalchemy_impl.template_repository import (
SQLAlchemyTemplateRepository,
)
from packages.application.template.commands import (
CreateTemplateCommand,
SegmentCommand,
)
from packages.application.template.use_cases import CreateTemplateUseCase
existing = (
db.query(TemplateModel)
.filter(TemplateModel.user_id == user_id, TemplateModel.is_active.is_(True))
.order_by(TemplateModel.created_at.asc())
.first()
)
if existing is not None:
has_seg = (
db.query(TemplateClipConfigModel.id).filter(TemplateClipConfigModel.template_id == existing.id).first()
)
if has_seg:
return existing.id
try:
repo = SQLAlchemyTemplateRepository(db)
cmd = CreateTemplateCommand(
user_id=user_id,
name="默认配音模板",
mode="voice_over",
category="default",
tags=[],
title_config={},
subtitle_config={},
bgm_config={},
estimated_duration=0.0,
segments=[SegmentCommand(segment_order=0, duration_min=1.0, duration_max=30.0)],
)
tpl = CreateTemplateUseCase(repo).execute(cmd)
db.commit()
logger.info("auto-created default voice_over template: id=%s user=%s", tpl.id, user_id)
return tpl.id
except Exception:
db.rollback()
# 重查:可能并发请求已建好
existing = (
db.query(TemplateModel)
.filter(TemplateModel.user_id == user_id, TemplateModel.is_active.is_(True))
.order_by(TemplateModel.created_at.asc())
.first()
)
if existing is not None:
has_seg = (
db.query(TemplateClipConfigModel.id).filter(TemplateClipConfigModel.template_id == existing.id).first()
)
if has_seg:
return existing.id
logger.exception("failed to auto-create default template user=%s", user_id)
return None
@@ -0,0 +1,90 @@
"""独立的从素材创建片段端点(不依赖 template_id 路径参数).
POST /api/v1/clips/from-assets
- 与 /api/v1/templates/{template_id}/editor/clips/from-assets 功能一致
- 区别:template_id 从 body 传入(可选),为空时后端自动创建/查找默认模板
- 解决前端首次加载时 templateId 为空导致双斜杠 404 的问题(P0 #1922)
- 内部复用 resolve_draft_plan_id 和 create_clips_from_assets_editor 的核心逻辑
"""
from __future__ import annotations
import logging
from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import get_asset_repository, get_db_session
from app.services.edit_plan_service import EditPlanService
from app.services.edit_template_service import EditTemplateService
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, status
from sqlalchemy.orm import Session
from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
from ._default_template import get_or_create_default_template_id
from .templates_editor.clips import create_clips_from_assets_editor
from .templates_editor.dependencies import resolve_draft_plan_id
from .templates_editor.schemas import ClipsFromAssetsRequest, ClipsFromAssetsResponse
logger = logging.getLogger(__name__)
router = APIRouter(tags=["Clips"])
class StandaloneClipsRequest(ClipsFromAssetsRequest):
"""扩展请求:template_id 可选(不传则后端自动兜底默认模板)。"""
template_id: str | None = None
def _get_editor_services_direct(db: Session) -> tuple[EditTemplateService, EditPlanService]:
"""直接构造服务实例(非 Depends 版本,供独立端点内部调用)。"""
return EditTemplateService(db), EditPlanService(db)
@router.post("/clips/from-assets", response_model=ClipsFromAssetsResponse)
def create_clips_from_assets(
body: StandaloneClipsRequest,
background_tasks: BackgroundTasks,
db: Session = Depends(get_db_session),
current_user: AuthenticatedUser = Depends(get_current_user),
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
) -> ClipsFromAssetsResponse:
"""从素材批量创建片段(template_id 可选,为空自动兜底)。"""
user_id = str(current_user.user.id)
services = _get_editor_services_direct(db)
# 1. 解析/兜底 template_id,拿到 plan_id
template_id = (body.template_id or "").strip()
if not template_id:
template_id = get_or_create_default_template_id(db, user_id)
if not template_id:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="无法自动创建默认模板,请刷新页面重试",
)
plan_id = resolve_draft_plan_id(
template_id=template_id,
services=services,
current_user=current_user,
db=db,
auto_create_default=False, # 上面已兜底过
)
# 2. 构造标准化请求(去除独立端扩展字段),复用原端点核心逻辑
core_body = ClipsFromAssetsRequest(
asset_ids=body.asset_ids,
clip_type=body.clip_type,
clip_count=body.clip_count,
required_clips_count=body.required_clips_count,
)
# 3. 直接调用原端点函数(此时所有 Depends 依赖已手动传入)
return create_clips_from_assets_editor(
template_id=template_id,
body=core_body,
background_tasks=background_tasks,
plan_id=plan_id,
services=services,
asset_repo=asset_repo,
db=db,
current_user=current_user,
)
+11 -4
View File
@@ -2,10 +2,11 @@
保留:
- GET /templates:列表查询(生成页使用)
- 默认模板自动创建兜底逻辑(``_get_or_create_default_template_id`` 位于
generation_variant_plans.py)继续通过 service 层 CreateTemplateUseCase 工作,
但不再暴露模板 CRUD / 分类 / 标签 / 收藏 / 复制 / 校验 / 使用统计等 HTTP 端点
(前端 PR#1911 已删除 my-templates / editing-planner / templates 管理页面)。
- 默认模板自动创建兜底逻辑(复用 _default_template.get_or_create_default_template_id)
其他模板 CRUD / 分类 / 标签 / 收藏 / 复制 / 校验 / 使用统计等 HTTP 端点
已在 PR#1918 中删除(前端 PR#1911 已删除 my-templates / editing-planner /
templates 管理页面)。
"""
from __future__ import annotations
@@ -27,6 +28,8 @@ from packages.adapters.sqlalchemy_impl.template_repository import SQLAlchemyTemp
from packages.application.template.commands import ListTemplatesFilter
from packages.application.template.use_cases import CountTemplatesUseCase, ListTemplatesUseCase
from ._default_template import get_or_create_default_template_id
router = APIRouter()
@@ -43,9 +46,13 @@ def list_templates(
page_size: int = Query(20, ge=1, le=100, description="每页条数,默认 20"),
current_user: AuthenticatedUser = Depends(get_current_user),
repo: SQLAlchemyTemplateRepository = Depends(_get_template_repository),
db: Session = Depends(get_db_session),
):
"""获取用户可用的模板列表(仅返回 active 状态)。"""
user_id = str(current_user.user.id)
# P0 兜底:无有效模板时自动创建默认配音模板(解决新用户首次进入生成页 404)
get_or_create_default_template_id(db, user_id)
list_uc = ListTemplatesUseCase(repo)
count_uc = CountTemplatesUseCase(repo)
filters = ListTemplatesFilter(
@@ -28,7 +28,7 @@ from .adjustments import router as adjustments_router
from .ai_features import router as ai_features_router
from .bgm import router as bgm_router
from .clips import router as clips_router
from .dependencies import get_draft_plan_id, get_editor_services # noqa: F401
from .dependencies import get_draft_plan_id, get_editor_services, resolve_draft_plan_id # noqa: F401
from .draft import router as draft_router
from .effects import router as effects_router
from .export import router as export_router
@@ -2,13 +2,16 @@
核心依赖:
- get_editor_services: 获取模板+计划服务
- get_draft_plan_id: 根据 template_id 获取或创建草稿,返回 plan_id
- get_draft_plan_id: Depends 形式的路径依赖(template_id 路径参数必填)
- resolve_draft_plan_id: 纯函数版本,供 clips_standalone 等非路径参数场景复用
(支持空 tid 时自动兜底创建默认模板)
"""
from __future__ import annotations
import logging
from app.api.routes._default_template import get_or_create_default_template_id
from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import get_db_session
from app.services.edit_plan_service import EditPlanService
@@ -30,50 +33,61 @@ def get_editor_services(
return EditTemplateService(db), EditPlanService(db)
def get_draft_plan_id(
def resolve_draft_plan_id(
template_id: str,
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
current_user: AuthenticatedUser = Depends(get_current_user),
db: Session = Depends(get_db_session),
services: tuple[EditTemplateService, EditPlanService],
current_user: AuthenticatedUser,
db: Session,
auto_create_default: bool = True,
) -> str:
"""路径依赖:根据 template_id 获取或创建草稿,返回 plan_id.
"""根据 template_id 获取或创建草稿,返回 plan_id(纯函数,不带 Depends)。
这是模板编辑器路由的核心依赖——所有编辑器端点都先经过这里,
确保 template_id → plan_id 的映射始终存在。
模板读取遵循单一数据源、显式判定(不使用异常降级):
- 用户自建模板在旧表 ``templates``(归属 user_id,is_active=True);
- 全局模板在新表 ``edit_templates``(无 user_id,全局可读)。
模板不存在、已删除或不归属于当前用户时,一律返回 404。
当 auto_create_default=True 且 template_id 为空时,自动调用
get_or_create_default_template_id 创建默认模板(用于 clips_standalone
等非路径参数场景)。
"""
tpl_svc, plan_svc = services
user_id = str(current_user.user.id)
# 0. 门禁:校验模板存在且可访问(即使草稿已缓存命中也要校验,
# 避免模板被删除/无权访问后仍可通过既有草稿 plan 继续操作)。
# 0. 空 tid 兜底
if not template_id:
if auto_create_default:
tid = get_or_create_default_template_id(db, user_id)
if not tid:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="无法自动创建默认模板,请刷新页面重试",
)
template_id = tid
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="template_id 不能为空",
)
# 1. 门禁:校验模板存在且可访问
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. 草稿已存在 → 直接返回
# 2. 草稿已存在 → 直接返回
draft = tpl_svc.get_template_draft(template_id)
if draft is not None:
return draft.id
# 2. 全局模板(新系统)→ 用新服务创建草稿
# 3. 全局模板(新系统)→ 用新服务创建草稿
if is_global_template:
draft = tpl_svc.create_template_draft(template_id, user_id=user_id)
return draft.id
# 3. 旧模板(templates 表)→ 基于旧模板创建草稿计划
# 4. 旧模板(templates 表)→ 基于旧模板创建草稿计划
from app.services.plan_generator_service import PlanGeneratorService
from packages.domain.edit_template import EditTemplate, EditTemplateStatus
from packages.domain.template_clip_config import ClipType, TemplateClipConfig
# 构造伪 EditTemplate 对象(只填 generate_from_template 需要的字段)
pseudo_template = EditTemplate(
id=old_template.id,
name=old_template.name,
@@ -81,7 +95,6 @@ def get_draft_plan_id(
status=EditTemplateStatus.ACTIVE,
)
# 将旧模板 segments 转换为 clip_configs
clip_configs: list[TemplateClipConfig] = []
for seg in old_template.segments or []:
clip_configs.append(
@@ -105,7 +118,6 @@ def get_draft_plan_id(
)
plan = result["plan"]
# 标记为模板草稿(后续可复用 tpl_svc.get_template_draft 的查找逻辑)
plan_svc.update_plan_config(plan.id, {"is_template_draft": True})
logger.info(
@@ -115,3 +127,23 @@ def get_draft_plan_id(
user_id,
)
return plan.id
def get_draft_plan_id(
template_id: str,
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
current_user: AuthenticatedUser = Depends(get_current_user),
db: Session = Depends(get_db_session),
) -> str:
"""路径依赖:根据 template_id 获取或创建草稿,返回 plan_id.
Depends 版本:路径参数 template_id 由 FastAPI 保证非空,不自动兜底。
兜底逻辑走 resolve_draft_plan_id(auto_create_default=False)。
"""
return resolve_draft_plan_id(
template_id=template_id,
services=services,
current_user=current_user,
db=db,
auto_create_default=False,
)
@@ -0,0 +1,385 @@
"""P0 #1922 默认模板自动兜底单元测试。
覆盖 get_or_create_default_template_id / resolve_draft_plan_id / list_templates / clips_standalone。
注意:_default_template.py 内部使用函数级 lazy import,测试创建/异常分支时
通过 patch.dict(sys.modules, ...) 临时替换 lazy import 目标为 mock 模块。
"""
from __future__ import annotations
import sys
import types
from datetime import datetime
from unittest.mock import MagicMock, patch
import pytest
from fastapi import BackgroundTasks, HTTPException
# ---------- helpers ----------
def _build_mock_tpl(tid="tpl-exists"):
tpl = MagicMock()
tpl.id = tid
tpl.name = "默认配音模板"
tpl.mode = "voice_over"
tpl.category = "default"
tpl.tags = []
tpl.is_active = True
tpl.created_at = datetime(2026, 1, 1)
tpl.updated_at = datetime(2026, 1, 1)
seg = MagicMock(id="s1", segment_order=0, duration_min=1.0, duration_max=30.0)
tpl.segments = [seg]
return tpl
def _db_chain_existing(tpl):
q_tpl = MagicMock()
q_tpl.filter.return_value.order_by.return_value.first.return_value = tpl
q_seg = MagicMock()
q_seg.filter.return_value.first.return_value = (1,)
db = MagicMock()
db.query.side_effect = [q_tpl, q_seg]
return db
def _db_chain_empty():
q = MagicMock()
q.filter.return_value.order_by.return_value.first.return_value = None
db = MagicMock()
db.query.return_value = q
return db
def _lazy_create_patches(new_tpl_or_exc):
m_models = types.ModuleType("mock_models")
m_models.TemplateModel = MagicMock()
m_models.TemplateClipConfigModel = MagicMock()
m_repo_mod = types.ModuleType("mock_repo")
m_repo_mod.SQLAlchemyTemplateRepository = MagicMock()
m_cmds = types.ModuleType("mock_cmds")
m_cmds.CreateTemplateCommand = MagicMock()
m_cmds.SegmentCommand = MagicMock()
m_uc = types.ModuleType("mock_uc")
uc_instance = MagicMock()
if isinstance(new_tpl_or_exc, BaseException):
uc_instance.execute.side_effect = new_tpl_or_exc
else:
uc_instance.execute.return_value = new_tpl_or_exc
m_uc.CreateTemplateUseCase = MagicMock(return_value=uc_instance)
patches = patch.dict(
sys.modules,
{
"packages.adapters.sqlalchemy_impl.models": m_models,
"packages.adapters.sqlalchemy_impl.template_repository": m_repo_mod,
"packages.application.template.commands": m_cmds,
"packages.application.template.use_cases": m_uc,
},
)
return patches, uc_instance
# ---------- get_or_create_default_template_id ----------
class TestGetOrCreateDefaultTemplateId:
def test_existing_valid_template_returned_directly(self):
from app.api.routes._default_template import get_or_create_default_template_id
tpl = _build_mock_tpl("tpl-1")
db = _db_chain_existing(tpl)
tid = get_or_create_default_template_id(db, "user-1")
assert tid == "tpl-1"
db.commit.assert_not_called()
def test_no_template_creates_default(self):
from app.api.routes._default_template import get_or_create_default_template_id
new_tpl = _build_mock_tpl("tpl-new")
db = _db_chain_empty()
patches, uc = _lazy_create_patches(new_tpl)
with patches:
tid = get_or_create_default_template_id(db, "user-new")
assert tid == "tpl-new"
db.commit.assert_called_once()
uc.execute.assert_called_once()
def test_create_exception_fallback_finds_existing(self):
from app.api.routes._default_template import get_or_create_default_template_id
existing = _build_mock_tpl("tpl-race")
q_none = MagicMock()
q_none.filter.return_value.order_by.return_value.first.return_value = None
q_hit = MagicMock()
q_hit.filter.return_value.order_by.return_value.first.return_value = existing
q_seg = MagicMock()
q_seg.filter.return_value.first.return_value = (1,)
db = MagicMock()
db.query.side_effect = [q_none, q_hit, q_seg]
patches, _ = _lazy_create_patches(Exception("unique violation"))
with patches:
tid = get_or_create_default_template_id(db, "user-race")
assert tid == "tpl-race"
db.rollback.assert_called_once()
def test_create_exception_no_fallback_returns_none(self):
from app.api.routes._default_template import get_or_create_default_template_id
db = _db_chain_empty()
patches, _ = _lazy_create_patches(Exception("DB down"))
with patches:
tid = get_or_create_default_template_id(db, "user-fail")
assert tid is None
db.rollback.assert_called_once()
def test_existing_template_without_segments_triggers_create(self):
from app.api.routes._default_template import get_or_create_default_template_id
tpl_empty = _build_mock_tpl("tpl-empty")
tpl_empty.segments = []
q_tpl = MagicMock()
q_tpl.filter.return_value.order_by.return_value.first.return_value = tpl_empty
q_seg_none = MagicMock()
q_seg_none.filter.return_value.first.return_value = None
new_tpl = _build_mock_tpl("tpl-new")
db = MagicMock()
db.query.side_effect = [q_tpl, q_seg_none]
patches, _ = _lazy_create_patches(new_tpl)
with patches:
tid = get_or_create_default_template_id(db, "user-2")
assert tid == "tpl-new"
db.commit.assert_called_once()
# ---------- resolve_draft_plan_id ----------
class TestResolveDraftPlanId:
def _services(self, global_tpl=None, draft=None):
tpl_svc = MagicMock()
plan_svc = MagicMock()
tpl_svc.get_template.return_value = global_tpl
tpl_svc.get_template_draft.return_value = draft
return (tpl_svc, plan_svc)
def _user(self, uid="user-1"):
u = MagicMock()
u.user.id = uid
return u
def test_empty_tid_auto_fallback(self):
db = MagicMock()
services = self._services()
user = self._user()
old_tpl = _build_mock_tpl("tpl-auto")
mock_repo = MagicMock()
mock_repo.get_active.return_value = old_tpl
mock_plan = MagicMock()
mock_plan.id = "plan-new"
mock_gen = MagicMock()
mock_gen.generate_from_template.return_value = {"plan": mock_plan}
with (
patch(
"app.api.routes.templates_editor.dependencies.SQLAlchemyTemplateRepository",
return_value=mock_repo,
),
patch(
"app.api.routes.templates_editor.dependencies.get_or_create_default_template_id",
return_value="tpl-auto",
) as mock_fb,
patch("app.services.plan_generator_service.PlanGeneratorService", return_value=mock_gen),
):
from app.api.routes.templates_editor.dependencies import resolve_draft_plan_id
plan_id = resolve_draft_plan_id("", services, user, db, auto_create_default=True)
assert plan_id == "plan-new"
mock_fb.assert_called_once_with(db, "user-1")
def test_empty_tid_no_auto_raises_400(self):
from app.api.routes.templates_editor.dependencies import resolve_draft_plan_id
with pytest.raises(HTTPException) as e:
resolve_draft_plan_id("", self._services(), self._user(), MagicMock(), auto_create_default=False)
assert e.value.status_code == 400
def test_missing_template_raises_404(self):
from app.api.routes.templates_editor.dependencies import resolve_draft_plan_id
db = MagicMock()
mock_repo = MagicMock()
mock_repo.get_active.return_value = None
with patch(
"app.api.routes.templates_editor.dependencies.SQLAlchemyTemplateRepository",
return_value=mock_repo,
):
with pytest.raises(HTTPException) as e:
resolve_draft_plan_id("tpl-x", self._services(), self._user(), db, auto_create_default=False)
assert e.value.status_code == 404
def test_existing_draft_returned_directly(self):
from app.api.routes.templates_editor.dependencies import resolve_draft_plan_id
db = MagicMock()
draft = MagicMock()
draft.id = "plan-exist"
services = self._services(draft=draft)
old_tpl = _build_mock_tpl("tpl-1")
mock_repo = MagicMock()
mock_repo.get_active.return_value = old_tpl
with patch(
"app.api.routes.templates_editor.dependencies.SQLAlchemyTemplateRepository",
return_value=mock_repo,
):
plan_id = resolve_draft_plan_id("tpl-1", services, self._user(), db, auto_create_default=False)
assert plan_id == "plan-exist"
def test_global_template_creates_draft(self):
from app.api.routes.templates_editor.dependencies import resolve_draft_plan_id
db = MagicMock()
services = self._services(global_tpl=MagicMock(), draft=None)
new_draft = MagicMock()
new_draft.id = "plan-g"
services[0].create_template_draft.return_value = new_draft
plan_id = resolve_draft_plan_id("tpl-g", services, self._user(), db, auto_create_default=False)
assert plan_id == "plan-g"
services[0].create_template_draft.assert_called_once_with("tpl-g", user_id="user-1")
# ---------- list_templates 路由 ----------
class TestTemplatesListRoute:
def _user(self):
u = MagicMock()
u.user.id = "user-1"
return u
def test_list_templates_calls_fallback_before_query(self):
from app.api.routes.templates import list_templates
db = MagicMock()
repo = MagicMock()
tpl = _build_mock_tpl("tpl-1")
fake_list_uc = MagicMock()
fake_list_uc.execute.return_value = [tpl]
fake_count_uc = MagicMock()
fake_count_uc.execute.return_value = 1
fake_resp = MagicMock(items=[MagicMock(id="tpl-1")], total=1)
with (
patch("app.api.routes.templates.get_or_create_default_template_id") as mock_fb,
patch("app.api.routes.templates.ListTemplatesUseCase", return_value=fake_list_uc),
patch("app.api.routes.templates.CountTemplatesUseCase", return_value=fake_count_uc),
patch(
"app.api.routes.templates.TemplateResponse.model_validate",
return_value=MagicMock(id="tpl-1"),
),
patch("app.api.routes.templates.ListTemplatesResponse", return_value=fake_resp),
):
resp = list_templates(
mode=None,
category=None,
tag=None,
page=1,
page_size=20,
current_user=self._user(),
repo=repo,
db=db,
)
mock_fb.assert_called_once_with(db, "user-1")
assert resp.total == 1
fake_list_uc.execute.assert_called_once()
# ---------- clips_standalone 端点 ----------
class TestClipsStandaloneEndpoint:
def _user(self):
u = MagicMock()
u.user.id = "user-1"
return u
def test_empty_template_id_uses_fallback(self):
import app.api.routes.clips_standalone as cs_mod
body = cs_mod.StandaloneClipsRequest(asset_ids=["a1"])
mock_core_ret = MagicMock(created_count=3)
with (
patch.object(cs_mod, "get_or_create_default_template_id", return_value="tpl-auto") as mock_fb,
patch.object(cs_mod, "resolve_draft_plan_id", return_value="plan-1") as mock_resolve,
patch.object(cs_mod, "create_clips_from_assets_editor", return_value=mock_core_ret) as mock_core,
patch.object(cs_mod, "EditTemplateService"),
patch.object(cs_mod, "EditPlanService"),
):
resp = cs_mod.create_clips_from_assets(
body=body,
background_tasks=BackgroundTasks(),
db=MagicMock(),
current_user=self._user(),
asset_repo=MagicMock(),
)
mock_fb.assert_called_once()
assert resp.created_count == 3
assert mock_core.call_args.kwargs["template_id"] == "tpl-auto"
assert mock_core.call_args.kwargs["plan_id"] == "plan-1"
assert mock_resolve.call_args.kwargs["auto_create_default"] is False
def test_provided_template_id_skips_fallback(self):
import app.api.routes.clips_standalone as cs_mod
body = cs_mod.StandaloneClipsRequest(asset_ids=["a1"], template_id="tpl-custom")
mock_core_ret = MagicMock(created_count=2)
with (
patch.object(cs_mod, "get_or_create_default_template_id") as mock_fb,
patch.object(cs_mod, "resolve_draft_plan_id", return_value="plan-c"),
patch.object(cs_mod, "create_clips_from_assets_editor", return_value=mock_core_ret),
patch.object(cs_mod, "EditTemplateService"),
patch.object(cs_mod, "EditPlanService"),
):
resp = cs_mod.create_clips_from_assets(
body=body,
background_tasks=BackgroundTasks(),
db=MagicMock(),
current_user=self._user(),
asset_repo=MagicMock(),
)
mock_fb.assert_not_called()
assert resp.created_count == 2
def test_fallback_none_raises_500(self):
import app.api.routes.clips_standalone as cs_mod
body = cs_mod.StandaloneClipsRequest(asset_ids=["a1"])
with (
patch.object(cs_mod, "get_or_create_default_template_id", return_value=None),
patch.object(cs_mod, "EditTemplateService"),
patch.object(cs_mod, "EditPlanService"),
):
with pytest.raises(HTTPException) as e:
cs_mod.create_clips_from_assets(
body=body,
background_tasks=BackgroundTasks(),
db=MagicMock(),
current_user=self._user(),
asset_repo=MagicMock(),
)
assert e.value.status_code == 500
# ---------- router 注册 ----------
class TestRouterRegistration:
def test_clips_standalone_router_has_endpoint(self):
import app.api.routes.clips_standalone as cs_mod
paths = [r.path for r in cs_mod.router.routes]
assert "/clips/from-assets" in paths
assert cs_mod.router.tags and "Clips" in cs_mod.router.tags
def test_api_router_imports_clips_standalone(self):
import app.api.router as main_router
assert main_router.clips_standalone_router is not None
assert main_router.templates_router is not None