Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 1db2ee0808 | |||
| ac416493e0 | |||
| 9e97473eec | |||
| 34bd9372ce | |||
| 2e53c59cd7 | |||
| 9aa0c69b26 |
@@ -16,13 +16,16 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
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_template_service import EditTemplateService
|
||||
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 .schemas import (
|
||||
ClipBatchDeleteRequest,
|
||||
@@ -43,30 +46,100 @@ logger = logging.getLogger(__name__)
|
||||
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(
|
||||
id=clip.id,
|
||||
plan_id=clip.plan_id,
|
||||
clip_type=clip.clip_type.value
|
||||
if hasattr(clip.clip_type, "value")
|
||||
else str(clip.clip_type),
|
||||
clip_type=_enum_str(getattr(clip, "clip_type", "")),
|
||||
order=clip.order,
|
||||
duration=clip.duration,
|
||||
start_time=getattr(clip, "start_time", 0.0) or 0.0,
|
||||
text_content=clip.text_content or "",
|
||||
transition_effect=clip.transition_effect.value
|
||||
if hasattr(clip.transition_effect, "value")
|
||||
else str(clip.transition_effect),
|
||||
transition_effect=_enum_str(getattr(clip, "transition_effect", "cut")),
|
||||
transition_duration=getattr(clip, "transition_duration", 0.0) or 0.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 {},
|
||||
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)
|
||||
def list_draft_clips(
|
||||
template_id: str,
|
||||
plan_id: str = Depends(get_draft_plan_id),
|
||||
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
|
||||
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
|
||||
skip: int = Query(default=0, ge=0),
|
||||
limit: int = Query(default=100, ge=1, le=500),
|
||||
_: AuthenticatedUser = Depends(get_current_user),
|
||||
@@ -75,8 +148,17 @@ def list_draft_clips(
|
||||
_, plan_svc = services
|
||||
clips = plan_svc.list_clips(plan_id, skip=skip, limit=limit)
|
||||
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(
|
||||
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,
|
||||
)
|
||||
|
||||
@@ -156,6 +238,7 @@ def get_draft_clip_detail(
|
||||
clip_id: str,
|
||||
plan_id: str = Depends(get_draft_plan_id),
|
||||
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
|
||||
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
|
||||
_: AuthenticatedUser = Depends(get_current_user),
|
||||
):
|
||||
"""获取草稿中的片段详情"""
|
||||
@@ -165,16 +248,20 @@ def get_draft_clip_detail(
|
||||
raise HTTPException(status_code=404, detail="片段不存在")
|
||||
if clip.plan_id != plan_id:
|
||||
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(
|
||||
template_id: str,
|
||||
clip_id: str,
|
||||
body: SplitClipRequest,
|
||||
plan_id: str = Depends(get_draft_plan_id),
|
||||
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
|
||||
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
|
||||
_: AuthenticatedUser = Depends(get_current_user),
|
||||
):
|
||||
"""将一个片段从指定时间点分割为两个片段"""
|
||||
@@ -190,32 +277,22 @@ def split_draft_clip(
|
||||
) from exc
|
||||
left = result["left_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 {
|
||||
"left_clip": {
|
||||
"id": left.id,
|
||||
"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,
|
||||
},
|
||||
"left_clip": _clip_to_response(left, asset_url=url_map.get(getattr(left, "asset_id", "") or "")),
|
||||
"right_clip": _clip_to_response(right, asset_url=url_map.get(getattr(right, "asset_id", "") or "")),
|
||||
}
|
||||
|
||||
|
||||
@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(
|
||||
template_id: str,
|
||||
body: MergeClipsRequest,
|
||||
plan_id: str = Depends(get_draft_plan_id),
|
||||
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
|
||||
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
|
||||
_: AuthenticatedUser = Depends(get_current_user),
|
||||
):
|
||||
"""将多个连续的同类型片段合并为一个片段"""
|
||||
@@ -230,13 +307,11 @@ def merge_draft_clips(
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST, detail=str(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 {
|
||||
"id": merged.id,
|
||||
"plan_id": merged.plan_id,
|
||||
"clip_type": merged.clip_type,
|
||||
"order": merged.order,
|
||||
"duration": merged.duration,
|
||||
"text_content": merged.text_content,
|
||||
"merged_clip": _clip_to_response(merged, asset_url=url_map.get(asset_id)),
|
||||
"deleted_clip_ids": body.clip_ids,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -227,6 +227,7 @@ class ClipsFromAssetsResponse(BaseModel):
|
||||
|
||||
success: bool = True
|
||||
created_count: int
|
||||
plan_id: str = ""
|
||||
message: str = ""
|
||||
clip_ids: List[str] = Field(default_factory=list, description="创建的片段ID列表")
|
||||
|
||||
@@ -439,17 +440,28 @@ class EditorUpdateRequest(BaseModel):
|
||||
|
||||
|
||||
class EditorClipResponse(BaseModel):
|
||||
"""片段响应"""
|
||||
"""片段响应 — 与数据库 edit_plan_clips 表字段对齐"""
|
||||
|
||||
id: str
|
||||
plan_id: str
|
||||
clip_type: str
|
||||
order: int
|
||||
duration: float
|
||||
start_time: float = 0.0
|
||||
text_content: str = ""
|
||||
transition_effect: str = "cut"
|
||||
transition_duration: float = 0.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)
|
||||
created_at: str = ""
|
||||
updated_at: str = ""
|
||||
|
||||
|
||||
class EditorClipListResponse(BaseModel):
|
||||
|
||||
@@ -101,6 +101,13 @@ class SQLAlchemyAssetRepository:
|
||||
return None
|
||||
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:
|
||||
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