"""SceneStillCheckpointLoader 단위 테스트 — W4 P3-1c. 체크포인트 디스크 I/O는 _load_cp injection으로 대체. loader는 순수 로직이어야 함. """ from __future__ import annotations from typing import Any, Dict, Optional import pytest def _make_loader(checkpoints: Dict[str, Optional[Dict[str, Any]]]): """step_id → checkpoint dict 매핑을 받아 SceneStillCheckpointLoader를 생성.""" from app.services.checkpoint_sync.scene_still_checkpoint_loader import ( SceneStillCheckpointLoader, ) def load_cp(step_id: str): return checkpoints.get(step_id) return SceneStillCheckpointLoader(load_cp) # ── 빈 상태 ── def test_load_empty_returns_defaults(): """모든 체크포인트가 없으면 빈 번들 반환, sd_completed=False.""" bundle = _make_loader({}).load() assert bundle.scenes == [] assert bundle.sd_completed is False assert bundle.shot_deps == [] assert bundle.shot_dep_completed is False assert bundle.shot_info_by_scene == {} assert bundle.selected_flag_by_scene == {} assert bundle.scene_director_ve == {} assert bundle.shot_director_ve_map == {} assert bundle.scene_summaries == [] assert bundle.shot_cine_shots == [] assert bundle.scene_cine_scenes == [] # ── scene_detail ── def test_load_scene_detail_sets_scenes_and_completed(): cp = {"status": "completed", "data": {"scenes": [{"scene_index": 1}, {"scene_index": 2}]}} bundle = _make_loader({"scene_detail": cp}).load() assert bundle.sd_completed is True assert len(bundle.scenes) == 2 assert bundle.scenes[0]["scene_index"] == 1 def test_load_scene_detail_partial_with_data_is_syncable(): """M2 Fix 2 (2026-05-01): partial이라도 데이터 있으면 sync 허용. 1 shot fail이 64 shot 전체를 차단하던 사고 재발 방지 — `is_cp_syncable` 단일 표준. """ cp = {"status": "partial", "data": {"scenes": [{"scene_index": 1}]}} bundle = _make_loader({"scene_detail": cp}).load() assert bundle.sd_completed is True assert len(bundle.scenes) == 1 def test_load_scene_detail_partial_with_empty_scenes_not_syncable(): """cascade 가드: partial + scenes=[] (cascade 직후)은 not syncable. 빈 plan으로 sync 시도하면 Writer가 모든 row를 stale로 마킹 → row 손상 위험. """ cp = {"status": "partial", "data": {"scenes": []}} bundle = _make_loader({"scene_detail": cp}).load() assert bundle.sd_completed is False def test_load_scene_detail_running_not_syncable(): """running/error 등 비완료 상태는 항상 skip.""" cp = {"status": "running", "data": {"scenes": [{"scene_index": 1}]}} bundle = _make_loader({"scene_detail": cp}).load() assert bundle.sd_completed is False # ── shot_validator + shot_selection ── def test_load_shot_validator_builds_shot_info_by_scene(): val_cp = { "data": { "scenes": [ {"scene_index": 1, "shots": [ {"shot_index": 1, "description": "s1"}, {"shot_index": 2, "description": "s2"}, ]}, {"scene_index": 2, "shots": [{"shot_index": 1, "description": "s3"}]}, ] } } bundle = _make_loader({"shot_validator": val_cp}).load() assert 1 in bundle.shot_info_by_scene assert 2 in bundle.shot_info_by_scene assert len(bundle.shot_info_by_scene[1]) == 2 def test_load_shot_selection_filters_selected_flags(): val_cp = {"data": {"scenes": [{"scene_index": 1, "shots": [ {"shot_index": 1}, {"shot_index": 2}, {"shot_index": 3}, ]}]}} sel_cp = {"data": {"scenes": [{"scene_index": 1, "selected_shot_indices": [1, 3]}]}} bundle = _make_loader({"shot_validator": val_cp, "shot_selection": sel_cp}).load() assert bundle.selected_flag_by_scene[1] == {1, 3} def test_load_shot_selection_missing_defaults_to_all_shots(): """shot_selection 체크포인트 없으면 모든 shot이 selected.""" val_cp = {"data": {"scenes": [{"scene_index": 1, "shots": [ {"shot_index": 1}, {"shot_index": 2}, ]}]}} bundle = _make_loader({"shot_validator": val_cp}).load() assert bundle.selected_flag_by_scene[1] == {1, 2} # ── scene_director ── def test_load_scene_director_ve_audio_hall(): cp = {"data": {"scenes": [ {"scene_index": 1, "present_entity_ids": ["C01", "L02"], "audio_entity_ids": ["C03"], "hallucination_entity_ids": ["C04"]}, {"scene_index": 2, "present_entity_ids": ["C05"], "audio_entity_ids": [], "hallucination_entity_ids": []}, ]}} bundle = _make_loader({"scene_director": cp}).load() assert bundle.scene_director_ve[1] == ["C01", "L02"] assert bundle.scene_director_audio[1] == ["C03"] assert bundle.scene_director_hall[1] == ["C04"] assert bundle.scene_director_ve[2] == ["C05"] assert bundle.scene_director_audio[2] == [] # ── shot_director ── def test_load_shot_director_ve_map(): cp = {"data": {"scenes": [ {"scene_index": 1, "shots": [ {"shot_index": 1, "visible_entity_ids": ["C01"]}, {"shot_index": 2, "visible_entity_ids": ["C02", "L01"]}, ]}, {"scene_index": 2, "shots": [ {"shot_index": 1, "visible_entity_ids": ["C03"]}, ]}, ]}} bundle = _make_loader({"shot_director": cp}).load() assert bundle.shot_director_ve_map[(1, 1)] == ["C01"] assert bundle.shot_director_ve_map[(1, 2)] == ["C02", "L01"] assert bundle.shot_director_ve_map[(2, 1)] == ["C03"] # ── shot_dependency ── def test_load_shot_dependency_collects_deps(): cp = {"status": "completed", "data": {"dependencies": [ {"scene_index": 3, "shot_index": 1, "location_refs": [{"scene_index": 1, "shot_index": 2}]}, ]}} bundle = _make_loader({"shot_dependency": cp}).load() assert len(bundle.shot_deps) == 1 assert bundle.shot_deps[0]["scene_index"] == 3 assert bundle.shot_dep_completed is True def test_load_shot_dependency_partial_not_completed(): cp = {"status": "partial", "data": {"dependencies": []}} bundle = _make_loader({"shot_dependency": cp}).load() assert bundle.shot_dep_completed is False # ── scene_dependency (legacy) ── def test_load_scene_dependency_map(): cp = {"data": {"dependencies": [ {"scene_index": 2, "location_refs": [{"scene_index": 1}]}, {"scene_index": 3, "prev_ref": 2}, ]}} bundle = _make_loader({"scene_dependency": cp}).load() assert "2" in bundle.dep_map assert "3" in bundle.dep_map def test_load_scene_dependency_dict_form(): """data.dependencies가 dict인 legacy 포맷.""" cp = {"data": {"dependencies": {"2": {"prev_ref": 1}}}} bundle = _make_loader({"scene_dependency": cp}).load() assert bundle.dep_map == {"2": {"prev_ref": 1}} # ── scene_summary ── def test_load_scene_summaries(): cp = {"status": "completed", "data": {"summaries": [ {"scene_index": 1, "scene_summary": "S1"}, {"scene_index": 2, "scene_summary": "S2"}, ]}} bundle = _make_loader({"scene_summary": cp}).load() assert len(bundle.scene_summaries) == 2 assert bundle.scene_summaries[0]["scene_index"] == 1 def test_load_scene_summaries_incomplete_ignored(): cp = {"status": "partial", "data": {"summaries": [{"scene_index": 1}]}} bundle = _make_loader({"scene_summary": cp}).load() assert bundle.scene_summaries == [] # ── cinematography ── def test_load_shot_cinematography(): cp = {"data": {"shots": [ {"scene_index": 1, "shot_index": 1, "technique_1": {"name": "wide"}, "technique_2": {"name": "close"}}, ]}} bundle = _make_loader({"shot_cinematography": cp}).load() assert len(bundle.shot_cine_shots) == 1 def test_load_scene_cinematography_legacy(): cp = {"status": "completed", "data": {"scenes": [ {"scene_index": 1, "shots": [{"name": "wide"}, {"name": "close"}]}, ]}} bundle = _make_loader({"scene_cinematography": cp}).load() assert len(bundle.scene_cine_scenes) == 1