Files
xiaoxia-saas/tests/unit/test_mediakit_smart_clips.py
T
xiaoxia 12f51d43bf
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 3m38s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 4m8s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 4m16s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 5m6s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 6m4s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 6m5s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 6m6s
CI/CD Pipeline / Build Staging API Image (push) Successful in 8m15s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 2m31s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 1m41s
AI Code Review / AI Code Review (pull_request) Successful in 5m15s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m24s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 4m34s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 5m55s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 10m32s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 7m27s
CI/CD Pipeline / Validate - Code Quality (push) Successful in 11m11s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 1m58s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 2m18s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 3m21s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 4m14s
CI/CD Pipeline / Integration Tests (push) Successful in 3m22s
CI/CD Pipeline / Unit Tests (push) Successful in 15m1s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 12m39s
CI/CD Pipeline / CI Gate (pull_request) Successful in 24s
CI/CD Pipeline / Canary Release to Production (pull_request) Failing after 458h40m56s
CI/CD Pipeline / Deploy Production (pull_request) Failing after 458h40m57s
CI/CD Pipeline / Build Production Worker Image (pull_request) Failing after 458h40m58s
CI/CD Pipeline / Build Production API Image (pull_request) Failing after 458h40m59s
CI/CD Pipeline / Production Browser E2E (push) Failing after 458h41m9s
CI/CD Pipeline / Build Production Web Image (push) Failing after 458h41m10s
CI/CD Pipeline / Canary Release to Production (push) Failing after 458h41m9s
CI/CD Pipeline / CI Gate (push) Failing after 458h41m10s
CI/CD Pipeline / Deploy Production (push) Failing after 458h41m10s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 458h51m3s
CI/CD Pipeline / PR Build Web Image (pull_request) Failing after 458h52m46s
CI/CD Pipeline / Build Production API Image (push) Failing after 458h41m12s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 458h53m38s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 458h51m3s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Failing after 458h54m38s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 458h51m5s
CI/CD Pipeline / PR Build Worker Image (push) Failing after 458h55m26s
CI/CD Pipeline / PR Build API Image (push) Failing after 458h55m30s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 458h53m40s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 458h55m32s
CI/CD Pipeline / Frontend Lint (push) Failing after 458h56m6s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 458h55m34s
CI/CD Pipeline / Check if frontend-only change (push) Failing after 458h59m46s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 458h55m36s
CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 459h14m58s
CI/CD Pipeline / Build Production Web Image (pull_request) Failing after 459h14m59s
CI/CD Pipeline / Build Production Worker Image (push) Failing after 459h15m11s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 459h25m30s
CI/CD Pipeline / PR Build Web Image (push) Failing after 459h29m28s
feat(from-assets): 改为异步执行MediaKit分析 (#1525)
Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
2026-08-28 12:26:00 +08:00

658 lines
26 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""测试 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
# 推荐 [10, 15],已用 [0, 10] — 边界相接不算冲突
assert _recommended_time_conflicts(10.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] 不冲突
assert _recommended_time_conflicts(15.0, 5.0, 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:
"""测试模板片段配置查询。"""
def test_returns_segments_from_new_template_system(self):
"""新模板系统(clip_configs)有数据时优先使用。"""
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.return_value = [cc2, cc1] # 乱序返回
result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
assert len(result) == 2
assert result[0] == (0, 3.0, 5.0)
assert result[1] == (1, 4.0, 8.0)
def test_falls_back_to_old_template_segments(self):
"""新模板系统无数据时回退到旧系统。"""
from app.api.routes.templates_editor.clips import _get_template_segments
mock_tpl_svc = MagicMock()
mock_tpl_svc.list_clip_configs.return_value = []
with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
mock_repo = MagicMock()
seg1 = MagicMock()
seg1.segment_order = 0
seg1.duration_min = 2.0
seg1.duration_max = 4.0
mock_repo.list_segments.return_value = [seg1]
MockRepo.return_value = mock_repo
result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
assert len(result) == 1
assert result[0] == (0, 2.0, 4.0)
def test_returns_empty_when_no_segments(self):
"""两套系统都没有片段配置时返回空列表。"""
from app.api.routes.templates_editor.clips import _get_template_segments
mock_tpl_svc = MagicMock()
mock_tpl_svc.list_clip_configs.return_value = []
with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
mock_repo = MagicMock()
mock_repo.list_segments.return_value = []
MockRepo.return_value = mock_repo
result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
assert result == []
def test_new_system_exception_falls_back(self):
"""新模板系统异常时回退到旧系统。"""
from app.api.routes.templates_editor.clips import _get_template_segments
mock_tpl_svc = MagicMock()
mock_tpl_svc.list_clip_configs.side_effect = RuntimeError("db error")
with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
mock_repo = MagicMock()
seg = MagicMock()
seg.segment_order = 0
seg.duration_min = 1.0
seg.duration_max = 3.0
mock_repo.list_segments.return_value = [seg]
MockRepo.return_value = mock_repo
result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
assert len(result) == 1
# ── from-assets 端点集成测试 ────────────────────────────────────────────────
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.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
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_segment_count(self):
"""片段数量 = segment 数量,忽略 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=4)
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 == 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)
assert 3.0 <= clips_data[0]["duration"] <= 5.0
assert 4.0 <= clips_data[1]["duration"] <= 8.0
def test_assets_round_robin_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"])
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]
assert asset_ids == ["a1", "a2", "a1", "a2"]
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_400(self):
"""模板没有 segment 配置时返回 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
mock_tpl_svc = MagicMock()
mock_tpl_svc.list_clip_configs.return_value = []
mock_plan_svc = _make_plan_svc()
with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
mock_repo = MagicMock()
mock_repo.list_segments.return_value = []
MockRepo.return_value = mock_repo
body = ClipsFromAssetsRequest(asset_ids=["a1"])
with pytest.raises(HTTPException) as exc_info:
create_clips_from_assets_editor(
template_id="tmpl-1",
body=body,
background_tasks=MagicMock(),
plan_id="plan-1",
services=(mock_tpl_svc, mock_plan_svc),
asset_repo=MagicMock(),
db=MagicMock(),
current_user=_make_auth_user(),
)
assert exc_info.value.status_code == 400
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.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