"""测试编辑器 from-assets 端点：按模板segment创建片段 + 事务性替换 + 随机起始. 覆盖： - 片段数量 = segment 数量（required_clips_count 被忽略） - 素材不足时同一素材轮询切多个片段 - 随机 start_time + used_segments 去重 - 素材时长不足时 clip duration 缩短 - 素材时长全部为 0/缺失时抛 400「素材可切区间不足」；混合池中零时长素材被跳过 - 使用 replace_all_clips_transactional 原子性替换 - order 从 0 开始 - start_time=None 时抛出 400 - mark_clips_ready 在事务方法内部完成 """ 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 from fastapi import HTTPException TEST_PLAN_ID = "plan-draft-001" TEST_USER_ID = "user-001" # 默认测试用 segments：4 个片段，每个 3~5 秒 DEFAULT_SEGMENTS = [(0, 3.0, 5.0), (1, 3.0, 5.0), (2, 3.0, 5.0), (3, 3.0, 5.0)] def _segments(count: int, dur_min: float = 3.0, dur_max: float = 5.0): return [(i, dur_min, dur_max) for i in range(count)] def _patch_zero_noise(): """消除 clips.py 排序随机噪声，用于确定性断言（如均衡分配）。 排序噪声（random.uniform(0, SCORE_RANDOM_NOISE_MAX)）返回 0； 其他 uniform 调用（片段时长随机）委托给独立 Random 实例，行为不变。 """ 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 _patch_segments(segments=None): return patch( "app.api.routes.templates_editor.clips._get_template_segments", return_value=segments if segments is not None else DEFAULT_SEGMENTS, ) def _make_auth_user(): auth = MagicMock() auth.user.id = TEST_USER_ID auth.user.email = "test@example.com" auth.user.display_name = "测试用户" auth.user_id = TEST_USER_ID return auth def _make_mock_asset(asset_id, duration): asset = MagicMock() asset.id = asset_id asset.duration = duration # score_asset 所需的属性（避免 MagicMock 导致类型比较错误） asset.quality_score = None asset.created_at = None asset.metadata = {} return asset def _make_plan_svc(replace_return_count=None): svc = MagicMock() # replace_all_clips_transactional 返回创建的片段数量 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 _get_clips_data_from_call(mock_plan_svc): """从 replace_all_clips_transactional 的调用中获取 clips_data。""" assert mock_plan_svc.replace_all_clips_transactional.called, "replace_all_clips_transactional 未被调用" call_args = mock_plan_svc.replace_all_clips_transactional.call_args # call_args = ((plan_id, clips_data), kwargs) if len(call_args.args) >= 2: return call_args.args[1] return call_args.kwargs.get("clips_data", []) @pytest.fixture(autouse=True) def _mock_segment_tracker(): """from-assets 现在会读/写素材 metadata 的片段区间记录，测试中 mock 掉避免依赖真实 DB。 get_used_segments 返回空 dict（等价历史行为：无历史区间）； record/remove/reset 回调均无副作用。 """ with ( patch( "app.api.routes.templates_editor.clips.get_used_segments", return_value={}, ), patch( "app.api.routes.templates_editor.clips.record_used_segments", return_value=None, ), patch( "app.api.routes.templates_editor.clips.make_reuse_callback", return_value=lambda asset_id, clip_duration: None, ), patch( "app.api.routes.templates_editor.clips.remove_used_segment", return_value=False, ), ): yield class TestEditorClipsBySegments: """测试按 segment 数量创建片段 + 素材轮询。""" @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_creates_clips_matching_segment_count(self, mock_storage): """4 个 segment 即使只有2个素材也创建4个片段，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 mock_plan_svc = _make_plan_svc(replace_return_count=4) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, {"a1": 30.0, "a2": 20.0}[aid])) body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"], required_clips_count=2) # 均衡分配由 use_count 贪心保证，消除排序噪声后确定性断言 with _patch_zero_noise(), _patch_segments(DEFAULT_SEGMENTS): result = create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), 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_from_call(mock_plan_svc) assert len(clips_data) == 4 # 验证均衡分配（贪心策略保证）：2个素材分4个片段，每个素材恰好使用2次 from collections import Counter asset_ids = [c["asset_id"] for c in clips_data] counts = Counter(asset_ids) assert counts["a1"] == 2 and counts["a2"] == 2 @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_orders_start_at_zero(self, mock_storage): """片段 order 从 0 开始递增。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc(replace_return_count=3) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3) with _patch_segments(_segments(3)): create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) clips_data = _get_clips_data_from_call(mock_plan_svc) assert clips_data[0]["order"] == 0 assert clips_data[1]["order"] == 1 assert clips_data[2]["order"] == 2 @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_uses_transactional_replace(self, mock_storage): """使用 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 mock_plan_svc = _make_plan_svc(replace_return_count=2) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) body = ClipsFromAssetsRequest(asset_ids=["a1"]) with _patch_segments(_segments(2)): create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), 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() # 不应调用单独的 delete 或 create assert not hasattr(mock_plan_svc, "create_clip") or not mock_plan_svc.create_clip.called @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_no_segments_raises_400(self, mock_storage): """模板没有 segment 配置时返回 400。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc() mock_asset_repo = MagicMock() body = ClipsFromAssetsRequest(asset_ids=["a1"]) with _patch_segments([]): with pytest.raises(HTTPException) as exc_info: create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, 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() class TestEditorClipsDurationAndStartTime: """测试素材时长获取、clip duration 缩短、start_time 传入。""" @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_clip_duration_shortened_for_short_assets(self, mock_storage): """素材只有 3s 时 clip duration 缩短到不超过 3.0。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc(replace_return_count=1) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("short", 3.0)) body = ClipsFromAssetsRequest(asset_ids=["short"]) with _patch_segments(_segments(1, dur_min=5.0, dur_max=10.0)): create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) clips_data = _get_clips_data_from_call(mock_plan_svc) assert clips_data[0]["duration"] <= 3.0 @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_zero_duration_asset_raises_400(self, mock_storage): """所有素材时长均为 0 时轮询无可用素材，抛出 400「素材可切区间不足」。 新轮询逻辑下零时长素材被跳过（而非立即报错）；全部素材都被跳过时 返回 400，不创建无效片段。 """ from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc() mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("bad", 0.0)) body = ClipsFromAssetsRequest(asset_ids=["bad"]) with _patch_segments(_segments(1)): with pytest.raises(HTTPException) as exc_info: create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) assert exc_info.value.status_code == 400 assert "素材可切区间不足" in exc_info.value.detail @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_zero_duration_asset_skipped_in_mixed_pool(self, mock_storage): """素材池混合零时长与正常素材时，零时长素材被跳过、正常素材承担片段。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc(replace_return_count=2) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock( side_effect=lambda aid: { "zero": _make_mock_asset("zero", 0.0), "good": _make_mock_asset("good", 30.0), }[aid] ) body = ClipsFromAssetsRequest(asset_ids=["zero", "good"], required_clips_count=2) with ( _patch_segments(_segments(2)), patch( "app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=[5.0, 12.0], ), ): create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) clips_data = _get_clips_data_from_call(mock_plan_svc) assert len(clips_data) == 2 # 所有片段都分配给正常素材，零时长素材被跳过 assert all(c["asset_id"] == "good" for c in clips_data) @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_missing_duration_asset_raises_400(self, mock_storage): """素材时长缺失（asset_repo.get 返回 None）时抛出 400。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc() mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(return_value=None) body = ClipsFromAssetsRequest(asset_ids=["missing"]) with _patch_segments(_segments(1)): with pytest.raises(HTTPException) as exc_info: create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) assert exc_info.value.status_code == 400 @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_start_time_passed_to_create(self, mock_storage): """_calc_random_start_time 返回值被传入 clips_data。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc(replace_return_count=2) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, 30.0)) body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"], required_clips_count=2) with ( _patch_segments(_segments(2)), patch( "app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=[12.5, 18.0], ) as mock_calc, ): create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) assert mock_calc.call_count == 2 clips_data = _get_clips_data_from_call(mock_plan_svc) # clips_data 按 order 排序，但分配顺序因 shuffle 而随机， # 因此只验证两个 start_time 值都存在 start_times = {c["start_time"] for c in clips_data} assert start_times == {12.5, 18.0} # 验证 order 仍然有序 orders = [c["order"] for c in clips_data] assert orders == sorted(orders) @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_asset_durations_deduped(self, mock_storage): """asset_ids 有重复时只查询一次素材时长。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc(replace_return_count=3) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) body = ClipsFromAssetsRequest(asset_ids=["a1", "a1", "a1"]) with _patch_segments(_segments(3)): create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) # 去重后只调用 1 次获取素材时长 assert mock_asset_repo.get.call_count == 1 @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_used_segments_maintained_across_clips(self, mock_storage): """同一素材切多个片段时，used_segments 应被维护。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc(replace_return_count=3) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3) captured_used_segments = [] def fake_calc(asset_id, clip_duration, asset_durations, used_segments, on_exhausted=None): captured_used_segments.append({aid: list(segs) for aid, segs in (used_segments or {}).items()}) return (len(captured_used_segments) - 1) * 5.0 with ( _patch_segments(_segments(3)), patch( "app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=fake_calc, ), ): create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) # 第一次没有已使用时间段 assert captured_used_segments[0] == {} # 第二次有第一次的记录 assert len(captured_used_segments[1]["a1"]) == 1 # 第三次有前两次的记录 assert len(captured_used_segments[2]["a1"]) == 2 class TestEditorClipsErrorHandling: """测试异常处理。""" @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_none_start_time_raises_400(self, mock_storage): """所有素材 calc 均返回 None（区间耗尽且复用被拒）→ 轮询失败抛 400。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc() mock_asset_repo = MagicMock() # 素材有 duration 但 calc 返回 None（模拟可用区间耗尽、复用被闸门拒绝） mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) body = ClipsFromAssetsRequest(asset_ids=["a1"]) with ( _patch_segments(_segments(1)), patch( "app.api.routes.templates_editor.clips._calc_random_start_time", return_value=None, ), ): with pytest.raises(HTTPException) as exc_info: create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) assert exc_info.value.status_code == 400 assert "素材可切区间不足" in exc_info.value.detail # 复用被拒导致无起点时，不应创建任何片段 assert not mock_plan_svc.replace_all_clips_transactional.called @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_transactional_replace_exception_propagates(self, mock_storage): """replace_all_clips_transactional 抛异常时应向上传播（事务已回滚）。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc() mock_plan_svc.replace_all_clips_transactional = MagicMock(side_effect=RuntimeError("DB connection lost")) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) body = ClipsFromAssetsRequest(asset_ids=["a1"]) with _patch_segments(_segments(1)): with pytest.raises(RuntimeError, match="DB connection lost"): create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) class TestReuseRatioGate: """素材区间耗尽后的受控复用与 15% 占比闸门（路由级）。""" @staticmethod def _make_calc_with_reuse(normal_starts, reused_durations): """构造模拟「区间耗尽后受控复用」的 _calc_random_start_time。 normal_starts: list[float | None]，前 N 次调用返回的空闲起点； 返回 None 表示随机找不到空闲 → 触发 on_exhausted 复用回调。 回调被调用时返回复用区间（固定 0.0 起点），复用片段时长由路由累加到 reused_durations；回调内部占比预判超 15% 时返回 None（calc 随之 None）。 """ calls = {"i": 0} def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None): i = calls["i"] calls["i"] += 1 if i < len(normal_starts) and normal_starts[i] is not None: return normal_starts[i] # 空闲耗尽 → 走受控复用回调（回调返回 (start, end) 元组，calc 取起点） if on_exhausted is not None: result = on_exhausted(asset_id, clip_duration) return result[0] if result else None return None return fake_calc, calls @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_reused_clip_ratio_within_threshold(self, mock_storage): """素材 60s、片段 5s：前 12 个用空闲区间，第 13 个复用， 复用占比 5/(12*5+5)=7.7% ≤ 15%，正常创建 13 个片段。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc(replace_return_count=13) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 60.0)) body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=13) reused: dict = {} def reuse_cb(aid, dur): # 模拟真实回调：返回复用区间前记录复用时长 reused[aid] = reused.get(aid, 0.0) + dur return (0.0, dur) # 前 12 次分配空闲起点；第 13 次 calc 直接走回调（normal_starts 越界 → None → 回调） normal_starts = [float(i * 5) for i in range(12)] fake_calc, _ = self._make_calc_with_reuse(normal_starts, reused) with ( _patch_segments(_segments(13, dur_min=5.0, dur_max=5.0)), patch("app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=fake_calc), patch("app.api.routes.templates_editor.clips.make_reuse_callback", return_value=reuse_cb), ): create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) clips_data = _get_clips_data_from_call(mock_plan_svc) assert len(clips_data) == 13 # 1 个复用片段，占比 1/13 ≈ 7.7% ≤ 15% # 转场补偿: raw_duration = 5.0 + (13-1)*0.5/13 ≈ 5.462 → round(5.462,1) = 5.5 assert abs(reused.get("a1", 0.0) - 5.5) < 0.1 @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_reuse_ratio_exceeded_returns_400(self, mock_storage): """复用占比将超 15% 时回调拒绝复用 → 无可用素材 → 400「素材可切区间不足」。 60s 素材、5s 片段：前 12 个空闲、随后复用占比累计；当 (reused+d)/(assigned+d) 超过 15% 时回调返回 None，calc 返回 None，轮询无素材 → 400。 """ from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc(replace_return_count=0) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 60.0)) body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=20) # 模拟真实回调：累计复用时长，预判超 15% 拒绝 reused: dict = {} assigned: dict = {} def fake_reuse_cb(aid, clip_duration): a = assigned.get(aid, 0.0) r = reused.get(aid, 0.0) if a > 0 and (r + clip_duration) / (a + clip_duration) > 0.15: return None # 占比闸门拒绝 reused[aid] = r + clip_duration return (0.0, clip_duration) def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None): a = assigned.get(asset_id, 0.0) # 前 12 个片段（60s/5s）有空闲区间 if a < 60.0: start = a assigned[asset_id] = a + clip_duration return start # 之后空闲耗尽 → 复用 if on_exhausted is not None: result = on_exhausted(asset_id, clip_duration) if result is not None: assigned[asset_id] = assigned.get(asset_id, 0.0) + clip_duration return result[0] if result else None return None with ( _patch_segments(_segments(20, dur_min=5.0, dur_max=5.0)), patch("app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=fake_calc), patch("app.api.routes.templates_editor.clips.make_reuse_callback", return_value=fake_reuse_cb), ): with pytest.raises(HTTPException) as exc_info: create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) assert exc_info.value.status_code == 400 assert "素材可切区间不足" in exc_info.value.detail # 闸门在复用占比达上限时拒绝：60s 空闲 + 至多 ~15% 复用 assert reused.get("a1", 0.0) <= 12.0 # 10.0 或 15.0 以内，不会无限复用 # 未创建任何片段（整批失败） assert not mock_plan_svc.replace_all_clips_transactional.called @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_calc_none_falls_through_to_next_asset(self, mock_storage): """一个素材区间耗尽且复用被拒（calc 返回 None）时，轮询到下一个可用素材。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc(replace_return_count=2) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock( side_effect=lambda aid: { "exhausted": _make_mock_asset("exhausted", 60.0), "fresh": _make_mock_asset("fresh", 60.0), }[aid] ) body = ClipsFromAssetsRequest(asset_ids=["exhausted", "fresh"], required_clips_count=2) def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None): if asset_id == "exhausted": # 空闲耗尽 + 回调拒绝 → None return on_exhausted(asset_id, clip_duration) if on_exhausted else None return 8.0 # 新鲜素材正常返回 with ( _patch_segments(_segments(2, dur_min=5.0, dur_max=5.0)), patch("app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=fake_calc), patch( "app.api.routes.templates_editor.clips.make_reuse_callback", return_value=lambda aid, d: None, # 复用始终被拒 ), ): create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) clips_data = _get_clips_data_from_call(mock_plan_svc) assert len(clips_data) == 2 # 耗尽素材被跳过，两个片段都分配给新鲜素材 assert all(c["asset_id"] == "fresh" for c in clips_data) # ── P0 回归：from-assets 对 undefined/null/空串 asset_ids 容错 ────────────── # 线上事故：前端 smart-match 拿到 {asset: {...}} 包装层后 items.map(a=>a.id) # 全为 undefined，POST /clips/from-assets 携带 null → 422，自动模式预览断裂。 class TestClipsFromAssetsInvalidIds: def test_schema_filters_null_and_empty_ids(self): """请求 schema 在 pre 阶段剔除 null/空串/空白 id，不触发 422。""" from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest body = ClipsFromAssetsRequest(asset_ids=["a1", None, "", " ", "a2"]) # type: ignore[list-item] assert body.asset_ids == ["a1", "a2"] def test_schema_all_invalid_raises(self): """全部为非法 id 时 min_length=1 兜底报校验错误（前端得到 422 而非脏数据）。""" from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest from pydantic import ValidationError with pytest.raises(ValidationError): ClipsFromAssetsRequest(asset_ids=[None, "", " "]) # type: ignore[list-item] @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_route_filters_invalid_ids_and_uses_valid(self, mock_storage): """路由层二次兜底：混有 null/空串时只用合法 id 正常创建片段。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc(replace_return_count=2) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, 30.0)) body = ClipsFromAssetsRequest(asset_ids=["a1", None, "", "a2"]) # type: ignore[list-item] # 消除排序噪声，确定性断言两条合法素材各被使用 with _patch_zero_noise(), _patch_segments(_segments(2, dur_min=3.0, dur_max=5.0)): result = create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), 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_from_call(mock_plan_svc) assert {c["asset_id"] for c in clips_data} == {"a1", "a2"} @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_route_all_empty_ids_raises_400(self, mock_storage): """schema 被绕过直接调路由、且 id 全非法时，路由 400 而非 500/422。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) mock_plan_svc = _make_plan_svc() mock_asset_repo = MagicMock() body = MagicMock() body.asset_ids = [None, "", " "] with _patch_segments(DEFAULT_SEGMENTS): with pytest.raises(HTTPException) as exc_info: create_clips_from_assets_editor( template_id="tpl-001", body=body, background_tasks=MagicMock(), plan_id=TEST_PLAN_ID, services=(MagicMock(), 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()