Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 1db2ee0808 | |||
| ac416493e0 | |||
| 9e97473eec | |||
| 34bd9372ce | |||
| 2e53c59cd7 | |||
| 9aa0c69b26 |
@@ -16,13 +16,16 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
from typing import Any
|
|
||||||
|
|
||||||
from app.auth import AuthenticatedUser, get_current_user
|
from app.auth import AuthenticatedUser, get_current_user
|
||||||
|
from app.core.storage import get_storage_service
|
||||||
|
from app.dependencies import get_asset_repository
|
||||||
from app.services.edit_plan_service import EditPlanService
|
from app.services.edit_plan_service import EditPlanService
|
||||||
from app.services.edit_template_service import EditTemplateService
|
from app.services.edit_template_service import EditTemplateService
|
||||||
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||||||
|
|
||||||
|
from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
|
||||||
|
|
||||||
from .dependencies import get_draft_plan_id, get_editor_services
|
from .dependencies import get_draft_plan_id, get_editor_services
|
||||||
from .schemas import (
|
from .schemas import (
|
||||||
ClipBatchDeleteRequest,
|
ClipBatchDeleteRequest,
|
||||||
@@ -43,30 +46,100 @@ logger = logging.getLogger(__name__)
|
|||||||
router = APIRouter(tags=["Template Editor"])
|
router = APIRouter(tags=["Template Editor"])
|
||||||
|
|
||||||
|
|
||||||
def _clip_to_response(clip) -> EditorClipResponse:
|
def _clip_to_response(clip, asset_url: str | None = None) -> EditorClipResponse:
|
||||||
"""统一构造片段响应"""
|
"""统一构造片段响应 — 与 edit_plan_clips 表字段完全对齐"""
|
||||||
|
|
||||||
|
def _enum_str(val) -> str:
|
||||||
|
return val.value if hasattr(val, "value") else str(val)
|
||||||
|
|
||||||
|
def _fmt_dt(val) -> str:
|
||||||
|
if val is None:
|
||||||
|
return ""
|
||||||
|
if hasattr(val, "isoformat"):
|
||||||
|
return val.isoformat()
|
||||||
|
return str(val)
|
||||||
|
|
||||||
return EditorClipResponse(
|
return EditorClipResponse(
|
||||||
id=clip.id,
|
id=clip.id,
|
||||||
plan_id=clip.plan_id,
|
plan_id=clip.plan_id,
|
||||||
clip_type=clip.clip_type.value
|
clip_type=_enum_str(getattr(clip, "clip_type", "")),
|
||||||
if hasattr(clip.clip_type, "value")
|
|
||||||
else str(clip.clip_type),
|
|
||||||
order=clip.order,
|
order=clip.order,
|
||||||
duration=clip.duration,
|
duration=clip.duration,
|
||||||
|
start_time=getattr(clip, "start_time", 0.0) or 0.0,
|
||||||
text_content=clip.text_content or "",
|
text_content=clip.text_content or "",
|
||||||
transition_effect=clip.transition_effect.value
|
transition_effect=_enum_str(getattr(clip, "transition_effect", "cut")),
|
||||||
if hasattr(clip.transition_effect, "value")
|
transition_duration=getattr(clip, "transition_duration", 0.0) or 0.0,
|
||||||
else str(clip.transition_effect),
|
|
||||||
playback_speed=clip.playback_speed or 1.0,
|
playback_speed=clip.playback_speed or 1.0,
|
||||||
|
asset_id=getattr(clip, "asset_id", "") or "",
|
||||||
|
asset_url=asset_url,
|
||||||
|
status=getattr(clip, "status", "pending") or "pending",
|
||||||
|
template_clip_config_id=getattr(clip, "template_clip_config_id", "") or "",
|
||||||
config=clip.config or {},
|
config=clip.config or {},
|
||||||
|
created_at=_fmt_dt(getattr(clip, "created_at", None)),
|
||||||
|
updated_at=_fmt_dt(getattr(clip, "updated_at", None)),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _build_asset_url_map(
|
||||||
|
asset_ids: list[str],
|
||||||
|
asset_repo: SQLAlchemyAssetRepository,
|
||||||
|
) -> dict[str, str | None]:
|
||||||
|
"""批量查询素材并生成签名URL映射.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
{asset_id: signed_url_or_None}
|
||||||
|
"""
|
||||||
|
if not asset_ids:
|
||||||
|
return {}
|
||||||
|
|
||||||
|
# 去重:多个 clip 可能引用同一个素材
|
||||||
|
# 去重并保持顺序
|
||||||
|
seen: set[str] = set()
|
||||||
|
unique_ids = []
|
||||||
|
for aid in asset_ids:
|
||||||
|
if aid and aid not in seen:
|
||||||
|
seen.add(aid)
|
||||||
|
unique_ids.append(aid)
|
||||||
|
|
||||||
|
result: dict[str, str | None] = {}
|
||||||
|
try:
|
||||||
|
storage = get_storage_service()
|
||||||
|
except Exception:
|
||||||
|
logger.warning("获取存储服务失败,跳过asset_url生成")
|
||||||
|
return {aid: None for aid in asset_ids}
|
||||||
|
|
||||||
|
# 批量查询所有 Asset(单次 SQL IN 查询,避免 N+1)
|
||||||
|
try:
|
||||||
|
assets = asset_repo.find_by_ids(unique_ids)
|
||||||
|
asset_map = {a.id: a for a in assets}
|
||||||
|
except Exception:
|
||||||
|
logger.warning("批量查询素材失败: asset_ids=%s", asset_ids, exc_info=True)
|
||||||
|
return {aid: None for aid in asset_ids if aid}
|
||||||
|
|
||||||
|
for aid in unique_ids:
|
||||||
|
try:
|
||||||
|
asset = asset_map.get(aid)
|
||||||
|
if asset is None:
|
||||||
|
result[aid] = None
|
||||||
|
continue
|
||||||
|
storage_key = getattr(asset, "storage_key", None) or ""
|
||||||
|
if not storage_key:
|
||||||
|
result[aid] = None
|
||||||
|
continue
|
||||||
|
result[aid] = storage.get_download_url(storage_key, expires_seconds=3600)
|
||||||
|
except Exception:
|
||||||
|
logger.warning("生成素材签名URL失败: asset_id=%s", aid, exc_info=True)
|
||||||
|
result[aid] = None
|
||||||
|
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
@router.get("/clips", response_model=EditorClipListResponse)
|
@router.get("/clips", response_model=EditorClipListResponse)
|
||||||
def list_draft_clips(
|
def list_draft_clips(
|
||||||
template_id: str,
|
template_id: str,
|
||||||
plan_id: str = Depends(get_draft_plan_id),
|
plan_id: str = Depends(get_draft_plan_id),
|
||||||
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
|
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
|
||||||
|
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
|
||||||
skip: int = Query(default=0, ge=0),
|
skip: int = Query(default=0, ge=0),
|
||||||
limit: int = Query(default=100, ge=1, le=500),
|
limit: int = Query(default=100, ge=1, le=500),
|
||||||
_: AuthenticatedUser = Depends(get_current_user),
|
_: AuthenticatedUser = Depends(get_current_user),
|
||||||
@@ -75,8 +148,17 @@ def list_draft_clips(
|
|||||||
_, plan_svc = services
|
_, plan_svc = services
|
||||||
clips = plan_svc.list_clips(plan_id, skip=skip, limit=limit)
|
clips = plan_svc.list_clips(plan_id, skip=skip, limit=limit)
|
||||||
total = plan_svc.count_clips(plan_id)
|
total = plan_svc.count_clips(plan_id)
|
||||||
|
|
||||||
|
# 批量解析素材签名URL
|
||||||
|
asset_ids = [getattr(c, "asset_id", "") or "" for c in clips]
|
||||||
|
asset_ids = [aid for aid in asset_ids if aid]
|
||||||
|
url_map = _build_asset_url_map(asset_ids, asset_repo)
|
||||||
|
|
||||||
return EditorClipListResponse(
|
return EditorClipListResponse(
|
||||||
items=[_clip_to_response(c) for c in clips],
|
items=[
|
||||||
|
_clip_to_response(c, asset_url=url_map.get(getattr(c, "asset_id", "") or ""))
|
||||||
|
for c in clips
|
||||||
|
],
|
||||||
total=total,
|
total=total,
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -156,6 +238,7 @@ def get_draft_clip_detail(
|
|||||||
clip_id: str,
|
clip_id: str,
|
||||||
plan_id: str = Depends(get_draft_plan_id),
|
plan_id: str = Depends(get_draft_plan_id),
|
||||||
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
|
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
|
||||||
|
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
|
||||||
_: AuthenticatedUser = Depends(get_current_user),
|
_: AuthenticatedUser = Depends(get_current_user),
|
||||||
):
|
):
|
||||||
"""获取草稿中的片段详情"""
|
"""获取草稿中的片段详情"""
|
||||||
@@ -165,16 +248,20 @@ def get_draft_clip_detail(
|
|||||||
raise HTTPException(status_code=404, detail="片段不存在")
|
raise HTTPException(status_code=404, detail="片段不存在")
|
||||||
if clip.plan_id != plan_id:
|
if clip.plan_id != plan_id:
|
||||||
raise HTTPException(status_code=404, detail="片段不存在")
|
raise HTTPException(status_code=404, detail="片段不存在")
|
||||||
return _clip_to_response(clip)
|
|
||||||
|
asset_id = getattr(clip, "asset_id", "") or ""
|
||||||
|
url_map = _build_asset_url_map([asset_id], asset_repo) if asset_id else {}
|
||||||
|
return _clip_to_response(clip, asset_url=url_map.get(asset_id))
|
||||||
|
|
||||||
|
|
||||||
@router.post("/clips/{clip_id}/split", response_model=dict[str, Any], status_code=status.HTTP_200_OK)
|
@router.post("/clips/{clip_id}/split", status_code=status.HTTP_200_OK)
|
||||||
def split_draft_clip(
|
def split_draft_clip(
|
||||||
template_id: str,
|
template_id: str,
|
||||||
clip_id: str,
|
clip_id: str,
|
||||||
body: SplitClipRequest,
|
body: SplitClipRequest,
|
||||||
plan_id: str = Depends(get_draft_plan_id),
|
plan_id: str = Depends(get_draft_plan_id),
|
||||||
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
|
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
|
||||||
|
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
|
||||||
_: AuthenticatedUser = Depends(get_current_user),
|
_: AuthenticatedUser = Depends(get_current_user),
|
||||||
):
|
):
|
||||||
"""将一个片段从指定时间点分割为两个片段"""
|
"""将一个片段从指定时间点分割为两个片段"""
|
||||||
@@ -190,32 +277,22 @@ def split_draft_clip(
|
|||||||
) from exc
|
) from exc
|
||||||
left = result["left_clip"]
|
left = result["left_clip"]
|
||||||
right = result["right_clip"]
|
right = result["right_clip"]
|
||||||
|
asset_ids = [getattr(left, "asset_id", "") or "", getattr(right, "asset_id", "") or ""]
|
||||||
|
asset_ids = [a for a in asset_ids if a]
|
||||||
|
url_map = _build_asset_url_map(asset_ids, asset_repo)
|
||||||
return {
|
return {
|
||||||
"left_clip": {
|
"left_clip": _clip_to_response(left, asset_url=url_map.get(getattr(left, "asset_id", "") or "")),
|
||||||
"id": left.id,
|
"right_clip": _clip_to_response(right, asset_url=url_map.get(getattr(right, "asset_id", "") or "")),
|
||||||
"plan_id": left.plan_id,
|
|
||||||
"clip_type": left.clip_type,
|
|
||||||
"order": left.order,
|
|
||||||
"duration": left.duration,
|
|
||||||
"start_time": left.start_time,
|
|
||||||
},
|
|
||||||
"right_clip": {
|
|
||||||
"id": right.id,
|
|
||||||
"plan_id": right.plan_id,
|
|
||||||
"clip_type": right.clip_type,
|
|
||||||
"order": right.order,
|
|
||||||
"duration": right.duration,
|
|
||||||
"start_time": right.start_time,
|
|
||||||
},
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@router.post("/clips/merge", response_model=dict[str, Any], status_code=status.HTTP_200_OK)
|
@router.post("/clips/merge", status_code=status.HTTP_200_OK)
|
||||||
def merge_draft_clips(
|
def merge_draft_clips(
|
||||||
template_id: str,
|
template_id: str,
|
||||||
body: MergeClipsRequest,
|
body: MergeClipsRequest,
|
||||||
plan_id: str = Depends(get_draft_plan_id),
|
plan_id: str = Depends(get_draft_plan_id),
|
||||||
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
|
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
|
||||||
|
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
|
||||||
_: AuthenticatedUser = Depends(get_current_user),
|
_: AuthenticatedUser = Depends(get_current_user),
|
||||||
):
|
):
|
||||||
"""将多个连续的同类型片段合并为一个片段"""
|
"""将多个连续的同类型片段合并为一个片段"""
|
||||||
@@ -230,13 +307,11 @@ def merge_draft_clips(
|
|||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)
|
status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)
|
||||||
) from exc
|
) from exc
|
||||||
|
asset_id = getattr(merged, "asset_id", "") or ""
|
||||||
|
url_map = _build_asset_url_map([asset_id], asset_repo) if asset_id else {}
|
||||||
return {
|
return {
|
||||||
"id": merged.id,
|
"merged_clip": _clip_to_response(merged, asset_url=url_map.get(asset_id)),
|
||||||
"plan_id": merged.plan_id,
|
"deleted_clip_ids": body.clip_ids,
|
||||||
"clip_type": merged.clip_type,
|
|
||||||
"order": merged.order,
|
|
||||||
"duration": merged.duration,
|
|
||||||
"text_content": merged.text_content,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -227,6 +227,7 @@ class ClipsFromAssetsResponse(BaseModel):
|
|||||||
|
|
||||||
success: bool = True
|
success: bool = True
|
||||||
created_count: int
|
created_count: int
|
||||||
|
plan_id: str = ""
|
||||||
message: str = ""
|
message: str = ""
|
||||||
clip_ids: List[str] = Field(default_factory=list, description="创建的片段ID列表")
|
clip_ids: List[str] = Field(default_factory=list, description="创建的片段ID列表")
|
||||||
|
|
||||||
@@ -439,17 +440,28 @@ class EditorUpdateRequest(BaseModel):
|
|||||||
|
|
||||||
|
|
||||||
class EditorClipResponse(BaseModel):
|
class EditorClipResponse(BaseModel):
|
||||||
"""片段响应"""
|
"""片段响应 — 与数据库 edit_plan_clips 表字段对齐"""
|
||||||
|
|
||||||
id: str
|
id: str
|
||||||
plan_id: str
|
plan_id: str
|
||||||
clip_type: str
|
clip_type: str
|
||||||
order: int
|
order: int
|
||||||
duration: float
|
duration: float
|
||||||
|
start_time: float = 0.0
|
||||||
text_content: str = ""
|
text_content: str = ""
|
||||||
transition_effect: str = "cut"
|
transition_effect: str = "cut"
|
||||||
|
transition_duration: float = 0.0
|
||||||
playback_speed: float = 1.0
|
playback_speed: float = 1.0
|
||||||
|
asset_id: str = ""
|
||||||
|
asset_url: str | None = Field(
|
||||||
|
default=None,
|
||||||
|
description="素材视频签名URL(1小时有效),用于前端预览播放",
|
||||||
|
)
|
||||||
|
status: str = "pending"
|
||||||
|
template_clip_config_id: str = ""
|
||||||
config: dict[str, Any] = Field(default_factory=dict)
|
config: dict[str, Any] = Field(default_factory=dict)
|
||||||
|
created_at: str = ""
|
||||||
|
updated_at: str = ""
|
||||||
|
|
||||||
|
|
||||||
class EditorClipListResponse(BaseModel):
|
class EditorClipListResponse(BaseModel):
|
||||||
|
|||||||
@@ -101,6 +101,13 @@ class SQLAlchemyAssetRepository:
|
|||||||
return None
|
return None
|
||||||
return self._to_domain(model)
|
return self._to_domain(model)
|
||||||
|
|
||||||
|
def find_by_ids(self, asset_ids: list[str]) -> list[Asset]:
|
||||||
|
"""批量查询素材(单次 SQL IN 查询,避免 N+1)。"""
|
||||||
|
if not asset_ids:
|
||||||
|
return []
|
||||||
|
models = self.session.query(AssetModel).filter(AssetModel.id.in_(asset_ids)).all()
|
||||||
|
return [self._to_domain(m) for m in models]
|
||||||
|
|
||||||
def get(self, asset_id: str) -> Asset | None:
|
def get(self, asset_id: str) -> Asset | None:
|
||||||
return self.find_by_id(asset_id)
|
return self.find_by_id(asset_id)
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,496 @@
|
|||||||
|
"""片段管理路由 clips.py 增量覆盖率测试.
|
||||||
|
|
||||||
|
覆盖 PR fix/clips-api-response-structure 新增代码:
|
||||||
|
- _clip_to_response: 枚举转换、日期格式化、asset_url 参数
|
||||||
|
- _build_asset_url_map: 批量素材 URL 解析(空列表/异常/正常路径)
|
||||||
|
- 路由层 asset_repo 注入与 URL 拼接逻辑
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
from enum import Enum
|
||||||
|
from pathlib import Path
|
||||||
|
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")
|
||||||
|
|
||||||
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi import FastAPI
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 常量与工厂
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
TEST_TEMPLATE_ID = "tmpl-test-001"
|
||||||
|
TEST_PLAN_ID = "plan-draft-001"
|
||||||
|
TEST_USER_ID = "user-001"
|
||||||
|
|
||||||
|
|
||||||
|
def _auth_user():
|
||||||
|
u = MagicMock()
|
||||||
|
u.user.id = TEST_USER_ID
|
||||||
|
u.user_id = TEST_USER_ID
|
||||||
|
return u
|
||||||
|
|
||||||
|
|
||||||
|
def _clip(**overrides):
|
||||||
|
"""构造 mock clip,支持 Enum 类型字段"""
|
||||||
|
c = MagicMock()
|
||||||
|
c.id = overrides.get("id", "clip-001")
|
||||||
|
c.plan_id = overrides.get("plan_id", TEST_PLAN_ID)
|
||||||
|
c.clip_type = overrides.get("clip_type", "video")
|
||||||
|
c.order = overrides.get("order", 0)
|
||||||
|
c.duration = overrides.get("duration", 10.0)
|
||||||
|
c.start_time = overrides.get("start_time", 0.0)
|
||||||
|
c.text_content = overrides.get("text_content", "")
|
||||||
|
c.transition_effect = overrides.get("transition_effect", "cut")
|
||||||
|
c.transition_duration = overrides.get("transition_duration", 0.0)
|
||||||
|
c.playback_speed = overrides.get("playback_speed", 1.0)
|
||||||
|
c.asset_id = overrides.get("asset_id", "")
|
||||||
|
c.status = overrides.get("status", "ready")
|
||||||
|
c.template_clip_config_id = overrides.get("template_clip_config_id", "")
|
||||||
|
c.config = overrides.get("config", {})
|
||||||
|
c.created_at = overrides.get("created_at", None)
|
||||||
|
c.updated_at = overrides.get("updated_at", None)
|
||||||
|
return c
|
||||||
|
|
||||||
|
|
||||||
|
def _services(plan_svc_overrides=None):
|
||||||
|
tpl = MagicMock()
|
||||||
|
plan = MagicMock()
|
||||||
|
if plan_svc_overrides:
|
||||||
|
for k, v in plan_svc_overrides.items():
|
||||||
|
setattr(plan, k, v)
|
||||||
|
return tpl, plan
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 单元测试: _clip_to_response
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestClipToResponse:
|
||||||
|
"""_clip_to_response 纯函数测试 — 覆盖行 53-80"""
|
||||||
|
|
||||||
|
def test_basic_fields(self):
|
||||||
|
from app.api.routes.templates_editor.clips import _clip_to_response
|
||||||
|
|
||||||
|
c = _clip(id="c1", order=3, duration=5.5, text_content="hello")
|
||||||
|
resp = _clip_to_response(c)
|
||||||
|
assert resp.id == "c1"
|
||||||
|
assert resp.order == 3
|
||||||
|
assert resp.duration == 5.5
|
||||||
|
assert resp.text_content == "hello"
|
||||||
|
assert resp.asset_url is None
|
||||||
|
|
||||||
|
def test_enum_clip_type(self):
|
||||||
|
"""Enum 值应被 .value 解包"""
|
||||||
|
from app.api.routes.templates_editor.clips import _clip_to_response
|
||||||
|
|
||||||
|
class ClipType(str, Enum):
|
||||||
|
VIDEO = "video"
|
||||||
|
AUDIO = "audio"
|
||||||
|
|
||||||
|
c = _clip(clip_type=ClipType.VIDEO)
|
||||||
|
resp = _clip_to_response(c)
|
||||||
|
assert resp.clip_type == "video"
|
||||||
|
|
||||||
|
def test_plain_string_clip_type(self):
|
||||||
|
"""非 Enum 字符串直接用 str()"""
|
||||||
|
from app.api.routes.templates_editor.clips import _clip_to_response
|
||||||
|
|
||||||
|
c = _clip(clip_type="main")
|
||||||
|
resp = _clip_to_response(c)
|
||||||
|
assert resp.clip_type == "main"
|
||||||
|
|
||||||
|
def test_enum_transition_effect(self):
|
||||||
|
from app.api.routes.templates_editor.clips import _clip_to_response
|
||||||
|
|
||||||
|
class Transition(str, Enum):
|
||||||
|
FADE = "fade"
|
||||||
|
|
||||||
|
c = _clip(transition_effect=Transition.FADE)
|
||||||
|
resp = _clip_to_response(c)
|
||||||
|
assert resp.transition_effect == "fade"
|
||||||
|
|
||||||
|
def test_default_transition_when_none(self):
|
||||||
|
"""transition_effect 缺失时默认 cut"""
|
||||||
|
from app.api.routes.templates_editor.clips import _clip_to_response
|
||||||
|
|
||||||
|
c = _clip()
|
||||||
|
del c.transition_effect # 触发 getattr default
|
||||||
|
resp = _clip_to_response(c)
|
||||||
|
assert resp.transition_effect == "cut"
|
||||||
|
|
||||||
|
def test_asset_url_passed(self):
|
||||||
|
"""asset_url 参数应透传到响应"""
|
||||||
|
from app.api.routes.templates_editor.clips import _clip_to_response
|
||||||
|
|
||||||
|
c = _clip(asset_id="a1")
|
||||||
|
resp = _clip_to_response(c, asset_url="https://signed-url.example.com/video.mp4")
|
||||||
|
assert resp.asset_url == "https://signed-url.example.com/video.mp4"
|
||||||
|
|
||||||
|
def test_asset_url_none_by_default(self):
|
||||||
|
from app.api.routes.templates_editor.clips import _clip_to_response
|
||||||
|
|
||||||
|
c = _clip()
|
||||||
|
resp = _clip_to_response(c)
|
||||||
|
assert resp.asset_url is None
|
||||||
|
|
||||||
|
def test_datetime_isoformat(self):
|
||||||
|
"""datetime 对象应被 isoformat()"""
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
|
from app.api.routes.templates_editor.clips import _clip_to_response
|
||||||
|
|
||||||
|
dt = datetime(2026, 8, 17, 12, 0, 0)
|
||||||
|
c = _clip(created_at=dt, updated_at=dt)
|
||||||
|
resp = _clip_to_response(c)
|
||||||
|
assert "2026-08-17" in resp.created_at
|
||||||
|
assert "2026-08-17" in resp.updated_at
|
||||||
|
|
||||||
|
def test_none_datetime_empty_string(self):
|
||||||
|
"""None 日期应格式化为空字符串"""
|
||||||
|
from app.api.routes.templates_editor.clips import _clip_to_response
|
||||||
|
|
||||||
|
c = _clip(created_at=None, updated_at=None)
|
||||||
|
resp = _clip_to_response(c)
|
||||||
|
assert resp.created_at == ""
|
||||||
|
assert resp.updated_at == ""
|
||||||
|
|
||||||
|
def test_string_datetime_passthrough(self):
|
||||||
|
"""已经是字符串的日期直接 str()"""
|
||||||
|
from app.api.routes.templates_editor.clips import _clip_to_response
|
||||||
|
|
||||||
|
c = _clip(created_at="2026-08-17T00:00:00")
|
||||||
|
resp = _clip_to_response(c)
|
||||||
|
assert resp.created_at == "2026-08-17T00:00:00"
|
||||||
|
|
||||||
|
def test_none_defaults_for_optional_fields(self):
|
||||||
|
"""None/缺失字段的默认值"""
|
||||||
|
from app.api.routes.templates_editor.clips import _clip_to_response
|
||||||
|
|
||||||
|
c = _clip(asset_id=None, status=None, template_clip_config_id=None)
|
||||||
|
resp = _clip_to_response(c)
|
||||||
|
assert resp.asset_id == ""
|
||||||
|
assert resp.status == "pending"
|
||||||
|
assert resp.template_clip_config_id == ""
|
||||||
|
|
||||||
|
def test_zero_duration_fallback(self):
|
||||||
|
"""duration=0 → playback_speed 默认 1.0"""
|
||||||
|
from app.api.routes.templates_editor.clips import _clip_to_response
|
||||||
|
|
||||||
|
c = _clip(playback_speed=None)
|
||||||
|
resp = _clip_to_response(c)
|
||||||
|
assert resp.playback_speed == 1.0
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 单元测试: _build_asset_url_map
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestBuildAssetUrlMap:
|
||||||
|
"""_build_asset_url_map 测试 — 覆盖行 93-118"""
|
||||||
|
|
||||||
|
def test_empty_list(self):
|
||||||
|
"""空 asset_ids 直接返回空 dict"""
|
||||||
|
from app.api.routes.templates_editor.clips import _build_asset_url_map
|
||||||
|
|
||||||
|
repo = MagicMock()
|
||||||
|
result = _build_asset_url_map([], repo)
|
||||||
|
assert result == {}
|
||||||
|
|
||||||
|
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||||||
|
def test_storage_service_failure(self, mock_get_storage):
|
||||||
|
"""存储服务获取失败时返回全 None"""
|
||||||
|
from app.api.routes.templates_editor.clips import _build_asset_url_map
|
||||||
|
|
||||||
|
mock_get_storage.side_effect = RuntimeError("storage unavailable")
|
||||||
|
repo = MagicMock()
|
||||||
|
result = _build_asset_url_map(["a1", "a2"], repo)
|
||||||
|
assert result == {"a1": None, "a2": None}
|
||||||
|
|
||||||
|
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||||||
|
def test_asset_not_found(self, mock_get_storage):
|
||||||
|
"""asset_id 找不到对应素材 → None"""
|
||||||
|
from app.api.routes.templates_editor.clips import _build_asset_url_map
|
||||||
|
|
||||||
|
storage = MagicMock()
|
||||||
|
mock_get_storage.return_value = storage
|
||||||
|
repo = MagicMock()
|
||||||
|
repo.find_by_ids.return_value = []
|
||||||
|
|
||||||
|
result = _build_asset_url_map(["missing-id"], repo)
|
||||||
|
assert result == {"missing-id": None}
|
||||||
|
|
||||||
|
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||||||
|
def test_no_storage_key(self, mock_get_storage):
|
||||||
|
"""素材没有 storage_key → None"""
|
||||||
|
from app.api.routes.templates_editor.clips import _build_asset_url_map
|
||||||
|
|
||||||
|
storage = MagicMock()
|
||||||
|
mock_get_storage.return_value = storage
|
||||||
|
repo = MagicMock()
|
||||||
|
asset = MagicMock()
|
||||||
|
asset.id = "a1"
|
||||||
|
asset.storage_key = ""
|
||||||
|
repo.find_by_ids.return_value = [asset]
|
||||||
|
|
||||||
|
result = _build_asset_url_map(["a1"], repo)
|
||||||
|
assert result == {"a1": None}
|
||||||
|
|
||||||
|
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||||||
|
def test_successful_url_generation(self, mock_get_storage):
|
||||||
|
"""正常路径:返回签名 URL"""
|
||||||
|
from app.api.routes.templates_editor.clips import _build_asset_url_map
|
||||||
|
|
||||||
|
storage = MagicMock()
|
||||||
|
storage.get_download_url.return_value = "https://cdn.example.com/signed.mp4"
|
||||||
|
mock_get_storage.return_value = storage
|
||||||
|
repo = MagicMock()
|
||||||
|
asset = MagicMock()
|
||||||
|
asset.id = "a1"
|
||||||
|
asset.storage_key = "videos/test.mp4"
|
||||||
|
repo.find_by_ids.return_value = [asset]
|
||||||
|
|
||||||
|
result = _build_asset_url_map(["a1"], repo)
|
||||||
|
assert result == {"a1": "https://cdn.example.com/signed.mp4"}
|
||||||
|
storage.get_download_url.assert_called_once_with("videos/test.mp4", expires_seconds=3600)
|
||||||
|
|
||||||
|
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||||||
|
def test_exception_during_url_generation(self, mock_get_storage):
|
||||||
|
"""单个 asset 生成 URL 异常 → None,不影响其他"""
|
||||||
|
from app.api.routes.templates_editor.clips import _build_asset_url_map
|
||||||
|
|
||||||
|
storage = MagicMock()
|
||||||
|
storage.get_download_url.side_effect = [Exception("boom"), "https://ok.com/v2"]
|
||||||
|
mock_get_storage.return_value = storage
|
||||||
|
|
||||||
|
repo = MagicMock()
|
||||||
|
asset1 = MagicMock()
|
||||||
|
asset1.id = "a1"
|
||||||
|
asset1.storage_key = "v1.mp4"
|
||||||
|
asset2 = MagicMock()
|
||||||
|
asset2.id = "a2"
|
||||||
|
asset2.storage_key = "v2.mp4"
|
||||||
|
repo.find_by_ids.return_value = [asset1, asset2]
|
||||||
|
|
||||||
|
result = _build_asset_url_map(["a1", "a2"], repo)
|
||||||
|
assert result["a1"] is None
|
||||||
|
assert result["a2"] == "https://ok.com/v2"
|
||||||
|
|
||||||
|
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||||||
|
def test_skip_empty_asset_id(self, mock_get_storage):
|
||||||
|
"""空字符串 asset_id 被跳过"""
|
||||||
|
from app.api.routes.templates_editor.clips import _build_asset_url_map
|
||||||
|
|
||||||
|
storage = MagicMock()
|
||||||
|
mock_get_storage.return_value = storage
|
||||||
|
repo = MagicMock()
|
||||||
|
|
||||||
|
result = _build_asset_url_map(["", "a1"], repo)
|
||||||
|
# "" not in result because it's skipped by `if not aid: continue`
|
||||||
|
assert "" not in result
|
||||||
|
|
||||||
|
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||||||
|
def test_multiple_assets_mixed(self, mock_get_storage):
|
||||||
|
"""混合场景:正常+异常+缺失"""
|
||||||
|
from app.api.routes.templates_editor.clips import _build_asset_url_map
|
||||||
|
|
||||||
|
storage = MagicMock()
|
||||||
|
storage.get_download_url.return_value = "https://cdn.com/ok.mp4"
|
||||||
|
mock_get_storage.return_value = storage
|
||||||
|
|
||||||
|
repo = MagicMock()
|
||||||
|
good_asset = MagicMock()
|
||||||
|
good_asset.id = "a1"
|
||||||
|
good_asset.storage_key = "good.mp4"
|
||||||
|
# a1=good, a2=not found, a3=good
|
||||||
|
good_asset2 = MagicMock()
|
||||||
|
good_asset2.id = "a3"
|
||||||
|
good_asset2.storage_key = "good.mp4"
|
||||||
|
repo.find_by_ids.return_value = [good_asset, good_asset2]
|
||||||
|
|
||||||
|
result = _build_asset_url_map(["a1", "a2", "a3"], repo)
|
||||||
|
assert result["a1"] == "https://cdn.com/ok.mp4"
|
||||||
|
assert result["a2"] is None
|
||||||
|
assert result["a3"] == "https://cdn.com/ok.mp4"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 集成测试: 路由层 asset_repo 注入
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestClipRoutesAssetIntegration:
|
||||||
|
"""路由层测试 — 覆盖 asset_url 在 list/detail/split/merge 中的拼接逻辑"""
|
||||||
|
|
||||||
|
def _create_app(self, plan_svc_config=None):
|
||||||
|
from app.api.routes import templates_editor as editor_module
|
||||||
|
from app.dependencies import get_asset_repository
|
||||||
|
|
||||||
|
mock_clip_1 = _clip(id="c1", asset_id="asset-001")
|
||||||
|
mock_clip_2 = _clip(id="c2", asset_id="")
|
||||||
|
|
||||||
|
mock_tpl_svc = MagicMock()
|
||||||
|
mock_plan_svc = MagicMock()
|
||||||
|
mock_plan_svc.list_clips.return_value = [mock_clip_1, mock_clip_2]
|
||||||
|
mock_plan_svc.count_clips.return_value = 2
|
||||||
|
mock_plan_svc.get_clip.return_value = mock_clip_1
|
||||||
|
mock_plan_svc.create_clip.return_value = _clip(id="c-new", asset_id="")
|
||||||
|
mock_plan_svc.update_clip.return_value = _clip(id="c1", duration=15.0)
|
||||||
|
mock_plan_svc.delete_clip.return_value = True
|
||||||
|
mock_plan_svc.split_clip.return_value = {
|
||||||
|
"left_clip": _clip(id="c-left", asset_id="asset-L"),
|
||||||
|
"right_clip": _clip(id="c-right", asset_id="asset-R"),
|
||||||
|
}
|
||||||
|
mock_plan_svc.merge_clips.return_value = _clip(id="c-merged", asset_id="asset-M")
|
||||||
|
|
||||||
|
if plan_svc_config:
|
||||||
|
for k, v in plan_svc_config.items():
|
||||||
|
setattr(mock_plan_svc, k, v)
|
||||||
|
|
||||||
|
def _deps():
|
||||||
|
return mock_tpl_svc, mock_plan_svc
|
||||||
|
|
||||||
|
mock_asset_repo = MagicMock()
|
||||||
|
|
||||||
|
app = FastAPI()
|
||||||
|
app.include_router(
|
||||||
|
editor_module.router,
|
||||||
|
prefix="/api/v1/templates/{template_id}/editor",
|
||||||
|
)
|
||||||
|
app.dependency_overrides[editor_module.get_current_user] = _auth_user
|
||||||
|
app.dependency_overrides[editor_module.get_draft_plan_id] = lambda: TEST_PLAN_ID
|
||||||
|
app.dependency_overrides[editor_module.get_editor_services] = _deps
|
||||||
|
app.dependency_overrides[get_asset_repository] = lambda: mock_asset_repo
|
||||||
|
|
||||||
|
return TestClient(app), mock_plan_svc, mock_asset_repo
|
||||||
|
|
||||||
|
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||||||
|
def test_list_clips_includes_asset_urls(self, mock_get_storage):
|
||||||
|
"""GET /clips 应为有 asset_id 的片段返回签名 URL"""
|
||||||
|
storage = MagicMock()
|
||||||
|
storage.get_download_url.return_value = "https://cdn.com/c1.mp4"
|
||||||
|
mock_get_storage.return_value = storage
|
||||||
|
|
||||||
|
client, _, asset_repo = self._create_app()
|
||||||
|
asset = MagicMock()
|
||||||
|
asset.id = "asset-001"
|
||||||
|
asset.storage_key = "videos/c1.mp4"
|
||||||
|
asset_repo.find_by_ids.return_value = [asset]
|
||||||
|
|
||||||
|
resp = client.get(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips")
|
||||||
|
assert resp.status_code == 200
|
||||||
|
data = resp.json()
|
||||||
|
items = data["items"]
|
||||||
|
assert len(items) == 2
|
||||||
|
# c1 has asset_id → should have url
|
||||||
|
assert items[0]["asset_url"] == "https://cdn.com/c1.mp4"
|
||||||
|
# c2 has empty asset_id → None
|
||||||
|
assert items[1]["asset_url"] is None
|
||||||
|
|
||||||
|
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||||||
|
def test_get_clip_detail_with_asset_url(self, mock_get_storage):
|
||||||
|
"""GET /clips/{clip_id} 应返回素材签名 URL"""
|
||||||
|
storage = MagicMock()
|
||||||
|
storage.get_download_url.return_value = "https://cdn.com/detail.mp4"
|
||||||
|
mock_get_storage.return_value = storage
|
||||||
|
|
||||||
|
client, _, asset_repo = self._create_app()
|
||||||
|
asset = MagicMock()
|
||||||
|
asset.id = "asset-001"
|
||||||
|
asset.storage_key = "videos/detail.mp4"
|
||||||
|
asset_repo.find_by_ids.return_value = [asset]
|
||||||
|
|
||||||
|
resp = client.get(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/clip-001")
|
||||||
|
assert resp.status_code == 200
|
||||||
|
assert resp.json()["asset_url"] == "https://cdn.com/detail.mp4"
|
||||||
|
|
||||||
|
def test_get_clip_detail_no_asset(self):
|
||||||
|
"""片段没有 asset_id 时不应调用 URL 解析"""
|
||||||
|
client, plan_svc, asset_repo = self._create_app()
|
||||||
|
# 返回没有 asset_id 的片段
|
||||||
|
plan_svc.get_clip.return_value = _clip(id="c-no-asset", asset_id="")
|
||||||
|
|
||||||
|
resp = client.get(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/c-no-asset")
|
||||||
|
assert resp.status_code == 200
|
||||||
|
assert resp.json()["asset_url"] is None
|
||||||
|
|
||||||
|
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||||||
|
def test_split_clip_returns_asset_urls(self, mock_get_storage):
|
||||||
|
"""POST /clips/{clip_id}/split 返回的左右片段应带签名 URL"""
|
||||||
|
storage = MagicMock()
|
||||||
|
storage.get_download_url.side_effect = ["https://cdn.com/L.mp4", "https://cdn.com/R.mp4"]
|
||||||
|
mock_get_storage.return_value = storage
|
||||||
|
|
||||||
|
client, _, asset_repo = self._create_app()
|
||||||
|
asset_l = MagicMock()
|
||||||
|
asset_l.storage_key = "videos/L.mp4"
|
||||||
|
asset_r = MagicMock()
|
||||||
|
asset_r.storage_key = "videos/R.mp4"
|
||||||
|
asset_l.id = "asset-L"
|
||||||
|
asset_r.id = "asset-R"
|
||||||
|
asset_repo.find_by_ids.return_value = [asset_l, asset_r]
|
||||||
|
|
||||||
|
resp = client.post(
|
||||||
|
f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/clip-001/split",
|
||||||
|
json={"split_time": 5.0},
|
||||||
|
)
|
||||||
|
assert resp.status_code == 200
|
||||||
|
data = resp.json()
|
||||||
|
assert data["left_clip"]["asset_url"] == "https://cdn.com/L.mp4"
|
||||||
|
assert data["right_clip"]["asset_url"] == "https://cdn.com/R.mp4"
|
||||||
|
|
||||||
|
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||||||
|
def test_merge_clips_returns_asset_url(self, mock_get_storage):
|
||||||
|
"""POST /clips/merge 返回的合并片段应带签名 URL"""
|
||||||
|
storage = MagicMock()
|
||||||
|
storage.get_download_url.return_value = "https://cdn.com/M.mp4"
|
||||||
|
mock_get_storage.return_value = storage
|
||||||
|
|
||||||
|
client, _, asset_repo = self._create_app()
|
||||||
|
asset = MagicMock()
|
||||||
|
asset.id = "asset-M"
|
||||||
|
asset.storage_key = "videos/M.mp4"
|
||||||
|
asset_repo.find_by_ids.return_value = [asset]
|
||||||
|
|
||||||
|
resp = client.post(
|
||||||
|
f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/merge",
|
||||||
|
json={"clip_ids": ["c1", "c2"]},
|
||||||
|
)
|
||||||
|
assert resp.status_code == 200
|
||||||
|
data = resp.json()
|
||||||
|
assert data["merged_clip"]["asset_url"] == "https://cdn.com/M.mp4"
|
||||||
|
assert data["deleted_clip_ids"] == ["c1", "c2"]
|
||||||
|
|
||||||
|
def test_merge_clips_not_found(self):
|
||||||
|
"""merge 时某片段不存在应返回 404"""
|
||||||
|
client, plan_svc, _ = self._create_app()
|
||||||
|
plan_svc.get_clip.return_value = None
|
||||||
|
|
||||||
|
resp = client.post(
|
||||||
|
f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/merge",
|
||||||
|
json={"clip_ids": ["nonexistent-1", "nonexistent-2"]},
|
||||||
|
)
|
||||||
|
assert resp.status_code == 404
|
||||||
|
|
||||||
|
def test_delete_clip_success(self):
|
||||||
|
"""DELETE /clips/{clip_id} 成功返回 204"""
|
||||||
|
client, _, _ = self._create_app()
|
||||||
|
resp = client.delete(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/clip-001")
|
||||||
|
assert resp.status_code == 204
|
||||||
|
|
||||||
|
def test_delete_clip_not_found(self):
|
||||||
|
"""DELETE 片段不存在返回 404"""
|
||||||
|
client, plan_svc, _ = self._create_app()
|
||||||
|
plan_svc.delete_clip.return_value = False
|
||||||
|
resp = client.delete(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/bad-id")
|
||||||
|
assert resp.status_code == 404
|
||||||
Reference in New Issue
Block a user