"""测试 SceneChange 场景检测前置到渲染前 + 场景点缓存读写。 验证： - 场景点缓存读取（extract_scene_points_from_metadata）：合法/非法/脏数据 - pick_scene_aware_start：随机镜头段选取、冲突避让、shuffle 随机化、无缓存回退 None - from-assets 路径：metadata 有 scene_change_points 时，start_time 落在镜头段内 - 一键生成路径：distribute_assets 传入 asset_scene_points 时使用镜头段 - 后台任务：检测结果写入素材 metadata（缓存） """ 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 packages.domain.plan_generator_utils import ( _calc_random_start_time, build_scene_segments, distribute_assets, extract_scene_points_from_metadata, pick_scene_aware_start, ) # ── metadata 缓存解析 ───────────────────────────────────────────────────────── class TestExtractScenePoints: def test_valid_points(self): md = {"scene_change_points": [0.0, 3.2, 7.8, 12.5]} assert extract_scene_points_from_metadata(md) == [0.0, 3.2, 7.8, 12.5] def test_missing_returns_none(self): assert extract_scene_points_from_metadata({}) is None assert extract_scene_points_from_metadata(None) is None assert extract_scene_points_from_metadata("not-a-dict") is None def test_empty_list_returns_none(self): assert extract_scene_points_from_metadata({"scene_change_points": []}) is None assert extract_scene_points_from_metadata({"scene_change_points": [0.0]}) is None def test_dirty_data_returns_none(self): assert extract_scene_points_from_metadata({"scene_change_points": ["a", 1.0]}) is None def test_auto_prepends_zero(self): result = extract_scene_points_from_metadata({"scene_change_points": [3.2, 7.8]}) assert result == [0.0, 3.2, 7.8] def test_sorts_unsorted(self): result = extract_scene_points_from_metadata({"scene_change_points": [0.0, 12.5, 3.2, 7.8]}) assert result == [0.0, 3.2, 7.8, 12.5] def test_negative_rejected(self): assert extract_scene_points_from_metadata({"scene_change_points": [-1.0, 3.2]}) is None # ── 镜头段选取 ──────────────────────────────────────────────────────────────── class TestPickSceneAwareStart: def test_start_within_some_scene_segment(self): """有缓存时，起点落在某个镜头段内部。""" points = [0.0, 5.0, 10.0, 15.0] durations = {"a1": 18.0} scene_points = {"a1": points} used: dict = {} start = pick_scene_aware_start("a1", 3.0, durations, scene_points, used) assert start is not None segments = build_scene_segments(points, 18.0) assert any(seg_start <= start and start + 3.0 <= seg_end for seg_start, seg_end in segments) def test_no_cache_returns_none(self): """无缓存返回 None（调用方回退随机起点）。""" result = pick_scene_aware_start("a1", 3.0, {"a1": 18.0}, {}, {}) assert result is None def test_bounds_respected(self): """起点 + 片段时长不超过素材总时长。""" points = [0.0, 5.0, 10.0, 15.0] for _ in range(30): start = pick_scene_aware_start("a1", 4.0, {"a1": 18.0}, {"a1": points}, {}) assert start is not None assert start + 4.0 <= 18.0 + 1e-6 def test_conflict_avoidance(self): """所有镜头段都被占满时返回 None（回退随机路径）。""" # 3 段各 6s，片段 5s；把所有段占满（[0,5.5] [5.5,11] 覆盖段1/2，段3太短放不下5s） points = [0.0, 6.0, 12.0] used = {"a1": [(0.0, 5.6), (6.0, 11.6)]} # 段3 [12, 18] 可用 → 应返回其中起点 start = pick_scene_aware_start("a1", 5.0, {"a1": 18.0}, {"a1": points}, used) assert start is not None assert start >= 12.0 def test_all_segments_conflict_returns_none(self): """全部镜头段都冲突时返回 None。""" points = [0.0, 6.0, 12.0] # 占满整个素材 used = {"a1": [(0.0, 18.0)]} start = pick_scene_aware_start("a1", 5.0, {"a1": 18.0}, {"a1": points}, used) assert start is None def test_shuffle_produces_varied_segments(self): """镜头段顺序被 shuffle：30 次选取，起点分布应覆盖多个镜头段。""" points = [0.0, 5.0, 10.0, 15.0] observed: set[int] = set() for _ in range(40): start = pick_scene_aware_start("a1", 2.0, {"a1": 18.0}, {"a1": points}, {}) assert start is not None # 记录起点落在哪个段（段宽 5s） observed.add(int(start // 5.0)) assert len(observed) >= 3, f"镜头段 shuffle 后应覆盖多个段，实际 {observed}" # ── 一键生成路径：distribute_assets 接入场景缓存 ────────────────────────────── class TestDistributeWithScenePoints: def _make_clips(self, n): from packages.domain.edit_plan_clip import EditPlanClip return [EditPlanClip(id=f"c{i}", plan_id="p1", clip_type="main", duration=4.0, order=i) for i in range(n)] def test_one_take_uses_scene_segments(self): """ONE_TAKE 模式下，有场景缓存的素材起点落在镜头段内。""" from packages.domain.editing_mode import EditingMode clips = self._make_clips(2) points = [0.0, 6.0, 12.0, 18.0] distribute_assets( clips, ["a1"], EditingMode.ONE_TAKE.value, asset_durations={"a1": 24.0}, asset_scene_points={"a1": points}, ) segments = build_scene_segments(points, 24.0) for clip in clips: assert clip.start_time is not None assert any( s <= clip.start_time and clip.start_time + 4.0 <= e for s, e in segments ), f"起点 {clip.start_time} 不在任何镜头段内" def test_no_scene_points_falls_back_random(self): """无场景缓存时正常分配（回退随机起点），不报错。""" from packages.domain.editing_mode import EditingMode clips = self._make_clips(1) distribute_assets( clips, ["a1"], EditingMode.ONE_TAKE.value, asset_durations={"a1": 24.0}, asset_scene_points={}, ) for clip in clips: assert clip.asset_id == "a1" assert clip.start_time is not None assert 0.0 <= clip.start_time <= 20.0 def test_random_preview_ignores_scene_points(self): """random_selection 预览模式行为不变（不崩溃、正常分配）。""" from packages.domain.editing_mode import EditingMode clips = self._make_clips(2) distribute_assets( clips, ["a1", "a2"], EditingMode.ONE_TAKE.value, random_selection=True, asset_durations={"a1": 24.0, "a2": 24.0}, asset_scene_points={"a1": [0.0, 6.0]}, ) assert all(c.asset_id for c in clips) # ── from-assets 路径：渲染前读缓存选镜头段 ──────────────────────────────────── def _make_auth_user(): auth = MagicMock() auth.user.id = "user-001" auth.user.email = "test@example.com" auth.user.display_name = "测试用户" auth.user_id = "user-001" return auth def _make_asset_with_scenes(aid, duration, scene_points=None): asset = MagicMock() asset.id = aid asset.duration = duration asset.quality_score = None asset.created_at = None asset.metadata = {"scene_change_points": scene_points} if scene_points is not None else {} return asset class TestFromAssetsSceneCache: def test_cached_scene_points_used_for_start_time(self): """素材 metadata 有场景点缓存时，片段起点落在镜头段内。""" from app.api.routes.templates_editor.clips import create_clips_from_assets_editor from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest # 镜头段宽度 10s，片段最长 ~5.3s（含转场补偿），每段都能容纳 scene_points = [0.0, 10.0, 20.0, 30.0] asset_duration = 40.0 mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock( side_effect=lambda aid: _make_asset_with_scenes(aid, asset_duration, scene_points) ) mock_plan_svc = MagicMock() mock_plan_svc.replace_all_clips_transactional = MagicMock(return_value=3) segments = [(0, 3.0, 5.0), (1, 3.0, 5.0), (2, 3.0, 5.0)] with ( patch( "app.api.routes.templates_editor.clips._get_template_segments", return_value=segments, ), 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), ): body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3) create_clips_from_assets_editor( template_id="tmpl-1", body=body, background_tasks=MagicMock(), plan_id="plan-scene-1", services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) clips_data = mock_plan_svc.replace_all_clips_transactional.call_args.args[1] scene_segments = build_scene_segments(scene_points, asset_duration) for clip in clips_data: start = clip["start_time"] dur = clip["duration"] in_segment = any(s <= start and start + dur <= e + 0.1 for s, e in scene_segments) assert in_segment, f"起点 {start:.2f} 时长 {dur:.2f} 不在任何镜头段内" def test_no_cache_falls_back_random_no_error(self): """素材无场景缓存时正常走随机起点，流程不报错。""" from app.api.routes.templates_editor.clips import create_clips_from_assets_editor from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_asset_with_scenes(aid, 30.0, None)) mock_plan_svc = MagicMock() mock_plan_svc.replace_all_clips_transactional = MagicMock(return_value=2) segments = [(0, 3.0, 5.0), (1, 3.0, 5.0)] with ( patch( "app.api.routes.templates_editor.clips._get_template_segments", return_value=segments, ), 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), ): body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=2) create_clips_from_assets_editor( template_id="tmpl-1", body=body, background_tasks=MagicMock(), plan_id="plan-scene-2", services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, db=MagicMock(), current_user=_make_auth_user(), ) clips_data = mock_plan_svc.replace_all_clips_transactional.call_args.args[1] assert len(clips_data) == 2 assert all(c["start_time"] is not None for c in clips_data) # ── 后台任务：检测结果写缓存 ────────────────────────────────────────────────── class TestAsyncCacheWrite: def test_scene_points_persisted_to_metadata(self): """detect_scene_changes 返回结果后写入素材 metadata 并调用 repo.update。""" from app.api.routes.templates_editor import clips as clips_module detected_points = [0.0, 4.5, 9.0, 14.2] mock_asset = MagicMock() mock_asset.id = "a1" mock_asset.duration = 20.0 mock_asset.storage_key = "v.mp4" mock_asset.mime_type = "video/mp4" mock_asset.metadata = {} mock_asset_repo = MagicMock() mock_asset_repo.find_by_ids = MagicMock(return_value=[mock_asset]) mock_asset_repo.update = MagicMock(side_effect=lambda a: a) mock_clip = MagicMock() mock_clip.id = "clip-1" mock_clip.asset_id = "a1" mock_clip.order = 0 mock_clip.start_time = 2.0 mock_clip.duration = 4.0 mock_plan_svc = MagicMock() mock_plan_svc.list_clips = MagicMock(return_value=[mock_clip]) mock_plan_svc.update_clip = MagicMock() plan_svc_factory = MagicMock(return_value=mock_plan_svc) mock_client = MagicMock() mock_client.is_available = True mock_client.detect_scene_changes = MagicMock(return_value=detected_points) mock_storage = MagicMock() mock_storage.get_download_url = MagicMock(return_value="https://example.com/v.mp4") mock_session = MagicMock() with ( patch( "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository", return_value=mock_asset_repo, ), patch( "app.api.routes.templates_editor.clips.EditPlanService", plan_svc_factory, ), patch( "packages.adapters.sqlalchemy_impl.session.SessionLocal", MagicMock(return_value=mock_session), ), patch( "app.api.routes.templates_editor.clips.get_storage_service", return_value=mock_storage, ), patch( "app.api.routes.templates_editor.clips.get_mediakit_client", return_value=mock_client, ), 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.remove_used_segment", return_value=False, ), ): clips_module._update_mediakit_recommendations_async("plan-1", ["a1"]) # 验证素材 metadata 被写入场景点并持久化 assert mock_asset.metadata.get("scene_change_points") == detected_points mock_asset_repo.update.assert_called()