"""scene_detail (Step 16) 단위 테스트. 검증 항목: - visible_entities를 director 체크포인트에서 구축 (LLM 아님) - 병렬 호출로 모든 세그먼트 처리 - 실패 씬 카운트 정확 """ import json from unittest.mock import MagicMock, patch import pytest # -- Fixtures -- @pytest.fixture() def fake_db(): db = MagicMock() db.execute.return_value.fetchone.return_value = None return db # -- Sample Data -- SAMPLE_SEGMENTS = [ {"scene_index": 1, "heading": "씬1", "start_char": 0, "end_char": 50, "length": 50}, {"scene_index": 2, "heading": "씬2", "start_char": 50, "end_char": 100, "length": 50}, {"scene_index": 3, "heading": "씬3", "start_char": 100, "end_char": 150, "length": 50}, ] SAMPLE_DIRECTOR_SCENES = [ {"scene_index": 1, "present_entity_ids": ["C01", "L01"]}, {"scene_index": 2, "present_entity_ids": ["C01", "C02", "L02"]}, {"scene_index": 3, "present_entity_ids": ["C03"]}, ] SAMPLE_DEPENDENCIES = [ {"scene_index": 1, "location_refs": [], "character_refs": []}, {"scene_index": 2, "location_refs": [1], "character_refs": [1]}, {"scene_index": 3, "location_refs": [], "character_refs": [2]}, ] FULLTEXT = "A" * 50 + "B" * 50 + "C" * 50 def _make_step(fake_db, checkpoints=None, monkeypatch=None): """SceneDetailStep 인스턴스를 생성하는 헬퍼.""" # G3.2: 본 fixture 들은 visible_entities/summary/dependency 검증 목적이라 # background_prompt cp 의존성이 없어야 함 — bg-mode off 강제. if monkeypatch is not None: monkeypatch.setattr( "app.core.config.settings.background_mode", "off", raising=False, ) with patch("app.core.steps.detail_steps.StepRunner.__init__", return_value=None): from app.core.steps.detail_steps import SceneDetailStep step = SceneDetailStep.__new__(SceneDetailStep) step.step_id = "scene_detail" step.project_id = "proj1" step.episode_id = "ep1" step.db = fake_db step.project_config = {} step.manifest = {"applicability": "always"} step.opik_context = {} if checkpoints: step._load_prev_checkpoint = lambda sid: checkpoints.get(sid) else: step._load_prev_checkpoint = lambda sid: None step._load_cleaned_text = lambda: FULLTEXT return step class TestSceneDetailStep: def test_visible_entities_from_director_not_llm(self, fake_db, monkeypatch): """visible_entities는 director 체크포인트의 present_entity_ids에서 구축. LLM 응답이 다른 엔티티를 반환해도 director 데이터가 최종.""" checkpoints = { "scene_save": {"data": {"segments": SAMPLE_SEGMENTS}}, "scene_director": {"data": {"scenes": SAMPLE_DIRECTOR_SCENES}}, "scene_dependency": {"data": {"dependencies": SAMPLE_DEPENDENCIES}}, "outlook_extraction": {"data": {}}, "scene_cinematography": {"data": {}}, "entity_t2i": {"data": {}}, "scene_summary": {"data": {"summaries": []}}, } step = _make_step(fake_db, checkpoints, monkeypatch=monkeypatch) # LLM returns different entities -- should be overridden def _fresh_response(**kwargs): return { "beat_title": "테스트", "representative_moment": "순간", "t2i_variations": [], "scene_type": "normal", "dependent_scene_index": -1, "dependency_reason": "", } with patch("app.core.steps.detail_steps.load_prompt", return_value="sys"), \ patch("app.core.steps.detail_steps.load_schema", return_value={}), \ patch("app.core.steps.detail_steps.call_structured", side_effect=_fresh_response): result = step._execute() scenes = result["data"]["scenes"] assert len(scenes) == 3 # visible_entities comes from director, not LLM scene1 = next(s for s in scenes if s["scene_index"] == 1) scene2 = next(s for s in scenes if s["scene_index"] == 2) scene3 = next(s for s in scenes if s["scene_index"] == 3) assert scene1["visible_entities"] == ["C01", "L01"] assert scene2["visible_entities"] == ["C01", "C02", "L02"] assert scene3["visible_entities"] == ["C03"] @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 이후 재평가.") def test_all_segments_processed(self, fake_db, monkeypatch): """모든 세그먼트가 처리되어 결과에 포함.""" checkpoints = { "scene_save": {"data": {"segments": SAMPLE_SEGMENTS}}, "scene_director": {"data": {"scenes": SAMPLE_DIRECTOR_SCENES}}, "scene_dependency": {"data": {"dependencies": []}}, "outlook_extraction": {"data": {}}, "scene_cinematography": {"data": {}}, "entity_t2i": {"data": {}}, "scene_summary": {"data": {"summaries": []}}, } step = _make_step(fake_db, checkpoints, monkeypatch=monkeypatch) def _fresh_response(**kwargs): return { "beat_title": "ok", "representative_moment": "m", "t2i_variations": [], "scene_type": "normal", "dependent_scene_index": -1, "dependency_reason": "", } with patch("app.core.steps.detail_steps.load_prompt", return_value="sys"), \ patch("app.core.steps.detail_steps.load_schema", return_value={}), \ patch("app.core.steps.detail_steps.call_structured", side_effect=_fresh_response): result = step._execute() assert result["completed_count"] == 3 assert result["applicable_count"] == 3 assert result["failed_count"] == 0 def test_results_sorted_by_scene_index(self, fake_db, monkeypatch): """결과가 scene_index 순서로 정렬.""" checkpoints = { "scene_save": {"data": {"segments": SAMPLE_SEGMENTS}}, "scene_director": {"data": {"scenes": SAMPLE_DIRECTOR_SCENES}}, "scene_dependency": {"data": {"dependencies": []}}, "outlook_extraction": {"data": {}}, "scene_cinematography": {"data": {}}, "entity_t2i": {"data": {}}, "scene_summary": {"data": {"summaries": []}}, } step = _make_step(fake_db, checkpoints, monkeypatch=monkeypatch) def _fresh_response(**kwargs): return { "beat_title": "ok", "representative_moment": "m", "t2i_variations": [], "scene_type": "normal", "dependent_scene_index": -1, "dependency_reason": "", } with patch("app.core.steps.detail_steps.load_prompt", return_value="sys"), \ patch("app.core.steps.detail_steps.load_schema", return_value={}), \ patch("app.core.steps.detail_steps.call_structured", side_effect=_fresh_response): result = step._execute() indices = [s["scene_index"] for s in result["data"]["scenes"]] assert indices == [1, 2, 3] @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 이후 재평가.") def test_failed_scenes_counted(self, fake_db, monkeypatch): """LLM 호출 실패 시 failed_count 증가.""" checkpoints = { "scene_save": {"data": {"segments": SAMPLE_SEGMENTS}}, "scene_director": {"data": {"scenes": SAMPLE_DIRECTOR_SCENES}}, "scene_dependency": {"data": {"dependencies": []}}, "outlook_extraction": {"data": {}}, "scene_cinematography": {"data": {}}, "entity_t2i": {"data": {}}, "scene_summary": {"data": {"summaries": []}}, } step = _make_step(fake_db, checkpoints, monkeypatch=monkeypatch) call_count = 0 def _failing_call(**kwargs): nonlocal call_count call_count += 1 if call_count == 2: raise RuntimeError("LLM error") return { "beat_title": "ok", "representative_moment": "m", "t2i_variations": [], "scene_type": "normal", "dependent_scene_index": -1, "dependency_reason": "", } with patch("app.core.steps.detail_steps.load_prompt", return_value="sys"), \ patch("app.core.steps.detail_steps.load_schema", return_value={}), \ patch("app.core.steps.detail_steps.call_structured", side_effect=_failing_call): result = step._execute() assert result["completed_count"] == 2 assert result["failed_count"] == 1 assert result["applicable_count"] == 3 def test_summary_included_in_user_prompt(self, fake_db, monkeypatch): """scene_summary 체크포인트의 요약이 user_prompt에 포함.""" checkpoints = { "scene_save": {"data": {"segments": [SAMPLE_SEGMENTS[0]]}}, "scene_director": {"data": {"scenes": [SAMPLE_DIRECTOR_SCENES[0]]}}, "scene_dependency": {"data": {"dependencies": []}}, "outlook_extraction": {"data": {}}, "scene_cinematography": {"data": {}}, "entity_t2i": {"data": {}}, "scene_summary": {"data": {"summaries": [ {"scene_index": 1, "scene_summary": "인물A가 카페에서 대화"}, ]}}, } step = _make_step(fake_db, checkpoints, monkeypatch=monkeypatch) captured_prompts = [] def _capture_call(**kwargs): captured_prompts.append(kwargs.get("user_prompt", "")) return { "beat_title": "ok", "representative_moment": "m", "t2i_variations": [], "scene_type": "normal", "dependent_scene_index": -1, "dependency_reason": "", } with patch("app.core.steps.detail_steps.load_prompt", return_value="sys"), \ patch("app.core.steps.detail_steps.load_schema", return_value={}), \ patch("app.core.steps.detail_steps.call_structured", side_effect=_capture_call): step._execute() assert len(captured_prompts) == 1 assert "인물A가 카페에서 대화" in captured_prompts[0] @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 이후 재평가.") def test_dependency_info_in_user_prompt(self, fake_db, monkeypatch): """scene_dependency 정보가 user_prompt에 포함.""" checkpoints = { "scene_save": {"data": {"segments": [SAMPLE_SEGMENTS[1]]}}, "scene_director": {"data": {"scenes": [SAMPLE_DIRECTOR_SCENES[1]]}}, "scene_dependency": {"data": {"dependencies": [SAMPLE_DEPENDENCIES[1]]}}, "outlook_extraction": {"data": {}}, "scene_cinematography": {"data": {}}, "entity_t2i": {"data": {}}, "scene_summary": {"data": {"summaries": []}}, } step = _make_step(fake_db, checkpoints, monkeypatch=monkeypatch) captured_prompts = [] def _capture_call(**kwargs): captured_prompts.append(kwargs.get("user_prompt", "")) return { "beat_title": "ok", "representative_moment": "m", "t2i_variations": [], "scene_type": "normal", "dependent_scene_index": -1, "dependency_reason": "", } with patch("app.core.steps.detail_steps.load_prompt", return_value="sys"), \ patch("app.core.steps.detail_steps.load_schema", return_value={}), \ patch("app.core.steps.detail_steps.call_structured", side_effect=_capture_call): step._execute() assert len(captured_prompts) == 1 assert "[1]" in captured_prompts[0] # location_refs or character_refs def test_no_segments_returns_empty(self, fake_db, monkeypatch): """세그먼트가 없으면 빈 결과 반환.""" checkpoints = { "scene_save": {"data": {"segments": []}}, "scene_director": {"data": {"scenes": []}}, "scene_dependency": {"data": {"dependencies": []}}, "outlook_extraction": {"data": {}}, "scene_cinematography": {"data": {}}, "entity_t2i": {"data": {}}, "scene_summary": {"data": {"summaries": []}}, } step = _make_step(fake_db, checkpoints, monkeypatch=monkeypatch) with patch("app.core.steps.detail_steps.load_prompt", return_value="sys"), \ patch("app.core.steps.detail_steps.load_schema", return_value={}): result = step._execute() assert result["completed_count"] == 0 assert result["applicable_count"] == 0 assert result["failed_count"] == 0 assert result["data"]["scenes"] == []