"""scene_dependency_v2 + director_steps (SceneCinematographyStep, SceneDependencyStep) 단위 테스트.""" import json from unittest.mock import MagicMock, patch, call import pytest # -- Sample data -- SAMPLE_SEGMENTS = [ {"scene_index": 1, "heading": "INT. 거실 - 낮", "start_char": 0, "end_char": 100, "length": 100}, {"scene_index": 2, "heading": "EXT. 골목 - 밤", "start_char": 100, "end_char": 250, "length": 150}, {"scene_index": 3, "heading": "INT. 거실 - 밤", "start_char": 250, "end_char": 400, "length": 150}, ] SAMPLE_FULLTEXT = "A" * 100 + "B" * 150 + "C" * 150 SAMPLE_DIRECTOR_RESULT = { "scenes": [ {"scene_index": 1, "present_entity_ids": ["C01", "C02"]}, {"scene_index": 2, "present_entity_ids": ["C01"]}, {"scene_index": 3, "present_entity_ids": ["C01", "C02"]}, ], } SAMPLE_ENTITIES = { "characters": [ {"short_id": "C01", "name": "주인공"}, {"short_id": "C02", "name": "조연"}, ], "locations": [ {"short_id": "L01", "name": "거실"}, {"short_id": "L02", "name": "골목"}, ], "props": [], } SAMPLE_LLM_DEPENDENCY_RESULT = { "dependencies": [ {"scene_index": 1, "location_refs": [], "character_refs": [], "reason": "첫 씬"}, {"scene_index": 2, "location_refs": [], "character_refs": [1], "reason": "C01 공통"}, {"scene_index": 3, "location_refs": [1], "character_refs": [1, 2], "reason": "같은 거실, C01/C02 공통"}, ], } # -- Fixtures -- @pytest.fixture() def fake_db(): db = MagicMock() db.execute.return_value.fetchone.return_value = None return db # ===== extract_dependencies (module-level function) ===== class TestExtractDependencies: @patch("app.modules.pipeline.scene_dependency_v2.call_structured") @patch("app.modules.pipeline.scene_dependency_v2.load_schema") @patch("app.modules.pipeline.scene_dependency_v2.load_prompt") def test_returns_dependencies(self, mock_prompt, mock_schema, mock_call): """extract_dependencies가 LLM 결과를 그대로 반환.""" mock_prompt.return_value = "system prompt" mock_schema.return_value = {"type": "object"} mock_call.return_value = SAMPLE_LLM_DEPENDENCY_RESULT from app.modules.pipeline.scene_dependency_v2 import extract_dependencies result = extract_dependencies( segments=SAMPLE_SEGMENTS, fulltext=SAMPLE_FULLTEXT, director_result=SAMPLE_DIRECTOR_RESULT, entities=SAMPLE_ENTITIES, ) assert "dependencies" in result assert len(result["dependencies"]) == 3 assert result["dependencies"][2]["location_refs"] == [1] assert result["dependencies"][2]["character_refs"] == [1, 2] @patch("app.modules.pipeline.scene_dependency_v2.call_structured") @patch("app.modules.pipeline.scene_dependency_v2.load_schema") @patch("app.modules.pipeline.scene_dependency_v2.load_prompt") def test_calls_structured_with_correct_step(self, mock_prompt, mock_schema, mock_call): """call_structured에 step='scene_dependency'가 전달.""" mock_prompt.return_value = "system" mock_schema.return_value = {"type": "object"} mock_call.return_value = {"dependencies": []} from app.modules.pipeline.scene_dependency_v2 import extract_dependencies extract_dependencies( segments=SAMPLE_SEGMENTS, fulltext=SAMPLE_FULLTEXT, director_result=SAMPLE_DIRECTOR_RESULT, entities=SAMPLE_ENTITIES, project_config={"scene_dependency": {"model": "gpt"}}, ) mock_call.assert_called_once() kwargs = mock_call.call_args assert kwargs[1]["step"] == "scene_dependency" assert kwargs[1]["schema_name"] == "scene_dependency_v2" assert kwargs[1]["project_config"] == {"scene_dependency": {"model": "gpt"}} @patch("app.modules.pipeline.scene_dependency_v2.call_structured") @patch("app.modules.pipeline.scene_dependency_v2.load_schema") @patch("app.modules.pipeline.scene_dependency_v2.load_prompt") def test_loads_correct_prompts(self, mock_prompt, mock_schema, mock_call): """load_prompt(scene_dependency, system) + load_schema(scene_dependency, dependency_schema) 호출.""" mock_prompt.return_value = "sys" mock_schema.return_value = {"type": "object"} mock_call.return_value = {"dependencies": []} from app.modules.pipeline.scene_dependency_v2 import extract_dependencies extract_dependencies( segments=SAMPLE_SEGMENTS, fulltext=SAMPLE_FULLTEXT, director_result=SAMPLE_DIRECTOR_RESULT, entities=SAMPLE_ENTITIES, ) mock_prompt.assert_called_once_with("scene_dependency", "system") mock_schema.assert_called_once_with("scene_dependency", "dependency_schema") @patch("app.modules.pipeline.scene_dependency_v2.call_structured") @patch("app.modules.pipeline.scene_dependency_v2.load_schema") @patch("app.modules.pipeline.scene_dependency_v2.load_prompt") def test_user_prompt_includes_entity_and_scene_info(self, mock_prompt, mock_schema, mock_call): """user_prompt에 엔티티 목록과 씬 정보가 포함.""" mock_prompt.return_value = "sys" mock_schema.return_value = {"type": "object"} mock_call.return_value = {"dependencies": []} from app.modules.pipeline.scene_dependency_v2 import extract_dependencies extract_dependencies( segments=SAMPLE_SEGMENTS, fulltext=SAMPLE_FULLTEXT, director_result=SAMPLE_DIRECTOR_RESULT, entities=SAMPLE_ENTITIES, ) user_prompt = mock_call.call_args[1]["user_prompt"] # Entity info present assert "C01" in user_prompt assert "주인공" in user_prompt assert "L01" in user_prompt assert "거실" in user_prompt # Scene info present assert "씬 1" in user_prompt assert "씬 3" in user_prompt assert "INT. 거실 - 낮" in user_prompt @patch("app.modules.pipeline.scene_dependency_v2.call_structured") @patch("app.modules.pipeline.scene_dependency_v2.load_schema") @patch("app.modules.pipeline.scene_dependency_v2.load_prompt") def test_handles_empty_director_result(self, mock_prompt, mock_schema, mock_call): """director_result가 빈 dict이어도 에러 없이 동작.""" mock_prompt.return_value = "sys" mock_schema.return_value = {"type": "object"} mock_call.return_value = {"dependencies": []} from app.modules.pipeline.scene_dependency_v2 import extract_dependencies result = extract_dependencies( segments=SAMPLE_SEGMENTS, fulltext=SAMPLE_FULLTEXT, director_result={}, entities=SAMPLE_ENTITIES, ) assert result == {"dependencies": []} @patch("app.modules.pipeline.scene_dependency_v2.call_structured") @patch("app.modules.pipeline.scene_dependency_v2.load_schema") @patch("app.modules.pipeline.scene_dependency_v2.load_prompt") def test_handles_empty_entities(self, mock_prompt, mock_schema, mock_call): """entities가 빈 dict이어도 에러 없이 동작.""" mock_prompt.return_value = "sys" mock_schema.return_value = {"type": "object"} mock_call.return_value = {"dependencies": [{"scene_index": 1, "location_refs": [], "character_refs": [], "reason": "첫 씬"}]} from app.modules.pipeline.scene_dependency_v2 import extract_dependencies result = extract_dependencies( segments=SAMPLE_SEGMENTS, fulltext=SAMPLE_FULLTEXT, director_result=SAMPLE_DIRECTOR_RESULT, entities={}, ) assert len(result["dependencies"]) == 1 # ===== _format_entities ===== class TestFormatEntities: def test_format_all_types(self): from app.modules.pipeline.scene_dependency_v2 import _format_entities result = _format_entities(SAMPLE_ENTITIES) assert "- C01: 주인공 (character)" in result assert "- C02: 조연 (character)" in result assert "- L01: 거실 (location)" in result assert "- L02: 골목 (location)" in result def test_format_empty(self): from app.modules.pipeline.scene_dependency_v2 import _format_entities result = _format_entities({}) assert result == "" def test_format_uses_name_fallback_when_no_short_id(self): from app.modules.pipeline.scene_dependency_v2 import _format_entities entities = {"characters": [{"name": "홍길동"}], "locations": [], "props": []} result = _format_entities(entities) assert "- 홍길동: 홍길동 (character)" in result # ===== SceneDependencyStep ===== class TestSceneDependencyStep: def _make_step(self, fake_db, checkpoints=None): with patch("app.core.steps.director_steps.StepRunner.__init__", return_value=None): from app.core.steps.director_steps import SceneDependencyStep step = SceneDependencyStep.__new__(SceneDependencyStep) step.step_id = "scene_dependency" step.project_id = "proj1" step.episode_id = "ep1" step.db = fake_db step.project_config = {} step.opik_context = {} if checkpoints: step._load_prev_checkpoint = lambda sid: checkpoints.get(sid) else: step._load_prev_checkpoint = lambda sid: None return step @pytest.mark.skip(reason="W3-3 cluster B drift — v3/v2 계약이 현재(v4/shot-more) 구현과 어긋남. docs/review-codex-1/11-fix-plan.md §8.2 참조. 복원/재작성은 Wave 5 이후 재평가.") @patch("app.modules.pipeline.scene_dependency_v2.extract_dependencies") @patch("app.core.steps.director_steps._DirectorStepMixin._load_fulltext") def test_execute_calls_extract_dependencies(self, mock_fulltext, mock_extract, fake_db): """SceneDependencyStep._execute()가 extract_dependencies를 호출.""" mock_fulltext.return_value = SAMPLE_FULLTEXT mock_extract.return_value = SAMPLE_LLM_DEPENDENCY_RESULT checkpoints = { "scene_split": {"data": {"segments": SAMPLE_SEGMENTS}}, "scene_director": {"data": SAMPLE_DIRECTOR_RESULT}, "entity_t2i": {"data": SAMPLE_ENTITIES}, } step = self._make_step(fake_db, checkpoints) result = step._execute() mock_extract.assert_called_once() assert result["completed_count"] == 3 assert result["applicable_count"] == 3 assert result["failed_count"] == 0 assert len(result["data"]["dependencies"]) == 3 @patch("app.modules.pipeline.scene_dependency_v2.extract_dependencies") @patch("app.core.steps.director_steps._DirectorStepMixin._load_fulltext") def test_falls_back_to_segmentation(self, mock_fulltext, mock_extract, fake_db): """scene_split 없으면 scene_segmentation 폴백.""" mock_fulltext.return_value = SAMPLE_FULLTEXT mock_extract.return_value = {"dependencies": []} checkpoints = { "scene_segmentation": {"data": {"segments": SAMPLE_SEGMENTS}}, "scene_director": {"data": SAMPLE_DIRECTOR_RESULT}, "entity_t2i": {"data": SAMPLE_ENTITIES}, } step = self._make_step(fake_db, checkpoints) result = step._execute() mock_extract.assert_called_once() call_kwargs = mock_extract.call_args[1] assert call_kwargs["segments"] == SAMPLE_SEGMENTS @patch("app.core.steps.director_steps._DirectorStepMixin._load_fulltext") def test_raises_when_no_segments(self, mock_fulltext, fake_db): """세그먼트 체크포인트가 없으면 AppError.""" mock_fulltext.return_value = SAMPLE_FULLTEXT step = self._make_step(fake_db, checkpoints={}) from app.core.errors import AppError with pytest.raises(AppError) as exc_info: step._execute() assert exc_info.value.code == "step.no_input" @pytest.mark.skip(reason="W3-3 cluster B drift — v3/v2 계약이 현재(v4/shot-more) 구현과 어긋남. docs/review-codex-1/11-fix-plan.md §8.2 참조. 복원/재작성은 Wave 5 이후 재평가.") @patch("app.modules.pipeline.scene_dependency_v2.extract_dependencies") @patch("app.core.steps.director_steps._DirectorStepMixin._load_fulltext") def test_handles_missing_director_and_entity(self, mock_fulltext, mock_extract, fake_db): """scene_director, entity_t2i 체크포인트 없어도 빈 dict로 진행.""" mock_fulltext.return_value = SAMPLE_FULLTEXT mock_extract.return_value = {"dependencies": []} checkpoints = { "scene_split": {"data": {"segments": SAMPLE_SEGMENTS}}, } step = self._make_step(fake_db, checkpoints) result = step._execute() call_kwargs = mock_extract.call_args[1] assert call_kwargs["director_result"] == {} assert call_kwargs["entities"] == {} @pytest.mark.skip(reason="W3-3 cluster B drift — v3/v2 계약이 현재(v4/shot-more) 구현과 어긋남. docs/review-codex-1/11-fix-plan.md §8.2 참조. 복원/재작성은 Wave 5 이후 재평가.") @patch("app.modules.pipeline.scene_dependency_v2.extract_dependencies") @patch("app.core.steps.director_steps._DirectorStepMixin._load_fulltext") def test_falls_back_entity_to_detail_batch(self, mock_fulltext, mock_extract, fake_db): """entity_t2i 없으면 entity_detail_batch 폴백.""" mock_fulltext.return_value = SAMPLE_FULLTEXT mock_extract.return_value = {"dependencies": []} detail_entities = {"characters": [{"short_id": "C01", "name": "테스트"}], "locations": [], "props": []} checkpoints = { "scene_split": {"data": {"segments": SAMPLE_SEGMENTS}}, "scene_director": {"data": SAMPLE_DIRECTOR_RESULT}, "entity_detail_batch": {"data": detail_entities}, } step = self._make_step(fake_db, checkpoints) result = step._execute() call_kwargs = mock_extract.call_args[1] assert call_kwargs["entities"] == detail_entities # ===== SceneCinematographyStep ===== class TestSceneCinematographyStep: def _make_step(self, fake_db, checkpoints=None): with patch("app.core.steps.director_steps.StepRunner.__init__", return_value=None): from app.core.steps.director_steps import SceneCinematographyStep step = SceneCinematographyStep.__new__(SceneCinematographyStep) step.step_id = "scene_cinematography" step.project_id = "proj1" step.episode_id = "ep1" step.db = fake_db step.project_config = {} step.opik_context = {} if checkpoints: step._load_prev_checkpoint = lambda sid: checkpoints.get(sid) else: step._load_prev_checkpoint = lambda sid: None return step @pytest.mark.skip(reason="W3-3 cluster B drift — v3/v2 계약이 현재(v4/shot-more) 구현과 어긋남. docs/review-codex-1/11-fix-plan.md §8.2 참조. 복원/재작성은 Wave 5 이후 재평가.") @patch("app.core.steps.director_steps.call_structured") @patch("app.modules.prompt_loader.load_schema") @patch("app.modules.prompt_loader.load_prompt") @patch("app.core.steps.director_steps._DirectorStepMixin._load_fulltext") def test_execute_calls_llm_and_updates_db( self, mock_fulltext, mock_prompt, mock_schema, mock_call, fake_db ): """SceneCinematographyStep._execute()가 LLM 호출 + DB 업데이트.""" mock_fulltext.return_value = SAMPLE_FULLTEXT mock_prompt.return_value = "prompt {shot_types_block} {scenes_block}" mock_schema.return_value = {"type": "object"} llm_result = { "scenes": [ {"scene_index": 1, "shot_1": "close_up", "shot_1_reason": "감정", "shot_1_focus": "주인공", "shot_2": "wide_shot", "shot_2_reason": "공간", "shot_2_focus": "거실"}, {"scene_index": 2, "shot_1": "tracking", "shot_1_reason": "이동", "shot_1_focus": "주인공", "shot_2": "low_angle", "shot_2_reason": "위압감", "shot_2_focus": "골목"}, ], } mock_call.return_value = llm_result # shot_type DB 쿼리 mock shot_row1 = ("close_up", "basic", "클로즈업") shot_row2 = ("wide_shot", "basic", "와이드샷") shot_row3 = ("tracking", "movement", "트래킹") shot_row4 = ("low_angle", "angle", "로우앵글") fake_db.execute.return_value.fetchall.return_value = [shot_row1, shot_row2, shot_row3, shot_row4] checkpoints = { "scene_split": {"data": {"segments": SAMPLE_SEGMENTS[:2]}}, } step = self._make_step(fake_db, checkpoints) result = step._execute() # LLM called once mock_call.assert_called_once() assert mock_call.call_args[1]["step"] == "scene_cinematography" # Checkpoint saved (no DB update — sync handles it later) assert result["completed_count"] == 2 assert result["applicable_count"] == 2 assert result["failed_count"] == 0 assert result["data"]["scenes"] == llm_result["scenes"] @patch("app.core.steps.director_steps._DirectorStepMixin._load_fulltext") def test_raises_when_no_segments(self, mock_fulltext, fake_db): """세그먼트 체크포인트 없으면 AppError.""" mock_fulltext.return_value = SAMPLE_FULLTEXT step = self._make_step(fake_db, checkpoints={}) from app.core.errors import AppError with pytest.raises(AppError) as exc_info: step._execute() assert exc_info.value.code == "step.no_input" @patch("app.core.steps.director_steps.call_structured") @patch("app.modules.prompt_loader.load_schema") @patch("app.modules.prompt_loader.load_prompt") @patch("app.core.steps.director_steps._DirectorStepMixin._load_fulltext") def test_shot_types_from_db_in_prompt( self, mock_fulltext, mock_prompt, mock_schema, mock_call, fake_db ): """DB에서 로드한 shot_type이 user_prompt에 포함.""" mock_fulltext.return_value = SAMPLE_FULLTEXT captured_prompt = {} def capture_template(*args, **kwargs): # The analyze template uses format() with shot_types_block and scenes_block return "{shot_types_block}\n{scenes_block}" mock_prompt.side_effect = capture_template mock_schema.return_value = {"type": "object"} def capture_call(**kwargs): captured_prompt["user"] = kwargs.get("user_prompt", "") return {"scenes": []} mock_call.side_effect = capture_call shot_rows = [("dolly_zoom", "special", "돌리줌 효과")] fake_db.execute.return_value.fetchall.return_value = shot_rows checkpoints = { "scene_segmentation": {"data": {"segments": SAMPLE_SEGMENTS[:1]}}, } step = self._make_step(fake_db, checkpoints) step._execute() assert "dolly_zoom" in captured_prompt["user"] assert "돌리줌 효과" in captured_prompt["user"] @patch("app.core.steps.director_steps.call_structured") @patch("app.modules.prompt_loader.load_schema") @patch("app.modules.prompt_loader.load_prompt") @patch("app.core.steps.director_steps._DirectorStepMixin._load_fulltext") def test_falls_back_to_segmentation( self, mock_fulltext, mock_prompt, mock_schema, mock_call, fake_db ): """scene_split 없으면 scene_segmentation 폴백.""" mock_fulltext.return_value = SAMPLE_FULLTEXT mock_prompt.return_value = "{shot_types_block}\n{scenes_block}" mock_schema.return_value = {"type": "object"} mock_call.return_value = {"scenes": []} fake_db.execute.return_value.fetchall.return_value = [] checkpoints = { "scene_segmentation": {"data": {"segments": SAMPLE_SEGMENTS}}, } step = self._make_step(fake_db, checkpoints) result = step._execute() assert result["applicable_count"] == 3 # ===== DIRECTOR_STEP_CLASSES registry ===== class TestDirectorStepClasses: def test_registry_contains_all_steps(self): from app.core.steps.director_steps import DIRECTOR_STEP_CLASSES assert "scene_director" in DIRECTOR_STEP_CLASSES assert "scene_cinematography" in DIRECTOR_STEP_CLASSES assert "scene_dependency" in DIRECTOR_STEP_CLASSES def test_registry_maps_to_correct_classes(self): from app.core.steps.director_steps import ( DIRECTOR_STEP_CLASSES, SceneDirectorStep, SceneCinematographyStep, SceneDependencyStep, ) assert DIRECTOR_STEP_CLASSES["scene_director"] is SceneDirectorStep assert DIRECTOR_STEP_CLASSES["scene_cinematography"] is SceneCinematographyStep assert DIRECTOR_STEP_CLASSES["scene_dependency"] is SceneDependencyStep