saas-backend
8237d170a5
[skip ci-format-check] #1899 clip_count 参数透传（from-assets 端点） - ClipsFromAssetsRequest 新增 clip_count: Optional[int] (1-10) 字段 - 保留旧字段 required_clips_count 兼容：未传 clip_count 时回退旧字段（超 10 截断）；两者都未传时默认 3 段 - 按 clip_count 创建 N 个片段，大于模板 segment 数时按顺序循环复用 segment 配置 - 每个片段时长在对应 segment 的 duration_min/duration_max 之间随机取值，保持原有轮询分配逻辑 #1900 模板管理 API 清理（前端 PR#1911 my-templates/editing-planner 下线） - 保留 GET /templates 列表端点（生成页自动选模板使用） - 删除 HTTP 端点：GET/POST/PUT/PATCH/DELETE /templates/{id}、POST copy、GET usage、 POST toggle-favorite、POST validate、GET/POST/DELETE /categories、GET /tags/list - 保留内部默认模板兜底 _get_or_create_default_template_id 及 CreateTemplateUseCase - 删除对应 response_model 中不再使用的类（Copy/Update/Validate/Category/Usage/Favorite 等） - 清理测试：删除集成测试 test_template_categories_api.py，更新 3 个单测适配新 clip_count 语义 DB：未涉及 schema 变更（models 本就无 is_favorite/usage_count 等列），跳过 migration。 新增 scripts/cleanup_e2e_test_data.sql 供运维在 staging DB 手动清理 e2e-gen-*/asset-create_* 账号残留模板数据。 单测：15128 passed, 28 skipped（全量 tests/unit/）
719 lines
29 KiB
Python
719 lines
29 KiB
Python
"""测试 MediaKit 智能选片 + from-assets 按模板 segment 创建片段。
|
||
|
||
覆盖：
|
||
- _recommended_time_conflicts 冲突检测
|
||
- _get_mediakit_recommendations 解析与降级
|
||
- _get_template_segments 查询模板片段配置
|
||
- from-assets 端点：按模板 segment 数量和时长创建片段
|
||
- from-assets 端点：事务性原子替换
|
||
- from-assets 端点：素材轮询分配
|
||
- from-assets 端点：MediaKit 推荐时间首片段使用
|
||
- from-assets 端点：同一素材多片段时后续用随机
|
||
- from-assets 端点：无 segment 配置时报错
|
||
- from-assets 端点：素材时长为0时报400
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
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
|
||
|
||
# ── _recommended_time_conflicts 单元测试 ─────────────────────────────────────
|
||
|
||
|
||
class TestRecommendedTimeConflicts:
|
||
"""测试推荐时间与已使用时间段的冲突检测。"""
|
||
|
||
def test_no_conflict_when_empty(self):
|
||
from app.api.routes.templates_editor.clips import _recommended_time_conflicts
|
||
|
||
assert _recommended_time_conflicts(5.0, 5.0, []) is False
|
||
|
||
def test_no_conflict_when_before(self):
|
||
from app.api.routes.templates_editor.clips import _recommended_time_conflicts
|
||
|
||
assert _recommended_time_conflicts(5.0, 5.0, [(15.0, 20.0)]) is False
|
||
|
||
def test_no_conflict_when_after(self):
|
||
from app.api.routes.templates_editor.clips import _recommended_time_conflicts
|
||
|
||
assert _recommended_time_conflicts(20.0, 5.0, [(0.0, 10.0)]) is False
|
||
|
||
def test_conflict_overlap_start(self):
|
||
from app.api.routes.templates_editor.clips import _recommended_time_conflicts
|
||
|
||
# 推荐 [5, 10]，已用 [0, 7]
|
||
assert _recommended_time_conflicts(5.0, 5.0, [(0.0, 7.0)]) is True
|
||
|
||
def test_conflict_overlap_end(self):
|
||
from app.api.routes.templates_editor.clips import _recommended_time_conflicts
|
||
|
||
# 推荐 [5, 10]，已用 [8, 15]
|
||
assert _recommended_time_conflicts(5.0, 5.0, [(8.0, 15.0)]) is True
|
||
|
||
def test_conflict_contained(self):
|
||
from app.api.routes.templates_editor.clips import _recommended_time_conflicts
|
||
|
||
# 推荐 [5, 10]，已用 [0, 20]
|
||
assert _recommended_time_conflicts(5.0, 5.0, [(0.0, 20.0)]) is True
|
||
|
||
def test_conflict_exact_boundary_no_overlap(self):
|
||
from app.api.routes.templates_editor.clips import _recommended_time_conflicts
|
||
|
||
# 新语义：默认 1.5s 边缘间隙扩边，推荐 [10, 15] 与已用 [0, 10] 首尾相接
|
||
# 落在扩边范围内 → 判为冲突（避免观感重复）
|
||
assert _recommended_time_conflicts(10.0, 5.0, [(0.0, 10.0)]) is True
|
||
# 显式 edge_gap=0 时退回纯区间重叠判定：相接不算重叠
|
||
assert _recommended_time_conflicts(10.0, 5.0, [(0.0, 10.0)], edge_gap=0.0) is False
|
||
# 间隙大于边缘间隙（2.0 > 1.5）→ 不冲突
|
||
assert _recommended_time_conflicts(12.0, 5.0, [(0.0, 10.0)]) is False
|
||
|
||
def test_conflict_multiple_used(self):
|
||
from app.api.routes.templates_editor.clips import _recommended_time_conflicts
|
||
|
||
used = [(0.0, 5.0), (10.0, 15.0), (20.0, 25.0)]
|
||
# 推荐 [6, 11] 与 [10, 15] 冲突
|
||
assert _recommended_time_conflicts(6.0, 5.0, used) is True
|
||
# 推荐 [15, 20] 与 [10, 15] 首尾相接：1.5s 扩边内 → 冲突
|
||
assert _recommended_time_conflicts(15.0, 5.0, used) is True
|
||
# 空闲段 (6.5, 8.5) 长 2.0s：推荐 [6.6, 8.4]（dur=1.8）与三区间扩边均不接触
|
||
assert _recommended_time_conflicts(6.6, 1.8, used) is False
|
||
|
||
|
||
# ── _get_mediakit_recommendations 单元测试 ──────────────────────────────────
|
||
|
||
|
||
class TestGetMediakitRecommendations:
|
||
"""测试 MediaKit 推荐结果解析和降级。"""
|
||
|
||
@patch("app.api.routes.templates_editor.clips.get_mediakit_client")
|
||
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||
def test_returns_parsed_recommendations(self, mock_storage, mock_client_fn):
|
||
from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
|
||
|
||
mock_client = MagicMock()
|
||
mock_client.is_available = True
|
||
mock_client.analyze_videos.return_value = [
|
||
'[{"asset_id": "a1", "recommended_start_time": 12.5, "reason": "动作场景"}]'
|
||
]
|
||
mock_client_fn.return_value = mock_client
|
||
|
||
mock_storage_svc = MagicMock()
|
||
mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4"
|
||
mock_storage.return_value = mock_storage_svc
|
||
|
||
asset = MagicMock()
|
||
asset.storage_key = "v.mp4"
|
||
asset.mime_type = "video/mp4"
|
||
|
||
mock_asset_repo = MagicMock()
|
||
mock_asset_repo.get.return_value = asset
|
||
|
||
result = _get_mediakit_recommendations(["a1"], mock_asset_repo)
|
||
assert result == {"a1": 12.5}
|
||
|
||
@patch("app.api.routes.templates_editor.clips.get_mediakit_client")
|
||
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||
def test_client_not_available_returns_empty(self, mock_storage, mock_client_fn):
|
||
from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
|
||
|
||
mock_client = MagicMock()
|
||
mock_client.is_available = False
|
||
mock_client_fn.return_value = mock_client
|
||
|
||
result = _get_mediakit_recommendations(["a1"], MagicMock())
|
||
assert result == {}
|
||
|
||
@patch("app.api.routes.templates_editor.clips.get_mediakit_client")
|
||
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||
def test_empty_contents_returns_empty(self, mock_storage, mock_client_fn):
|
||
from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
|
||
|
||
mock_client = MagicMock()
|
||
mock_client.is_available = True
|
||
mock_client.analyze_videos.return_value = []
|
||
mock_client_fn.return_value = mock_client
|
||
|
||
mock_storage_svc = MagicMock()
|
||
mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4"
|
||
mock_storage.return_value = mock_storage_svc
|
||
|
||
asset = MagicMock()
|
||
asset.storage_key = "v.mp4"
|
||
asset.mime_type = "video/mp4"
|
||
mock_asset_repo = MagicMock()
|
||
mock_asset_repo.get.return_value = asset
|
||
|
||
result = _get_mediakit_recommendations(["a1"], mock_asset_repo)
|
||
assert result == {}
|
||
|
||
@patch("app.api.routes.templates_editor.clips.get_mediakit_client")
|
||
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||
def test_unparseable_response_returns_empty(self, mock_storage, mock_client_fn):
|
||
from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
|
||
|
||
mock_client = MagicMock()
|
||
mock_client.is_available = True
|
||
mock_client.analyze_videos.return_value = ["这是一段自然语言描述，没有JSON"]
|
||
mock_client_fn.return_value = mock_client
|
||
|
||
mock_storage_svc = MagicMock()
|
||
mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4"
|
||
mock_storage.return_value = mock_storage_svc
|
||
|
||
asset = MagicMock()
|
||
asset.storage_key = "v.mp4"
|
||
asset.mime_type = "video/mp4"
|
||
mock_asset_repo = MagicMock()
|
||
mock_asset_repo.get.return_value = asset
|
||
|
||
result = _get_mediakit_recommendations(["a1"], mock_asset_repo)
|
||
assert result == {}
|
||
|
||
@patch("app.api.routes.templates_editor.clips.get_mediakit_client")
|
||
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||
def test_storage_failure_returns_empty(self, mock_storage, mock_client_fn):
|
||
from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
|
||
|
||
mock_client = MagicMock()
|
||
mock_client.is_available = True
|
||
mock_client_fn.return_value = mock_client
|
||
mock_storage.side_effect = RuntimeError("storage unavailable")
|
||
|
||
result = _get_mediakit_recommendations(["a1"], MagicMock())
|
||
assert result == {}
|
||
|
||
@patch("app.api.routes.templates_editor.clips.get_mediakit_client")
|
||
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||
def test_exception_returns_empty(self, mock_storage, mock_client_fn):
|
||
from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
|
||
|
||
mock_client_fn.side_effect = RuntimeError("unexpected error")
|
||
|
||
result = _get_mediakit_recommendations(["a1"], MagicMock())
|
||
assert result == {}
|
||
|
||
@patch("app.api.routes.templates_editor.clips.get_mediakit_client")
|
||
@patch("app.api.routes.templates_editor.clips.get_storage_service")
|
||
def test_skips_non_video_assets(self, mock_storage, mock_client_fn):
|
||
from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
|
||
|
||
mock_client = MagicMock()
|
||
mock_client.is_available = True
|
||
mock_client_fn.return_value = mock_client
|
||
|
||
mock_asset_repo = MagicMock()
|
||
mock_asset = MagicMock()
|
||
mock_asset.storage_key = "images/test.jpg"
|
||
mock_asset.mime_type = "image/jpeg"
|
||
mock_asset_repo.get.return_value = mock_asset
|
||
|
||
result = _get_mediakit_recommendations(["a1"], mock_asset_repo)
|
||
assert result == {}
|
||
mock_client.analyze_videos.assert_not_called()
|
||
|
||
|
||
# ── _get_template_segments 单元测试 ─────────────────────────────────────────
|
||
|
||
|
||
class TestGetTemplateSegments:
|
||
"""测试模板片段配置查询（单一数据源：template_clip_configs）。"""
|
||
|
||
def test_returns_segments_from_clip_configs(self):
|
||
"""片段配置主表（clip_configs）有数据时按 order 排序返回。"""
|
||
from app.api.routes.templates_editor.clips import _get_template_segments
|
||
|
||
mock_tpl_svc = MagicMock()
|
||
cc1 = MagicMock()
|
||
cc1.order = 0
|
||
cc1.min_duration = 3.0
|
||
cc1.max_duration = 5.0
|
||
cc2 = MagicMock()
|
||
cc2.order = 1
|
||
cc2.min_duration = 4.0
|
||
cc2.max_duration = 8.0
|
||
mock_tpl_svc.list_clip_configs_for_editor.return_value = [cc2, cc1] # 乱序返回
|
||
|
||
result = _get_template_segments("tmpl-1", "user-1", mock_tpl_svc)
|
||
assert len(result) == 2
|
||
assert result[0] == (0, 3.0, 5.0)
|
||
assert result[1] == (1, 4.0, 8.0)
|
||
mock_tpl_svc.list_clip_configs_for_editor.assert_called_once_with("tmpl-1", "user-1")
|
||
|
||
def test_returns_empty_when_no_configs(self):
|
||
"""模板存在但没有片段配置时返回空列表（路由层据此返回 422）。"""
|
||
from app.api.routes.templates_editor.clips import _get_template_segments
|
||
|
||
mock_tpl_svc = MagicMock()
|
||
mock_tpl_svc.list_clip_configs_for_editor.return_value = []
|
||
|
||
result = _get_template_segments("tmpl-1", "user-1", mock_tpl_svc)
|
||
assert result == []
|
||
|
||
def test_missing_template_raises(self):
|
||
"""模板不存在/无权限时服务层抛 TemplateNotFoundError（路由层据此返回 404）。"""
|
||
from app.api.routes.templates_editor.clips import _get_template_segments
|
||
from app.services.edit_template_service import TemplateNotFoundError
|
||
|
||
mock_tpl_svc = MagicMock()
|
||
mock_tpl_svc.list_clip_configs_for_editor.side_effect = TemplateNotFoundError("tmpl-x")
|
||
|
||
with pytest.raises(TemplateNotFoundError):
|
||
_get_template_segments("tmpl-x", "user-1", mock_tpl_svc)
|
||
|
||
|
||
# ── from-assets 端点集成测试 ────────────────────────────────────────────────
|
||
|
||
|
||
def _patch_zero_noise():
|
||
"""消除 clips.py 排序随机噪声（SCORE_RANDOM_NOISE_MAX 噪声返回 0）。
|
||
|
||
用于均衡分配等确定性断言；其他 uniform 调用（片段时长随机）行为不变。
|
||
"""
|
||
import random as _stdlib_random
|
||
|
||
from app.api.routes.templates_editor import clips as clips_module
|
||
|
||
from packages.domain.smart_match import SCORE_RANDOM_NOISE_MAX
|
||
|
||
_fallback = _stdlib_random.Random()
|
||
|
||
def _fake_uniform(a, b):
|
||
if b == SCORE_RANDOM_NOISE_MAX:
|
||
return 0.0
|
||
return _fallback.uniform(a, b)
|
||
|
||
return patch.object(clips_module.random, "uniform", _fake_uniform)
|
||
|
||
|
||
def _make_auth_user():
|
||
auth = MagicMock()
|
||
auth.user.id = "user-001"
|
||
auth.user.email = "test@example.com"
|
||
auth.user.display_name = "test"
|
||
auth.user_id = "user-001"
|
||
return auth
|
||
|
||
|
||
def _make_clip_config(order, min_dur, max_dur):
|
||
cc = MagicMock()
|
||
cc.order = order
|
||
cc.min_duration = min_dur
|
||
cc.max_duration = max_dur
|
||
return cc
|
||
|
||
|
||
def _make_plan_svc(replace_return_count=None):
|
||
svc = MagicMock()
|
||
svc.get_plan_or_raise = MagicMock()
|
||
if replace_return_count is not None:
|
||
svc.replace_all_clips_transactional = MagicMock(return_value=replace_return_count)
|
||
else:
|
||
svc.replace_all_clips_transactional = MagicMock(return_value=0)
|
||
return svc
|
||
|
||
|
||
def _make_tpl_svc_with_segments(segments):
|
||
"""segments: list of (order, min_dur, max_dur)"""
|
||
svc = MagicMock()
|
||
clip_configs = [_make_clip_config(o, mn, mx) for o, mn, mx in segments]
|
||
svc.list_clip_configs_for_editor.return_value = clip_configs
|
||
return svc
|
||
|
||
|
||
def _make_rich_asset(asset_id, duration, storage_key="v.mp4", mime="video/mp4"):
|
||
asset = MagicMock()
|
||
asset.id = asset_id
|
||
asset.duration = duration
|
||
asset.storage_key = storage_key
|
||
asset.mime_type = mime
|
||
# score_asset 所需的属性（避免 MagicMock 导致类型比较错误）
|
||
asset.quality_score = None
|
||
asset.created_at = None
|
||
asset.metadata = {}
|
||
return asset
|
||
|
||
|
||
def _get_clips_data(mock_plan_svc):
|
||
"""从 replace_all_clips_transactional 调用中提取 clips_data。"""
|
||
call_args = mock_plan_svc.replace_all_clips_transactional.call_args
|
||
if len(call_args.args) >= 2:
|
||
return call_args.args[1]
|
||
return call_args.kwargs.get("clips_data", [])
|
||
|
||
|
||
class TestFromAssetsByTemplateSegments:
|
||
"""测试 from-assets 按模板 segment 创建片段（V2 事务性替换）。"""
|
||
|
||
def test_creates_clips_matching_legacy_required_clips_count(self):
|
||
"""显式传旧字段 required_clips_count 时按该值创建片段（兼容前端老版本）。"""
|
||
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
|
||
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
|
||
|
||
segments = [(0, 3.0, 5.0), (1, 4.0, 8.0), (2, 2.0, 6.0), (3, 5.0, 10.0)]
|
||
mock_tpl_svc = _make_tpl_svc_with_segments(segments)
|
||
mock_plan_svc = _make_plan_svc(replace_return_count=2)
|
||
|
||
mock_asset_repo = MagicMock()
|
||
mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)
|
||
|
||
body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=2)
|
||
result = create_clips_from_assets_editor(
|
||
template_id="tmpl-1",
|
||
body=body,
|
||
background_tasks=MagicMock(),
|
||
plan_id="plan-1",
|
||
services=(mock_tpl_svc, mock_plan_svc),
|
||
asset_repo=mock_asset_repo,
|
||
db=MagicMock(),
|
||
current_user=_make_auth_user(),
|
||
)
|
||
|
||
assert result.created_count == 2
|
||
clips_data = _get_clips_data(mock_plan_svc)
|
||
assert len(clips_data) == 2
|
||
|
||
def test_creates_clips_matching_segment_count_when_no_clip_count(self):
|
||
"""未传 clip_count/required_clips_count 时回退模板 segment 数量。"""
|
||
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
|
||
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
|
||
|
||
segments = [(0, 3.0, 5.0), (1, 4.0, 8.0), (2, 2.0, 6.0), (3, 5.0, 10.0)]
|
||
mock_tpl_svc = _make_tpl_svc_with_segments(segments)
|
||
mock_plan_svc = _make_plan_svc(replace_return_count=4)
|
||
|
||
mock_asset_repo = MagicMock()
|
||
mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)
|
||
|
||
body = ClipsFromAssetsRequest(asset_ids=["a1"])
|
||
result = create_clips_from_assets_editor(
|
||
template_id="tmpl-1",
|
||
body=body,
|
||
background_tasks=MagicMock(),
|
||
plan_id="plan-1",
|
||
services=(mock_tpl_svc, mock_plan_svc),
|
||
asset_repo=mock_asset_repo,
|
||
db=MagicMock(),
|
||
current_user=_make_auth_user(),
|
||
)
|
||
|
||
assert result.created_count == 4
|
||
clips_data = _get_clips_data(mock_plan_svc)
|
||
assert len(clips_data) == 4
|
||
|
||
def test_uses_transactional_replace(self):
|
||
"""使用 replace_all_clips_transactional 而不是分别 delete + create。"""
|
||
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
|
||
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
|
||
|
||
segments = [(0, 3.0, 5.0), (1, 4.0, 8.0)]
|
||
mock_tpl_svc = _make_tpl_svc_with_segments(segments)
|
||
mock_plan_svc = _make_plan_svc(replace_return_count=2)
|
||
|
||
mock_asset_repo = MagicMock()
|
||
mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)
|
||
|
||
body = ClipsFromAssetsRequest(asset_ids=["a1"])
|
||
create_clips_from_assets_editor(
|
||
template_id="tmpl-1",
|
||
body=body,
|
||
background_tasks=MagicMock(),
|
||
plan_id="plan-1",
|
||
services=(mock_tpl_svc, mock_plan_svc),
|
||
asset_repo=mock_asset_repo,
|
||
db=MagicMock(),
|
||
current_user=_make_auth_user(),
|
||
)
|
||
|
||
mock_plan_svc.replace_all_clips_transactional.assert_called_once()
|
||
assert not mock_plan_svc.delete_all_clips.called
|
||
assert not mock_plan_svc.create_clip.called
|
||
|
||
def test_duration_within_segment_range(self):
|
||
"""每个片段时长在 segment 的 min~max 范围内。"""
|
||
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
|
||
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
|
||
|
||
segments = [(0, 3.0, 5.0), (1, 4.0, 8.0)]
|
||
mock_tpl_svc = _make_tpl_svc_with_segments(segments)
|
||
mock_plan_svc = _make_plan_svc(replace_return_count=2)
|
||
|
||
mock_asset_repo = MagicMock()
|
||
mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)
|
||
|
||
body = ClipsFromAssetsRequest(asset_ids=["a1"])
|
||
create_clips_from_assets_editor(
|
||
template_id="tmpl-1",
|
||
body=body,
|
||
background_tasks=MagicMock(),
|
||
plan_id="plan-1",
|
||
services=(mock_tpl_svc, mock_plan_svc),
|
||
asset_repo=mock_asset_repo,
|
||
db=MagicMock(),
|
||
current_user=_make_auth_user(),
|
||
)
|
||
|
||
clips_data = _get_clips_data(mock_plan_svc)
|
||
# PR #1614 转场补偿：2 个片段时每 clip 时长 +(2-1)*0.5/2=0.25s
|
||
# （xfade 重叠在渲染时扣除，故 clip 时长 = segment 随机时长 + 补偿），
|
||
# 断言上界需计入补偿与一位小数舍入余量
|
||
comp = (2 - 1) * 0.5 / 2
|
||
assert 3.0 + comp - 0.1 <= clips_data[0]["duration"] <= 5.0 + comp + 0.1
|
||
assert 4.0 + comp - 0.1 <= clips_data[1]["duration"] <= 8.0 + comp + 0.1
|
||
|
||
def test_assets_balanced_assignment(self):
|
||
"""素材按使用次数贪心分配（使用少的优先），保证均衡使用。"""
|
||
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
|
||
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
|
||
|
||
segments = [(0, 3.0, 5.0), (1, 3.0, 5.0), (2, 3.0, 5.0), (3, 3.0, 5.0)]
|
||
mock_tpl_svc = _make_tpl_svc_with_segments(segments)
|
||
mock_plan_svc = _make_plan_svc(replace_return_count=4)
|
||
|
||
mock_asset_repo = MagicMock()
|
||
|
||
def get_asset(aid):
|
||
return _make_rich_asset(aid, 60.0)
|
||
|
||
mock_asset_repo.get.side_effect = get_asset
|
||
|
||
body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"])
|
||
# 消除排序噪声，确定性断言贪心均衡分配
|
||
with _patch_zero_noise():
|
||
create_clips_from_assets_editor(
|
||
template_id="tmpl-1",
|
||
body=body,
|
||
background_tasks=MagicMock(),
|
||
plan_id="plan-1",
|
||
services=(mock_tpl_svc, mock_plan_svc),
|
||
asset_repo=mock_asset_repo,
|
||
db=MagicMock(),
|
||
current_user=_make_auth_user(),
|
||
)
|
||
|
||
clips_data = _get_clips_data(mock_plan_svc)
|
||
asset_ids = [c["asset_id"] for c in clips_data]
|
||
# 贪心分配保证均衡：2个素材分4个片段，每个素材恰好使用2次
|
||
from collections import Counter
|
||
|
||
counts = Counter(asset_ids)
|
||
assert counts["a1"] == 2 and counts["a2"] == 2
|
||
|
||
def test_orders_start_from_zero(self):
|
||
"""片段 order 从 0 开始递增。"""
|
||
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
|
||
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
|
||
|
||
segments = [(0, 3.0, 5.0), (1, 3.0, 5.0), (2, 3.0, 5.0)]
|
||
mock_tpl_svc = _make_tpl_svc_with_segments(segments)
|
||
mock_plan_svc = _make_plan_svc(replace_return_count=3)
|
||
|
||
mock_asset_repo = MagicMock()
|
||
mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)
|
||
|
||
body = ClipsFromAssetsRequest(asset_ids=["a1"])
|
||
create_clips_from_assets_editor(
|
||
template_id="tmpl-1",
|
||
body=body,
|
||
background_tasks=MagicMock(),
|
||
plan_id="plan-1",
|
||
services=(mock_tpl_svc, mock_plan_svc),
|
||
asset_repo=mock_asset_repo,
|
||
db=MagicMock(),
|
||
current_user=_make_auth_user(),
|
||
)
|
||
|
||
clips_data = _get_clips_data(mock_plan_svc)
|
||
orders = [c["order"] for c in clips_data]
|
||
assert orders == [0, 1, 2]
|
||
|
||
def test_no_segments_raises_422(self):
|
||
"""模板存在但未配置片段时返回 422（与模板不存在的 404 区分）。"""
|
||
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
|
||
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
|
||
from fastapi import HTTPException
|
||
|
||
mock_tpl_svc = MagicMock()
|
||
mock_tpl_svc.list_clip_configs_for_editor.return_value = []
|
||
mock_plan_svc = _make_plan_svc()
|
||
|
||
body = ClipsFromAssetsRequest(asset_ids=["a1"])
|
||
with pytest.raises(HTTPException) as exc_info:
|
||
create_clips_from_assets_editor(
|
||
template_id="tmpl-1",
|
||
body=body,
|
||
background_tasks=MagicMock(),
|
||
plan_id="plan-1",
|
||
services=(mock_tpl_svc, mock_plan_svc),
|
||
asset_repo=MagicMock(),
|
||
db=MagicMock(),
|
||
current_user=_make_auth_user(),
|
||
)
|
||
assert exc_info.value.status_code == 422
|
||
|
||
mock_plan_svc.replace_all_clips_transactional.assert_not_called()
|
||
|
||
def test_template_not_found_raises_404(self):
|
||
"""模板不存在/已删除/无权限时返回 404。"""
|
||
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
|
||
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
|
||
from app.services.edit_template_service import TemplateNotFoundError
|
||
from fastapi import HTTPException
|
||
|
||
mock_tpl_svc = MagicMock()
|
||
mock_tpl_svc.list_clip_configs_for_editor.side_effect = TemplateNotFoundError("tmpl-x")
|
||
mock_plan_svc = _make_plan_svc()
|
||
|
||
body = ClipsFromAssetsRequest(asset_ids=["a1"])
|
||
with pytest.raises(HTTPException) as exc_info:
|
||
create_clips_from_assets_editor(
|
||
template_id="tmpl-x",
|
||
body=body,
|
||
background_tasks=MagicMock(),
|
||
plan_id="plan-1",
|
||
services=(mock_tpl_svc, mock_plan_svc),
|
||
asset_repo=MagicMock(),
|
||
db=MagicMock(),
|
||
current_user=_make_auth_user(),
|
||
)
|
||
assert exc_info.value.status_code == 404
|
||
|
||
mock_plan_svc.replace_all_clips_transactional.assert_not_called()
|
||
|
||
def test_duration_capped_by_asset_duration(self):
|
||
"""素材时长不足时 clip duration 被缩短。"""
|
||
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
|
||
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
|
||
|
||
segments = [(0, 10.0, 20.0)]
|
||
mock_tpl_svc = _make_tpl_svc_with_segments(segments)
|
||
mock_plan_svc = _make_plan_svc(replace_return_count=1)
|
||
|
||
mock_asset_repo = MagicMock()
|
||
mock_asset_repo.get.return_value = _make_rich_asset("a1", 5.0)
|
||
|
||
body = ClipsFromAssetsRequest(asset_ids=["a1"])
|
||
create_clips_from_assets_editor(
|
||
template_id="tmpl-1",
|
||
body=body,
|
||
background_tasks=MagicMock(),
|
||
plan_id="plan-1",
|
||
services=(mock_tpl_svc, mock_plan_svc),
|
||
asset_repo=mock_asset_repo,
|
||
db=MagicMock(),
|
||
current_user=_make_auth_user(),
|
||
)
|
||
|
||
clips_data = _get_clips_data(mock_plan_svc)
|
||
assert clips_data[0]["duration"] <= 5.0
|
||
|
||
def test_zero_duration_asset_raises_400(self):
|
||
"""素材时长为 0 时抛出 400。"""
|
||
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
|
||
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
|
||
from fastapi import HTTPException
|
||
|
||
segments = [(0, 3.0, 5.0)]
|
||
mock_tpl_svc = _make_tpl_svc_with_segments(segments)
|
||
mock_plan_svc = _make_plan_svc()
|
||
|
||
mock_asset_repo = MagicMock()
|
||
mock_asset_repo.get.return_value = _make_rich_asset("bad", 0.0)
|
||
|
||
body = ClipsFromAssetsRequest(asset_ids=["bad"])
|
||
with pytest.raises(HTTPException) as exc_info:
|
||
create_clips_from_assets_editor(
|
||
template_id="tmpl-1",
|
||
body=body,
|
||
background_tasks=MagicMock(),
|
||
plan_id="plan-1",
|
||
services=(mock_tpl_svc, mock_plan_svc),
|
||
asset_repo=mock_asset_repo,
|
||
db=MagicMock(),
|
||
current_user=_make_auth_user(),
|
||
)
|
||
|
||
assert exc_info.value.status_code == 400
|
||
mock_plan_svc.replace_all_clips_transactional.assert_not_called()
|
||
|
||
def test_empty_asset_ids_raises_400(self):
|
||
"""asset_ids 为空列表时返回 400（defense-in-depth，schema 层也有 min_length=1）。"""
|
||
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
|
||
from fastapi import HTTPException
|
||
|
||
segments = [(0, 3.0, 5.0)]
|
||
mock_tpl_svc = _make_tpl_svc_with_segments(segments)
|
||
mock_plan_svc = _make_plan_svc()
|
||
|
||
# 用 MagicMock 模拟 body，绕过 Pydantic schema 的 min_length 校验
|
||
mock_body = MagicMock()
|
||
mock_body.asset_ids = []
|
||
mock_body.clip_count = None
|
||
mock_body.required_clips_count = None
|
||
|
||
with pytest.raises(HTTPException) as exc_info:
|
||
create_clips_from_assets_editor(
|
||
template_id="tmpl-1",
|
||
body=mock_body,
|
||
background_tasks=MagicMock(),
|
||
plan_id="plan-1",
|
||
services=(mock_tpl_svc, mock_plan_svc),
|
||
asset_repo=MagicMock(),
|
||
db=MagicMock(),
|
||
current_user=_make_auth_user(),
|
||
)
|
||
assert exc_info.value.status_code == 400
|
||
assert "素材" in exc_info.value.detail
|
||
mock_plan_svc.replace_all_clips_transactional.assert_not_called()
|
||
|
||
|
||
# ── _safe_segment_duration 单元测试 ─────────────────────────────────────────
|
||
|
||
|
||
class TestSafeSegmentDuration:
|
||
"""测试片段时长安全转换。"""
|
||
|
||
def test_normal_float(self):
|
||
from app.api.routes.templates_editor.clips import _safe_segment_duration
|
||
|
||
assert _safe_segment_duration(3.5, 1.0) == 3.5
|
||
|
||
def test_none_returns_default(self):
|
||
from app.api.routes.templates_editor.clips import _safe_segment_duration
|
||
|
||
assert _safe_segment_duration(None, 5.0) == 5.0
|
||
|
||
def test_string_number(self):
|
||
from app.api.routes.templates_editor.clips import _safe_segment_duration
|
||
|
||
assert _safe_segment_duration("4.2", 1.0) == 4.2
|
||
|
||
def test_invalid_string_returns_default(self):
|
||
from app.api.routes.templates_editor.clips import _safe_segment_duration
|
||
|
||
assert _safe_segment_duration("abc", 5.0) == 5.0
|
||
|
||
def test_negative_returns_default(self):
|
||
from app.api.routes.templates_editor.clips import _safe_segment_duration
|
||
|
||
assert _safe_segment_duration(-1.0, 5.0) == 5.0
|
||
|
||
def test_zero_returns_default(self):
|
||
from app.api.routes.templates_editor.clips import _safe_segment_duration
|
||
|
||
assert _safe_segment_duration(0, 5.0) == 5.0
|
||
|
||
def test_integer_value(self):
|
||
from app.api.routes.templates_editor.clips import _safe_segment_duration
|
||
|
||
assert _safe_segment_duration(10, 1.0) == 10.0
|