"""scene_director_v2 모듈 테스트 -- V/A/H 3분류 검증.""" import copy import json from unittest.mock import MagicMock, patch, call import pytest # ── Sample data ── SAMPLE_ENTITIES = { "characters": [ {"name": "홍길동", "short_id": "C01"}, {"name": "이몽룡", "short_id": "C02"}, ], "locations": [ {"name": "사무실", "short_id": "L01"}, ], "props": [ {"name": "권총", "short_id": "P01"}, ], } SAMPLE_SEGMENTS = [ {"scene_index": 1, "heading": "INT. 사무실 - 낮", "start_char": 0, "end_char": 100}, {"scene_index": 2, "heading": "EXT. 골목 - 밤", "start_char": 100, "end_char": 250}, ] SAMPLE_FULLTEXT = "A" * 250 # 2026-08-06: 출력 키가 segment_key(불투명 토큰)로 바뀌었다. 래퍼 인덱스와 # 대본 본문 번호가 둘 다 "씬 번호"로 보여 LLM 이 실행마다 다른 쪽을 회신하던 # 결함 봉합 — tests/unit/test_scene_director_index_parity.py 참조. SAMPLE_LLM_RESPONSE = { "scenes": [ { "segment_key": "SEG-001", "scene_type": "normal", "present_entity_ids": ["C01", "L01", "P01"], "audio_entity_ids": ["C02"], "hallucination_entity_ids": [], "not_present": [], }, { "segment_key": "SEG-002", "scene_type": "flashback", "present_entity_ids": ["C02", "L01"], "audio_entity_ids": [], "hallucination_entity_ids": ["C01"], "not_present": [], }, ] } # ── direct_scenes 단위 테스트 ── class TestDirectScenes: @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_director_v2.load_schema") @patch("app.modules.pipeline.scene_director_v2.load_prompt") @patch("app.modules.pipeline.scene_director_v2.call_structured") def test_calls_call_structured_with_correct_step( self, mock_call, mock_prompt, mock_schema ): """call_structured가 step='scene_director'로 호출되는지 검증.""" mock_prompt.return_value = "시스템 프롬프트" mock_schema.return_value = { "type": "object", "properties": { "scenes": { "type": "array", "items": { "type": "object", "properties": { "scene_index": {"type": "integer"}, "scene_type": {"type": "string"}, "present_entity_ids": {"type": "array", "items": {"type": "string"}}, "audio_entity_ids": {"type": "array", "items": {"type": "string"}}, "hallucination_entity_ids": {"type": "array", "items": {"type": "string"}}, "not_present": {"type": "array", "items": {"type": "object", "properties": {"entity_id": {"type": "string"}, "reason": {"type": "string"}}}}, }, }, } }, } mock_call.return_value = SAMPLE_LLM_RESPONSE from app.modules.pipeline.scene_director_v2 import direct_scenes direct_scenes(SAMPLE_SEGMENTS, SAMPLE_FULLTEXT, SAMPLE_ENTITIES) mock_call.assert_called_once() kwargs = mock_call.call_args.kwargs assert kwargs["step"] == "scene_director" assert kwargs["schema_name"] == "scene_director_vah" @patch("app.modules.pipeline.scene_director_v2.load_schema") @patch("app.modules.pipeline.scene_director_v2.load_prompt") @patch("app.modules.pipeline.scene_director_v2.call_structured") def test_schema_has_vah_fields(self, mock_call, mock_prompt, mock_schema): """응답 스키마에 V/A/H 필드가 모두 포함되는지 검증.""" mock_prompt.return_value = "시스템" mock_schema.return_value = { "type": "object", "properties": { "scenes": { "type": "array", "items": { "type": "object", "properties": { "scene_index": {"type": "integer"}, "scene_type": {"type": "string"}, "present_entity_ids": {"type": "array", "items": {"type": "string"}}, "audio_entity_ids": {"type": "array", "items": {"type": "string"}}, "hallucination_entity_ids": {"type": "array", "items": {"type": "string"}}, "not_present": {"type": "array", "items": {"type": "object", "properties": {"entity_id": {"type": "string"}, "reason": {"type": "string"}}}}, }, }, } }, } mock_call.return_value = SAMPLE_LLM_RESPONSE from app.modules.pipeline.scene_director_v2 import direct_scenes direct_scenes(SAMPLE_SEGMENTS, SAMPLE_FULLTEXT, SAMPLE_ENTITIES) schema_passed = mock_call.call_args.kwargs["response_schema"] item_props = schema_passed["properties"]["scenes"]["items"]["properties"] assert "present_entity_ids" in item_props assert "audio_entity_ids" in item_props assert "hallucination_entity_ids" in item_props assert "not_present" in item_props @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_director_v2.load_schema") @patch("app.modules.pipeline.scene_director_v2.load_prompt") @patch("app.modules.pipeline.scene_director_v2.call_structured") def test_enum_injection_with_entities(self, mock_call, mock_prompt, mock_schema): """엔티티 short_id가 스키마 enum에 주입되는지 검증.""" mock_prompt.return_value = "시스템" mock_schema.return_value = { "type": "object", "properties": { "scenes": { "type": "array", "items": { "type": "object", "properties": { "scene_index": {"type": "integer"}, "scene_type": {"type": "string"}, "present_entity_ids": {"type": "array", "items": {"type": "string"}}, "audio_entity_ids": {"type": "array", "items": {"type": "string"}}, "hallucination_entity_ids": {"type": "array", "items": {"type": "string"}}, "not_present": { "type": "array", "items": { "type": "object", "properties": { "entity_id": {"type": "string"}, "reason": {"type": "string"}, }, }, }, }, }, } }, } mock_call.return_value = SAMPLE_LLM_RESPONSE from app.modules.pipeline.scene_director_v2 import direct_scenes direct_scenes(SAMPLE_SEGMENTS, SAMPLE_FULLTEXT, SAMPLE_ENTITIES) schema_passed = mock_call.call_args.kwargs["response_schema"] item_props = schema_passed["properties"]["scenes"]["items"]["properties"] expected_ids = ["C01", "C02", "L01", "P01"] # V field enum assert item_props["present_entity_ids"]["items"]["enum"] == expected_ids # A field enum assert item_props["audio_entity_ids"]["items"]["enum"] == expected_ids # H field enum assert item_props["hallucination_entity_ids"]["items"]["enum"] == expected_ids # not_present entity_id enum assert item_props["not_present"]["items"]["properties"]["entity_id"]["enum"] == expected_ids @patch("app.modules.pipeline.scene_director_v2.load_schema") @patch("app.modules.pipeline.scene_director_v2.load_prompt") @patch("app.modules.pipeline.scene_director_v2.call_structured") def test_no_enum_injection_when_entities_empty(self, mock_call, mock_prompt, mock_schema): """엔티티가 비어있으면 enum 주입을 건너뜀.""" mock_prompt.return_value = "시스템" original_items = {"type": "string"} mock_schema.return_value = { "type": "object", "properties": { "scenes": { "type": "array", "items": { "type": "object", "properties": { "scene_index": {"type": "integer"}, "scene_type": {"type": "string"}, "present_entity_ids": {"type": "array", "items": {"type": "string"}}, "audio_entity_ids": {"type": "array", "items": {"type": "string"}}, "hallucination_entity_ids": {"type": "array", "items": {"type": "string"}}, "not_present": { "type": "array", "items": { "type": "object", "properties": { "entity_id": {"type": "string"}, "reason": {"type": "string"}, }, }, }, }, }, } }, } mock_call.return_value = copy.deepcopy(SAMPLE_LLM_RESPONSE) from app.modules.pipeline.scene_director_v2 import direct_scenes direct_scenes(SAMPLE_SEGMENTS, SAMPLE_FULLTEXT, {}) schema_passed = mock_call.call_args.kwargs["response_schema"] item_props = schema_passed["properties"]["scenes"]["items"]["properties"] # No enum injected assert "enum" not in item_props["present_entity_ids"]["items"] # 단, segment_key enum 은 엔티티 유무와 무관하게 항상 잠긴다. assert item_props["segment_key"]["enum"] == ["SEG-001", "SEG-002"] @patch("app.modules.pipeline.scene_director_v2.load_schema") @patch("app.modules.pipeline.scene_director_v2.load_prompt") @patch("app.modules.pipeline.scene_director_v2.call_structured") def test_returns_proper_structure(self, mock_call, mock_prompt, mock_schema): """반환 구조에 scenes 배열이 포함되는지 검증.""" mock_prompt.return_value = "시스템" mock_schema.return_value = { "type": "object", "properties": { "scenes": { "type": "array", "items": { "type": "object", "properties": { "scene_index": {"type": "integer"}, "scene_type": {"type": "string"}, "present_entity_ids": {"type": "array", "items": {"type": "string"}}, "audio_entity_ids": {"type": "array", "items": {"type": "string"}}, "hallucination_entity_ids": {"type": "array", "items": {"type": "string"}}, "not_present": {"type": "array", "items": {"type": "object", "properties": {"entity_id": {"type": "string"}, "reason": {"type": "string"}}}}, }, }, } }, } mock_call.return_value = SAMPLE_LLM_RESPONSE from app.modules.pipeline.scene_director_v2 import direct_scenes result = direct_scenes(SAMPLE_SEGMENTS, SAMPLE_FULLTEXT, SAMPLE_ENTITIES) assert "scenes" in result assert len(result["scenes"]) == 2 scene1 = result["scenes"][0] assert scene1["scene_type"] == "normal" assert "C01" in scene1["present_entity_ids"] assert "C02" in scene1["audio_entity_ids"] assert scene1["hallucination_entity_ids"] == [] @patch("app.modules.pipeline.scene_director_v2.load_schema") @patch("app.modules.pipeline.scene_director_v2.load_prompt") @patch("app.modules.pipeline.scene_director_v2.call_structured") def test_user_prompt_contains_entity_lines(self, mock_call, mock_prompt, mock_schema): """user_prompt에 엔티티 목록이 포함되는지 검증.""" mock_prompt.return_value = "시스템" mock_schema.return_value = { "type": "object", "properties": { "scenes": { "type": "array", "items": { "type": "object", "properties": { "scene_index": {"type": "integer"}, "scene_type": {"type": "string"}, "present_entity_ids": {"type": "array", "items": {"type": "string"}}, "audio_entity_ids": {"type": "array", "items": {"type": "string"}}, "hallucination_entity_ids": {"type": "array", "items": {"type": "string"}}, "not_present": {"type": "array", "items": {"type": "object", "properties": {"entity_id": {"type": "string"}, "reason": {"type": "string"}}}}, }, }, } }, } mock_call.return_value = SAMPLE_LLM_RESPONSE from app.modules.pipeline.scene_director_v2 import direct_scenes direct_scenes(SAMPLE_SEGMENTS, SAMPLE_FULLTEXT, SAMPLE_ENTITIES) user_prompt = mock_call.call_args.kwargs["user_prompt"] assert "C01: 홍길동 (character)" in user_prompt assert "C02: 이몽룡 (character)" in user_prompt assert "L01: 사무실 (location)" in user_prompt assert "P01: 권총 (prop)" in user_prompt @patch("app.modules.pipeline.scene_director_v2.load_schema") @patch("app.modules.pipeline.scene_director_v2.load_prompt") @patch("app.modules.pipeline.scene_director_v2.call_structured") def test_user_prompt_contains_scene_headings(self, mock_call, mock_prompt, mock_schema): """user_prompt에 씬 헤딩이 포함되는지 검증.""" mock_prompt.return_value = "시스템" mock_schema.return_value = { "type": "object", "properties": { "scenes": { "type": "array", "items": { "type": "object", "properties": { "scene_index": {"type": "integer"}, "scene_type": {"type": "string"}, "present_entity_ids": {"type": "array", "items": {"type": "string"}}, "audio_entity_ids": {"type": "array", "items": {"type": "string"}}, "hallucination_entity_ids": {"type": "array", "items": {"type": "string"}}, "not_present": {"type": "array", "items": {"type": "object", "properties": {"entity_id": {"type": "string"}, "reason": {"type": "string"}}}}, }, }, } }, } mock_call.return_value = SAMPLE_LLM_RESPONSE from app.modules.pipeline.scene_director_v2 import direct_scenes direct_scenes(SAMPLE_SEGMENTS, SAMPLE_FULLTEXT, SAMPLE_ENTITIES) user_prompt = mock_call.call_args.kwargs["user_prompt"] assert "## SEG-001: INT. 사무실 - 낮" in user_prompt assert "## SEG-002: EXT. 골목 - 밤" in user_prompt @patch("app.modules.pipeline.scene_director_v2.load_schema") @patch("app.modules.pipeline.scene_director_v2.load_prompt") @patch("app.modules.pipeline.scene_director_v2.call_structured") def test_schema_not_mutated_across_calls(self, mock_call, mock_prompt, mock_schema): """load_schema 반환값이 호출 간에 변형되지 않는지 검증 (deepcopy).""" base_schema = { "type": "object", "properties": { "scenes": { "type": "array", "items": { "type": "object", "properties": { "scene_index": {"type": "integer"}, "scene_type": {"type": "string"}, "present_entity_ids": {"type": "array", "items": {"type": "string"}}, "audio_entity_ids": {"type": "array", "items": {"type": "string"}}, "hallucination_entity_ids": {"type": "array", "items": {"type": "string"}}, "not_present": { "type": "array", "items": { "type": "object", "properties": { "entity_id": {"type": "string"}, "reason": {"type": "string"}, }, }, }, }, }, } }, } original_snapshot = copy.deepcopy(base_schema) mock_prompt.return_value = "시스템" mock_schema.return_value = base_schema mock_call.return_value = SAMPLE_LLM_RESPONSE from app.modules.pipeline.scene_director_v2 import direct_scenes direct_scenes(SAMPLE_SEGMENTS, SAMPLE_FULLTEXT, SAMPLE_ENTITIES) # The base_schema returned by load_schema should NOT be mutated assert base_schema == original_snapshot @patch("app.modules.pipeline.scene_director_v2.load_schema") @patch("app.modules.pipeline.scene_director_v2.load_prompt") @patch("app.modules.pipeline.scene_director_v2.call_structured") def test_loads_prompt_and_schema_for_scene_director(self, mock_call, mock_prompt, mock_schema): """load_prompt와 load_schema가 scene_director 모듈로 호출되는지 검증.""" mock_prompt.return_value = "시스템" mock_schema.return_value = { "type": "object", "properties": { "scenes": { "type": "array", "items": { "type": "object", "properties": { "scene_index": {"type": "integer"}, "scene_type": {"type": "string"}, "present_entity_ids": {"type": "array", "items": {"type": "string"}}, "audio_entity_ids": {"type": "array", "items": {"type": "string"}}, "hallucination_entity_ids": {"type": "array", "items": {"type": "string"}}, "not_present": {"type": "array", "items": {"type": "object", "properties": {"entity_id": {"type": "string"}, "reason": {"type": "string"}}}}, }, }, } }, } mock_call.return_value = copy.deepcopy(SAMPLE_LLM_RESPONSE) from app.modules.pipeline.scene_director_v2 import direct_scenes # 세그먼트가 없으면 LLM 을 태우지 않고 즉시 반환하므로 로더도 안 탄다 — # 로더 호출 계약은 실제 입력이 있을 때로 확인한다. direct_scenes(SAMPLE_SEGMENTS, SAMPLE_FULLTEXT, {}) mock_prompt.assert_called_once_with("scene_director", "system") mock_schema.assert_called_once_with("scene_director", "analyze_schema") # ── SceneDirectorStep 단위 테스트 ── class TestSceneDirectorStep: def _make_step(self, fake_db, checkpoints=None): """SceneDirectorStep 인스턴스를 생성하는 헬퍼.""" with patch("app.core.steps.director_steps.StepRunner.__init__", return_value=None): from app.core.steps.director_steps import SceneDirectorStep step = SceneDirectorStep.__new__(SceneDirectorStep) step.step_id = "scene_director" 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_director_v2.direct_scenes") def test_execute_calls_direct_scenes(self, mock_direct, fake_db=None): """_execute가 direct_scenes를 호출하는지 검증.""" fake_db = MagicMock() checkpoints = { "text_cleanup": {"data": {"cleaned_text": "정리된 텍스트"}}, "scene_split": {"data": {"segments": SAMPLE_SEGMENTS}}, "entity_t2i": {"data": SAMPLE_ENTITIES}, } step = self._make_step(fake_db, checkpoints) mock_direct.return_value = SAMPLE_LLM_RESPONSE result = step._execute() mock_direct.assert_called_once() assert result["completed_count"] == 2 assert result["applicable_count"] == 2 assert result["failed_count"] == 0 assert result["data"] == SAMPLE_LLM_RESPONSE @patch("app.modules.pipeline.scene_director_v2.direct_scenes") def test_execute_falls_back_to_scene_save(self, mock_direct): """scene_split이 없으면 scene_save 세그먼트를 사용.""" fake_db = MagicMock() checkpoints = { "text_cleanup": {"data": {"cleaned_text": "텍스트"}}, "scene_save": {"data": {"segments": SAMPLE_SEGMENTS}}, "entity_t2i": {"data": SAMPLE_ENTITIES}, } step = self._make_step(fake_db, checkpoints) mock_direct.return_value = SAMPLE_LLM_RESPONSE result = step._execute() segments_passed = mock_direct.call_args.args[0] assert segments_passed == SAMPLE_SEGMENTS @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_director_v2.direct_scenes") def test_execute_empty_entities(self, mock_direct): """entity_t2i 체크포인트가 없으면 빈 엔티티로 호출.""" fake_db = MagicMock() checkpoints = { "text_cleanup": {"data": {"cleaned_text": "텍스트"}}, "scene_split": {"data": {"segments": SAMPLE_SEGMENTS}}, } step = self._make_step(fake_db, checkpoints) mock_direct.return_value = {"scenes": []} result = step._execute() entities_passed = mock_direct.call_args.args[2] assert entities_passed == {} assert result["completed_count"] == 0