"""LocationConsistencyStep 단위 테스트.""" from __future__ import annotations import json from pathlib import Path from unittest.mock import MagicMock, patch import pytest from app.core.steps.location_consistency_step import LocationConsistencyStep def _write_cp(tmp_path: Path, pid: str, eid: str, step_id: str, payload: dict): cp_dir = tmp_path / pid / "checkpoints" / "episodes" / eid / step_id cp_dir.mkdir(parents=True, exist_ok=True) (cp_dir / "manifest.json").write_text(json.dumps(payload), encoding="utf-8") @pytest.fixture def step_fixture(tmp_path: Path, monkeypatch): monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) pid, eid = "p1", "e1" step = LocationConsistencyStep.__new__(LocationConsistencyStep) step.project_id = pid step.episode_id = eid step.db = MagicMock() step.project_config = {} step.build_opik_metadata = lambda tags=None: {"tags": tags or []} return step, tmp_path, pid, eid def test_no_locations_returns_empty(step_fixture): step, tmp_path, pid, eid = step_fixture _write_cp(tmp_path, pid, eid, "entity_merge", {"data": {"locations": []}}) result = step._execute() assert result["applicable_count"] == 0 assert result["completed_count"] == 0 assert result["data"]["locations"] == [] def test_missing_entity_merge_raises(step_fixture): step, tmp_path, pid, eid = step_fixture # entity_merge 없음 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 이후 재평가.") def test_processes_all_locations_with_llm_success(step_fixture): step, tmp_path, pid, eid = step_fixture _write_cp(tmp_path, pid, eid, "entity_merge", {"data": { "locations": [ {"short_id": "L01", "name": "항구"}, {"short_id": "L02", "name": "옥탑방"}, ], }}) _write_cp(tmp_path, pid, eid, "entity_detail", {"data": { "entity_details": { "항구:location": {"description": "작은 어촌 포구", "visual_traits": ["bamboo pier"]}, "옥탑방:location": {"description": "좁은 옥탑방", "visual_traits": ["single window"]}, }, }}) _write_cp(tmp_path, pid, eid, "scene_save", {"data": { "segments": [ {"scene_index": 1, "heading": "항구 - 낮", "text": "항구에 배가 정박."}, {"scene_index": 2, "heading": "옥탑방 - 밤", "text": "옥탑방에 불빛이 새어나온다."}, ], }}) # LLM이 각 location마다 locations 배열 반환 def _fake_call(**kwargs): sn = kwargs["schema_name"] if "L01" in sn: return {"locations": [{ "location_id": "L01", "name": "항구", "fixed_visual_description": "A small harbor with wooden piers and rusted dock cleats.", }]} return {"locations": [{ "location_id": "L02", "name": "옥탑방", "fixed_visual_description": "A narrow rooftop room with a single square window.", }]} with patch("app.core.steps.location_consistency_step.call_structured", side_effect=_fake_call), \ patch("app.core.steps.location_consistency_step.load_prompt", return_value="sys"), \ patch("app.core.steps.location_consistency_step.load_schema", return_value={}): result = step._execute() assert result["completed_count"] == 2 assert result["failed_count"] == 0 locs = result["data"]["locations"] assert [l["location_id"] for l in locs] == ["L01", "L02"] assert "harbor" in locs[0]["fixed_visual_description"] assert "rooftop" in locs[1]["fixed_visual_description"] @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_llm_returns_empty_falls_back_to_gpt(step_fixture): step, tmp_path, pid, eid = step_fixture _write_cp(tmp_path, pid, eid, "entity_merge", {"data": { "locations": [{"short_id": "L01", "name": "항구"}], }}) call_count = {"n": 0} def _fake_call(**kwargs): call_count["n"] += 1 if call_count["n"] == 1: # 첫 호출(primary): 빈 locations 반환 → ValueError 트리거 return {"locations": []} # GPT fallback은 성공 return {"locations": [{ "location_id": "L01", "name": "항구", "fixed_visual_description": "GPT fallback description.", }]} with patch("app.core.steps.location_consistency_step.call_structured", side_effect=_fake_call), \ patch("app.core.steps.location_consistency_step.load_prompt", return_value="sys"), \ patch("app.core.steps.location_consistency_step.load_schema", return_value={}): result = step._execute() assert call_count["n"] == 2 # primary 실패 → gpt fallback locs = result["data"]["locations"] assert locs[0]["fixed_visual_description"] == "GPT fallback description." def test_all_attempts_fail_returns_entity_detail_fallback(step_fixture): """LLM 완전 실패 + entity_detail description 있음: 기존 설명으로 보존.""" step, tmp_path, pid, eid = step_fixture _write_cp(tmp_path, pid, eid, "entity_merge", {"data": { "locations": [{"short_id": "L01", "name": "항구"}], }}) _write_cp(tmp_path, pid, eid, "entity_detail", {"data": { "entity_details": {"항구:location": {"description": "기존 설명", "visual_traits": []}}, }}) def _fake_call(**kwargs): raise RuntimeError("LLM error") with patch("app.core.steps.location_consistency_step.call_structured", side_effect=_fake_call), \ patch("app.core.steps.location_consistency_step.load_prompt", return_value="sys"), \ patch("app.core.steps.location_consistency_step.load_schema", return_value={}): result = step._execute() # 기존 설명이 남아 있으므로 processed로 카운트되지만 analysis_summary가 실패를 표시 locs = result["data"]["locations"] assert locs[0]["fixed_visual_description"] == "기존 설명" assert "실패" in locs[0].get("analysis_summary", "") def test_all_attempts_fail_without_entity_detail_counted_as_failed(step_fixture): """LLM 완전 실패 + entity_detail 없음: failed_count=1, description은 빈 문자열.""" step, tmp_path, pid, eid = step_fixture _write_cp(tmp_path, pid, eid, "entity_merge", {"data": { "locations": [{"short_id": "L01", "name": "항구"}], }}) # entity_detail 체크포인트 아예 없음 def _fake_call(**kwargs): raise RuntimeError("LLM error") with patch("app.core.steps.location_consistency_step.call_structured", side_effect=_fake_call), \ patch("app.core.steps.location_consistency_step.load_prompt", return_value="sys"), \ patch("app.core.steps.location_consistency_step.load_schema", return_value={}): result = step._execute() assert result["failed_count"] == 1 assert result["data"]["locations"][0]["fixed_visual_description"] == "" @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_resume_mode_skips_completed_locations(step_fixture): step, tmp_path, pid, eid = step_fixture _write_cp(tmp_path, pid, eid, "entity_merge", {"data": { "locations": [ {"short_id": "L01", "name": "항구"}, {"short_id": "L02", "name": "옥탑방"}, ], }}) # 이미 L01 완료된 기존 체크포인트 _write_cp(tmp_path, pid, eid, "location_consistency", {"data": { "locations": [ {"location_id": "L01", "name": "항구", "fixed_visual_description": "Previously computed harbor."}, ], }}) def _fake_call(**kwargs): sn = kwargs["schema_name"] assert "L01" not in sn, f"L01은 재시도하면 안 됨: {sn}" return {"locations": [{ "location_id": "L02", "name": "옥탑방", "fixed_visual_description": "New rooftop description.", }]} with patch("app.core.steps.location_consistency_step.call_structured", side_effect=_fake_call), \ patch("app.core.steps.location_consistency_step.load_prompt", return_value="sys"), \ patch("app.core.steps.location_consistency_step.load_schema", return_value={}): result = step._execute(mode="resume") locs = {l["location_id"]: l for l in result["data"]["locations"]} assert locs["L01"]["fixed_visual_description"] == "Previously computed harbor." # 유지 assert locs["L02"]["fixed_visual_description"] == "New rooftop description." # 새로 생성 @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_location_id_enforced_from_input_not_llm(step_fixture): """LLM이 location_id를 잘못 써도 입력 short_id로 강제 덮어쓰기.""" step, tmp_path, pid, eid = step_fixture _write_cp(tmp_path, pid, eid, "entity_merge", {"data": { "locations": [{"short_id": "L05", "name": "숲속"}], }}) def _fake_call(**kwargs): return {"locations": [{ "location_id": "L_WRONG", # LLM이 잘못 쓴 ID "name": "잘못된 이름", "fixed_visual_description": "Dense forest.", }]} with patch("app.core.steps.location_consistency_step.call_structured", side_effect=_fake_call), \ patch("app.core.steps.location_consistency_step.load_prompt", return_value="sys"), \ patch("app.core.steps.location_consistency_step.load_schema", return_value={}): result = step._execute() loc = result["data"]["locations"][0] assert loc["location_id"] == "L05" # 입력으로 강제 assert loc["name"] == "숲속" # 입력으로 강제 assert loc["fixed_visual_description"] == "Dense forest."