"""SceneAnalysisContext + SceneContextLoader 단위 테스트. Phase 3.6. """ from __future__ import annotations import json from pathlib import Path from unittest.mock import MagicMock import pytest from app.core.dto import SceneAnalysisContext from app.core.steps.scene_context_loader import SceneContextLoader @pytest.fixture def project_episode(tmp_path: Path, monkeypatch): monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) # G3.2: bg-mode off — 본 fixture 는 outlook/segments/dependency loader 검증용 이며 # background_prompt cp 부재 시 fail-fast 가 발동되면 안 됨. monkeypatch.setattr( "app.core.config.settings.background_mode", "off", raising=False, ) return "p1", "e1" def _write_cp(tmp_path: Path, project_id: str, episode_id: str, step_id: str, payload: dict): cp_dir = tmp_path / project_id / "checkpoints" / "episodes" / episode_id / step_id cp_dir.mkdir(parents=True, exist_ok=True) (cp_dir / "manifest.json").write_text(json.dumps(payload), encoding="utf-8") class _FakeRunner: """최소 runner — SceneContextLoader가 요구하는 속성만.""" def __init__(self, project_id: str, episode_id: str, tmp_path: Path): self.project_id = project_id self.episode_id = episode_id self.tmp_path = tmp_path self.db = MagicMock() # shot_type 테이블 조회 → 빈 리스트 self.db.execute.return_value.fetchall.return_value = [] def _load_prev_checkpoint(self, step_id: str): cp = ( self.tmp_path / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if not cp.exists(): return None return json.loads(cp.read_text(encoding="utf-8")) # ── DTO 기본 ────────────────────────────────────────────── def test_dto_defaults_are_empty(): ctx = SceneAnalysisContext(project_id="p", episode_id="e") assert ctx.segments == [] assert ctx.director_scenes == [] assert ctx.dependencies == [] assert ctx.outlook_data == {} assert ctx.staging_map == {} assert ctx.fixed_elements_by_scene == {} assert ctx.scene_visible == {} assert ctx.world_rules is None assert ctx.planning_context is None # ── Loader: 빈 상태 ────────────────────────────────────── def test_loader_no_checkpoints_returns_empty_context(project_episode, tmp_path, monkeypatch): pid, eid = project_episode runner = _FakeRunner(pid, eid, tmp_path) # planning_context 함수 stub monkeypatch.setattr( "app.core.planning_doc_context.get_planning_context", lambda *a, **k: None, ) ctx = SceneContextLoader(runner).load_all() assert ctx.project_id == pid assert ctx.segments == [] assert ctx.director_scenes == [] assert ctx.staging_map == {} assert ctx.fixed_elements_by_scene == {} # ── Loader: 부분 체크포인트 ────────────────────────────── def test_loader_loads_segments_and_director(project_episode, tmp_path, monkeypatch): pid, eid = project_episode _write_cp(tmp_path, pid, eid, "scene_save", { "data": {"segments": [{"scene_index": 1, "text": "씬 1"}]}, }) _write_cp(tmp_path, pid, eid, "scene_director", { "data": {"scenes": [{"scene_index": 1, "present_entity_ids": ["C01", "L01"]}]}, }) runner = _FakeRunner(pid, eid, tmp_path) monkeypatch.setattr( "app.core.planning_doc_context.get_planning_context", lambda *a, **k: None, ) ctx = SceneContextLoader(runner).load_all() assert len(ctx.segments) == 1 assert ctx.segments[0]["scene_index"] == 1 assert ctx.scene_visible == {1: ["C01", "L01"]} def test_loader_shot_dependency_preferred_over_scene_dependency(project_episode, tmp_path, monkeypatch): pid, eid = project_episode _write_cp(tmp_path, pid, eid, "shot_dependency", { "data": {"dependencies": [{"scene_index": 1, "shot_index": 1, "source": "shot"}]}, }) _write_cp(tmp_path, pid, eid, "scene_dependency", { "data": {"dependencies": [{"scene_index": 1, "source": "scene"}]}, }) runner = _FakeRunner(pid, eid, tmp_path) monkeypatch.setattr( "app.core.planning_doc_context.get_planning_context", lambda *a, **k: None, ) ctx = SceneContextLoader(runner).load_all() assert ctx.dependencies[0]["source"] == "shot" def test_loader_scene_dependency_fallback_when_shot_empty(project_episode, tmp_path, monkeypatch): pid, eid = project_episode _write_cp(tmp_path, pid, eid, "scene_dependency", { "data": {"dependencies": [{"scene_index": 1, "source": "scene"}]}, }) runner = _FakeRunner(pid, eid, tmp_path) monkeypatch.setattr( "app.core.planning_doc_context.get_planning_context", lambda *a, **k: None, ) ctx = SceneContextLoader(runner).load_all() assert ctx.dependencies[0]["source"] == "scene" def test_loader_outlook_phase3_preferred_over_extraction(project_episode, tmp_path, monkeypatch): pid, eid = project_episode _write_cp(tmp_path, pid, eid, "outlook_phase3", { "data": {"outlooks": [{"name": "new", "short_id": "O01"}]}, }) _write_cp(tmp_path, pid, eid, "outlook_extraction", { "data": {"outlooks": [{"name": "legacy", "short_id": "O99"}]}, }) runner = _FakeRunner(pid, eid, tmp_path) monkeypatch.setattr( "app.core.planning_doc_context.get_planning_context", lambda *a, **k: None, ) ctx = SceneContextLoader(runner).load_all() assert ctx.outlook_data["outlooks"][0]["name"] == "new" def test_loader_fixed_elements_mapping(project_episode, tmp_path, monkeypatch): pid, eid = project_episode _write_cp(tmp_path, pid, eid, "scene_consistency", { "data": { "scenes": [ {"scene_index": 12, "fixed_elements": [{"type": "dead_character", "entity_id": "C01"}]}, {"scene_index": 13, "fixed_elements": []}, # 빈 리스트는 제외 ], }, }) runner = _FakeRunner(pid, eid, tmp_path) monkeypatch.setattr( "app.core.planning_doc_context.get_planning_context", lambda *a, **k: None, ) ctx = SceneContextLoader(runner).load_all() assert 12 in ctx.fixed_elements_by_scene assert 13 not in ctx.fixed_elements_by_scene def test_loader_shot_selection_filters_shots(project_episode, tmp_path, monkeypatch): pid, eid = project_episode _write_cp(tmp_path, pid, eid, "shot_selection", { "data": {"scenes": [{"scene_index": 1, "selected_shot_indices": [1, 3]}]}, }) # 다운스트림 loader는 shot_validator 체크포인트를 읽는다 (shot-validator 도입 이후) _write_cp(tmp_path, pid, eid, "shot_validator", { "data": {"scenes": [{"scene_index": 1, "shots": [ {"shot_index": 1, "description": "a"}, {"shot_index": 2, "description": "b"}, {"shot_index": 3, "description": "c"}, ]}]}, }) runner = _FakeRunner(pid, eid, tmp_path) monkeypatch.setattr( "app.core.planning_doc_context.get_planning_context", lambda *a, **k: None, ) ctx = SceneContextLoader(runner).load_all() assert ctx.selected_map == {1: {1, 3}} shot_indices = [sh["shot_index"] for sh in ctx.shot_scenes_map[1]] assert shot_indices == [1, 3]