"""scene_summary 모듈 테스트 -- summarize_scene / summarize_scenes + ThreadPool 검증.""" from unittest.mock import patch, call # ── summarize_scene 단위 테스트 ── def test_summarize_scene_returns_string(): """summarize_scene이 scene_summary 문자열을 반환.""" from app.modules.pipeline.scene_summarizer import summarize_scene with patch("app.modules.pipeline.scene_summarizer.call_structured") as mock: mock.return_value = {"scene_summary": "인물이 카페에서 대화한다."} result = summarize_scene( scene_text="씬 텍스트", prev_scenes_text="(없음)", episode_summary="에피소드 요약", visual_rules="규칙", ) assert isinstance(result, str) assert result == "인물이 카페에서 대화한다." def test_summarize_scene_returns_empty_on_missing_key(): """call_structured 응답에 scene_summary 키 없으면 빈 문자열 반환.""" from app.modules.pipeline.scene_summarizer import summarize_scene with patch("app.modules.pipeline.scene_summarizer.call_structured") as mock: mock.return_value = {} result = summarize_scene( scene_text="텍스트", prev_scenes_text="(없음)", episode_summary="요약", visual_rules="규칙", ) assert result == "" def test_summarize_scene_calls_call_structured_with_correct_step(): """call_structured가 step='scene_summary'로 호출되는지 검증.""" from app.modules.pipeline.scene_summarizer import summarize_scene with patch("app.modules.pipeline.scene_summarizer.call_structured") as mock: mock.return_value = {"scene_summary": "요약"} summarize_scene( scene_text="텍스트", prev_scenes_text="이전", episode_summary="에피소드", visual_rules="규칙", max_length=200, ) mock.assert_called_once() _, kwargs = mock.call_args assert kwargs["step"] == "scene_summary" assert kwargs["schema_name"] == "scene_summary" def test_summarize_scene_loads_external_prompt(): """summarize_scene이 load_prompt와 load_schema를 호출하는지 확인.""" from app.modules.pipeline.scene_summarizer import summarize_scene with patch("app.modules.pipeline.scene_summarizer.call_structured") as mock_call, \ patch("app.modules.pipeline.scene_summarizer.load_prompt") as mock_prompt, \ patch("app.modules.pipeline.scene_summarizer.load_schema") as mock_schema: mock_call.return_value = {"scene_summary": "ok"} mock_prompt.return_value = "시스템 프롬프트" mock_schema.return_value = {"type": "object", "properties": {}} summarize_scene( scene_text="텍스트", prev_scenes_text="(없음)", episode_summary="요약", visual_rules="규칙", max_length=300, ) mock_prompt.assert_called_once_with("scene_summary", "system", max_length=300) mock_schema.assert_called_once_with("scene_summary", "summary_schema") def test_summarize_scene_includes_all_context_in_user_prompt(): """user_prompt에 에피소드 요약, 시각적 규칙, 이전 씬, 현재 씬이 모두 포함.""" from app.modules.pipeline.scene_summarizer import summarize_scene with patch("app.modules.pipeline.scene_summarizer.call_structured") as mock: mock.return_value = {"scene_summary": "ok"} summarize_scene( scene_text="현재_씬_텍스트", prev_scenes_text="이전_씬_텍스트", episode_summary="에피소드_요약_텍스트", visual_rules="시각적_규칙_텍스트", ) user_prompt = mock.call_args.kwargs.get("user_prompt", "") assert "현재_씬_텍스트" in user_prompt assert "이전_씬_텍스트" in user_prompt assert "에피소드_요약_텍스트" in user_prompt assert "시각적_규칙_텍스트" in user_prompt # ── summarize_scenes 병렬 테스트 ── def test_summarize_scenes_returns_list_sorted_by_index(): """summarize_scenes가 scene_index 순으로 정렬된 리스트를 반환.""" from app.modules.pipeline.scene_summarizer import summarize_scenes segments = [ {"scene_index": 1, "start_char": 0, "end_char": 10}, {"scene_index": 2, "start_char": 10, "end_char": 20}, {"scene_index": 3, "start_char": 20, "end_char": 30}, ] fulltext = "0123456789" * 3 with patch("app.modules.pipeline.scene_summarizer.call_structured") as mock: mock.return_value = {"scene_summary": "요약 텍스트"} results = summarize_scenes( segments=segments, fulltext=fulltext, episode_summary="에피소드 요약", visual_rules="규칙", ) assert len(results) == 3 assert results[0]["scene_index"] == 1 assert results[1]["scene_index"] == 2 assert results[2]["scene_index"] == 3 for r in results: assert r["scene_summary"] == "요약 텍스트" def test_summarize_scenes_uses_threadpool(): """summarize_scenes가 ThreadPoolExecutor를 사용하는지 확인.""" from app.modules.pipeline.scene_summarizer import summarize_scenes segments = [ {"scene_index": 1, "start_char": 0, "end_char": 5}, {"scene_index": 2, "start_char": 5, "end_char": 10}, ] fulltext = "0123456789" with patch("app.modules.pipeline.scene_summarizer.call_structured") as mock_call, \ patch("app.modules.pipeline.scene_summarizer.ThreadPoolExecutor") as mock_pool_cls: mock_call.return_value = {"scene_summary": "요약"} # ThreadPoolExecutor context manager mock 구성 mock_pool = mock_pool_cls.return_value.__enter__.return_value # submit이 실제 함수를 호출하도록 설정 from concurrent.futures import Future futures = [] for i, seg in enumerate(segments): f = Future() f.set_result({"scene_index": seg["scene_index"], "scene_summary": "요약"}) futures.append(f) mock_pool.submit.side_effect = futures results = summarize_scenes( segments=segments, fulltext=fulltext, episode_summary="요약", visual_rules="규칙", max_workers=3, ) mock_pool_cls.assert_called_once_with(max_workers=3) assert mock_pool.submit.call_count == 2 def test_summarize_scenes_prev_3_context(): """이전 3개 씬 텍스트가 올바르게 전달되는지 확인.""" from app.modules.pipeline.scene_summarizer import summarize_scene, summarize_scenes segments = [ {"scene_index": 1, "start_char": 0, "end_char": 2}, {"scene_index": 2, "start_char": 2, "end_char": 4}, {"scene_index": 3, "start_char": 4, "end_char": 6}, {"scene_index": 4, "start_char": 6, "end_char": 8}, {"scene_index": 5, "start_char": 8, "end_char": 10}, ] fulltext = "AABBCCDDEE" call_args_list = [] def fake_call(**kwargs): call_args_list.append(kwargs) return {"scene_summary": "요약"} with patch("app.modules.pipeline.scene_summarizer.call_structured", side_effect=fake_call): results = summarize_scenes( segments=segments, fulltext=fulltext, episode_summary="에피소드", visual_rules="규칙", max_workers=1, # 순차 실행으로 순서 보장 ) assert len(results) == 5 # 씬 1 (index 0): 이전 씬 없음 → "(없음)" scene1_call = [c for c in call_args_list if "현재 씬:\nAA" in c["user_prompt"]][0] assert "(없음)" in scene1_call["user_prompt"] # 씬 4 (index 3): 이전 3개 씬 = 씬1(AA), 씬2(BB), 씬3(CC) scene4_call = [c for c in call_args_list if "현재 씬:\nDD" in c["user_prompt"]][0] assert "AA" in scene4_call["user_prompt"] assert "BB" in scene4_call["user_prompt"] assert "CC" in scene4_call["user_prompt"] # 씬 5 (index 4): 이전 3개 씬 = 씬2(BB), 씬3(CC), 씬4(DD) scene5_call = [c for c in call_args_list if "현재 씬:\nEE" in c["user_prompt"]][0] assert "BB" in scene5_call["user_prompt"] assert "CC" in scene5_call["user_prompt"] assert "DD" in scene5_call["user_prompt"] # 씬1(AA)은 이전 3개에 포함되지 않음 # (user_prompt에 "에피소드" 등에 AA가 없으므로 prev 부분만 체크) def test_summarize_scenes_handles_failure_gracefully(): """개별 씬 요약 실패 시 빈 문자열로 대체되고 나머지는 정상 반환.""" from app.modules.pipeline.scene_summarizer import summarize_scenes segments = [ {"scene_index": 1, "start_char": 0, "end_char": 5}, {"scene_index": 2, "start_char": 5, "end_char": 10}, {"scene_index": 3, "start_char": 10, "end_char": 15}, ] fulltext = "AAAAABBBBBCCCCC" call_count = 0 def fake_call(**kwargs): nonlocal call_count call_count += 1 # "현재 씬" 바로 뒤의 텍스트만 검사 (이전 씬 컨텍스트 제외) user = kwargs["user_prompt"] current_marker = "현재 씬:\n" current_idx = user.find(current_marker) if current_idx >= 0: current_text = user[current_idx + len(current_marker):] if current_text.startswith("BBBBB"): raise RuntimeError("LLM 실패") return {"scene_summary": "정상 요약"} with patch("app.modules.pipeline.scene_summarizer.call_structured", side_effect=fake_call): results = summarize_scenes( segments=segments, fulltext=fulltext, episode_summary="요약", visual_rules="규칙", max_workers=1, ) assert len(results) == 3 assert results[0]["scene_summary"] == "정상 요약" assert results[1]["scene_summary"] == "" # 실패한 씬 assert results[2]["scene_summary"] == "정상 요약"