"""Phase 5 — sequential pipeline 통합 테스트 (Task 8). 5 시나리오로 4-step 흐름의 모드별 동작을 mock E2E 검증: background_planner → location_floor_plan → background_chain_planning → background_chain_render 각 step의 _execute()를 직접 호출하되, LLM/이미지 호출은 module 경계에서 mock. 중간 step의 결과(체크포인트 manifest.json)는 tmp_path의 가짜 파일로 미리 작성하거나, 이전 step _execute의 result["data"]를 다음 step이 읽도록 직접 manifest를 기록한다. CLAUDE.md 규칙 준수: - 시나리오 텍스트/씬 텍스트 truncation 금지 — fixture는 짧은 자체 fixture-only 텍스트 사용. - prompt 본문은 LLM mock이라 prompt 내용 검사 X (호출 수/결과만 검증). """ from __future__ import annotations import base64 import json from pathlib import Path from typing import Any, Dict, List from unittest.mock import MagicMock, patch import pytest # ────────────────────────────────────────────── # 공통 fixture helpers # ────────────────────────────────────────────── PROJECT_ID = "p" EPISODE_ID = "eid" def _b64_png() -> str: """fake non-empty PNG b64 (>=1024 bytes after decode).""" return base64.b64encode(b"PNG" * 1024).decode() def _ckpt_dir(tmp_path: Path, step_id: str) -> Path: d = tmp_path / PROJECT_ID / "checkpoints" / "episodes" / EPISODE_ID / step_id d.mkdir(parents=True, exist_ok=True) return d def _write_checkpoint(tmp_path: Path, step_id: str, data: Dict[str, Any]) -> None: """tmp_path/p/checkpoints/episodes/eid/{step_id}/manifest.json 작성.""" d = _ckpt_dir(tmp_path, step_id) (d / "manifest.json").write_text(json.dumps({"data": data}), encoding="utf-8") def _write_planner_dependencies(tmp_path: Path) -> None: """background_planner step이 읽는 6개 의존 체크포인트를 작성.""" # shot_validator: 6 location, 30 selected shot 분포 scenes = [] # L01 (indoor) — scene 5, 6 — 각 5 shot = 10 shot for si in (5, 6): scenes.append({"scene_index": si, "shots": [ {"shot_index": k, "description": f"L01 scene{si} shot{k}"} for k in range(1, 6) ]}) # L02 (outdoor) — scene 7, 8 — 각 4 shot = 8 shot for si in (7, 8): scenes.append({"scene_index": si, "shots": [ {"shot_index": k, "description": f"L02 scene{si} shot{k}"} for k in range(1, 5) ]}) # L03 (indoor) — scene 9 — 4 shot scenes.append({"scene_index": 9, "shots": [ {"shot_index": k, "description": f"L03 shot{k}"} for k in range(1, 5) ]}) # L04 (indoor) — scene 10 — 4 shot scenes.append({"scene_index": 10, "shots": [ {"shot_index": k, "description": f"L04 shot{k}"} for k in range(1, 5) ]}) # L05 (outdoor) — scene 11 — 2 shot (저빈도) scenes.append({"scene_index": 11, "shots": [ {"shot_index": k, "description": f"L05 shot{k}"} for k in range(1, 3) ]}) # L06 (indoor) — scene 12 — 2 shot (저빈도) scenes.append({"scene_index": 12, "shots": [ {"shot_index": k, "description": f"L06 shot{k}"} for k in range(1, 3) ]}) _write_checkpoint(tmp_path, "shot_validator", {"scenes": scenes}) # shot_selection: 모든 shot select sel_scenes = [ {"scene_index": s["scene_index"], "selected_shot_indices": [sh["shot_index"] for sh in s["shots"]]} for s in scenes ] _write_checkpoint(tmp_path, "shot_selection", {"scenes": sel_scenes}) # scene_director: scene_index → primary_location director_scenes = [ {"scene_index": 5, "primary_location": "L01"}, {"scene_index": 6, "primary_location": "L01"}, {"scene_index": 7, "primary_location": "L02"}, {"scene_index": 8, "primary_location": "L02"}, {"scene_index": 9, "primary_location": "L03"}, {"scene_index": 10, "primary_location": "L04"}, {"scene_index": 11, "primary_location": "L05"}, {"scene_index": 12, "primary_location": "L06"}, ] _write_checkpoint(tmp_path, "scene_director", {"scenes": director_scenes}) # entity_merge locations = [ {"short_id": f"L0{i}", "name": f"loc-{i}"} for i in range(1, 7) ] _write_checkpoint(tmp_path, "entity_merge", {"locations": locations}) # entity_detail detail_locations = [ {"short_id": "L01", "kind": "indoor", "description": "main interior A"}, {"short_id": "L02", "kind": "outdoor", "description": "outdoor street"}, {"short_id": "L03", "kind": "indoor", "description": "indoor B"}, {"short_id": "L04", "kind": "indoor", "description": "indoor C"}, {"short_id": "L05", "kind": "outdoor", "description": "outdoor field"}, {"short_id": "L06", "kind": "indoor", "description": "indoor D"}, ] _write_checkpoint(tmp_path, "entity_detail", {"locations": detail_locations}) # visual_world_rules _write_checkpoint(tmp_path, "visual_world_rules", { "rules": [{"rule_type": "interior", "description": "주거", "visual_guideline": "warm"}], "era": "현대", "region": "서울", }) def _ok_planner_output() -> Dict[str, Any]: """LLM이 반환할 invariant-만족 plan: 2 floor_plan, 4 chain_bg group.""" return { "rationale_summary": "okay plan", "floor_plans": [ {"id": "FP_L01", "primary_location_id": "L01", "building_group": "house", "location_ids": ["L01"], "shot_count": 10, "rationale": "10 shot indoor"}, {"id": "FP_L03", "primary_location_id": "L03", "building_group": "office", "location_ids": ["L03", "L04"], "shot_count": 8, "rationale": "indoor B+C 같은 건물"}, ], "floor_plan_order": ["FP_L01", "FP_L03"], "chain_bg_groups": [ {"id": "CB_L01_DAY", "floor_plan_id": "FP_L01", "location_id": "L01", "kind": "interior", "parent_id": "", "time": "day", "shot_ids": ["S5_Shot1", "S5_Shot2", "S5_Shot3"], "rationale": "day"}, {"id": "CB_L01_NIGHT", "floor_plan_id": "FP_L01", "location_id": "L01", "kind": "interior", "parent_id": "CB_L01_DAY", "time": "night", "shot_ids": ["S6_Shot1", "S6_Shot2", "S6_Shot3"], "rationale": "night"}, {"id": "CB_L03_DAY", "floor_plan_id": "FP_L03", "location_id": "L03", "kind": "interior", "parent_id": "", "time": "day", "shot_ids": ["S9_Shot1", "S9_Shot2"], "rationale": "L03 day"}, {"id": "CB_L04_DAY", "floor_plan_id": "FP_L03", "location_id": "L04", "kind": "interior", "parent_id": "CB_L03_DAY", "time": "day", "shot_ids": ["S10_Shot1", "S10_Shot2"], "rationale": "L04 같은 건물"}, ], "chain_bg_order": ["CB_L01_DAY", "CB_L01_NIGHT", "CB_L03_DAY", "CB_L04_DAY"], "prev_shot_only": [ {"location_id": "L02", "kind": "outdoor", "rationale": "outdoor"}, {"location_id": "L05", "kind": "outdoor", "rationale": "outdoor"}, {"location_id": "L06", "kind": "indoor", "rationale": "low_freq 2 shot"}, ], } def _make_planner_runner(tmp_path: Path): from app.core.steps.background_planner_step import BackgroundPlannerStep runner = BackgroundPlannerStep.__new__(BackgroundPlannerStep) runner.project_id = PROJECT_ID runner.episode_id = EPISODE_ID runner.db = MagicMock() runner.project_config = {} runner.build_opik_metadata = MagicMock(return_value={}) runner.step_id = "background_planner" runner.run_id = "test-run" runner.opik_context = {} runner.manifest = {"label": "planner", "default_model": "gpt"} return runner def _make_floor_plan_runner(tmp_path: Path, *, canon_overrides: List[Any] = None): from app.core.steps.location_floor_plan_step import LocationFloorPlanStep runner = LocationFloorPlanStep.__new__(LocationFloorPlanStep) runner.project_id = PROJECT_ID runner.episode_id = EPISODE_ID runner.db = MagicMock() if canon_overrides is None: canon_overrides = [ MagicMock(id=f"canon-l0{i}", short_id=f"L0{i}", entity_type="location", name=f"loc-{i}") for i in range(1, 7) ] runner.db.query.return_value.filter.return_value.all.return_value = canon_overrides runner.db.query.return_value.filter_by.return_value.first.return_value = None runner.project_config = MagicMock() runner.build_opik_metadata = MagicMock(return_value={}) runner.step_id = "location_floor_plan" runner.run_id = "test-run" runner.opik_context = {} runner.manifest = {"label": "floor_plan", "default_model": "gpt"} return runner def _make_chain_planning_runner(tmp_path: Path): from app.core.steps.background_chain_planning_step import BackgroundChainPlanningStep runner = BackgroundChainPlanningStep.__new__(BackgroundChainPlanningStep) runner.project_id = PROJECT_ID runner.episode_id = EPISODE_ID runner.db = MagicMock() runner.project_config = MagicMock() runner.build_opik_metadata = MagicMock(return_value={}) runner.step_id = "background_chain_planning" runner.run_id = "test-run" runner.opik_context = {} runner.manifest = {"label": "chain_planning", "default_model": "gpt"} return runner def _make_chain_render_runner(tmp_path: Path, canons: List[Any] = None): from app.core.steps.background_chain_render_step import BackgroundChainRenderStep runner = BackgroundChainRenderStep.__new__(BackgroundChainRenderStep) runner.project_id = PROJECT_ID runner.episode_id = EPISODE_ID runner.db = MagicMock() if canons is None: canons = [ MagicMock(id=f"canon-l0{i}", short_id=f"L0{i}", entity_type="location", name=f"loc-{i}") for i in range(1, 7) ] runner.db.query.return_value.filter.return_value.all.return_value = canons runner.db.query.return_value.filter_by.return_value.first.return_value = None runner.project_config = MagicMock() runner.build_opik_metadata = MagicMock(return_value={}) runner.step_id = "background_chain_render" runner.run_id = "test-run" runner.opik_context = {} runner.manifest = {"label": "chain_render", "default_model": "gpt"} return runner # ────────────────────────────────────────────── # 시나리오 1: floor_plan_anchored mode 전체 4-step E2E # ────────────────────────────────────────────── def test_phase5_full_pipeline_floor_plan_anchored(tmp_path, monkeypatch): """mode=floor_plan_anchored: 4-step 모두 실행되며 planner-driven path 통과. 검증: - background_planner: 2 floor_plan + 4 chain_bg_group 산출 - location_floor_plan: 2 PNG 생성 + same building_group ref attach - background_chain_planning: 4 group prompt 생성 (group단위 LLM call) - background_chain_render: 4 PNG, ref priority 확인 """ _write_planner_dependencies(tmp_path) monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) monkeypatch.setattr("app.core.config.settings.openai_api_key", "test-key") monkeypatch.setattr("app.core.config.settings.background_mode", "floor_plan_anchored") # ── Step 1: background_planner ── plan = _ok_planner_output() monkeypatch.setattr( "app.modules.llm.llm_client.call_structured", lambda **kwargs: plan, ) planner_runner = _make_planner_runner(tmp_path) p_result = planner_runner._execute(mode="resume") assert p_result["completed_count"] == 1 assert p_result["failed_count"] == 0 assert len(p_result["data"]["floor_plans"]) == 2 assert len(p_result["data"]["chain_bg_groups"]) == 4 # planner 체크포인트를 manifest.json으로 기록 (downstream loader 입력) _write_checkpoint(tmp_path, "background_planner", p_result["data"]) # ── Step 2: location_floor_plan ── # _process_floor_plan을 mock — 실제 LLM/이미지 호출 우회. fp_call_log = [] def fake_process_floor_plan(self, *, fp_id, spec, prev_summaries, same_group_ref_paths, **kw): fp_call_log.append({ "fp_id": fp_id, "prev_count": len(prev_summaries), "same_group_ref_count": len(same_group_ref_paths), "primary": spec["primary_location_id"], "building_group": spec.get("building_group", ""), }) png_path = kw["image_dir"] / f"{fp_id}.png" png_path.write_bytes(b"x" * 2048) return { "status": "ok", "prompt_text": f"Top-down floor plan of {spec['primary_location_id']} " * 30, "png_path": str(png_path), "fp_id": fp_id, "primary_location_id": spec["primary_location_id"], "building_group": spec.get("building_group", ""), "location_ids": list(spec.get("location_ids") or []), "ref_used": "single_ref" if same_group_ref_paths else "text_only", } monkeypatch.setattr( "app.core.steps.location_floor_plan_step.LocationFloorPlanStep._process_floor_plan", fake_process_floor_plan, ) fp_runner = _make_floor_plan_runner(tmp_path) fp_result = fp_runner._execute(mode="resume") assert fp_result["applicable_count"] == 2 assert fp_result["completed_count"] == 2 assert fp_result["failed_count"] == 0 # 처리 순서 검증 (planner.floor_plan_order대로) assert [c["fp_id"] for c in fp_call_log] == ["FP_L01", "FP_L03"] # FP_L01은 prev/ref 없음, FP_L03도 다른 building_group이라 ref 0 assert fp_call_log[0] == {"fp_id": "FP_L01", "prev_count": 0, "same_group_ref_count": 0, "primary": "L01", "building_group": "house"} assert fp_call_log[1] == {"fp_id": "FP_L03", "prev_count": 1, "same_group_ref_count": 0, "primary": "L03", "building_group": "office"} # location_floor_plan 체크포인트 기록 (downstream loader 입력) _write_checkpoint(tmp_path, "location_floor_plan", fp_result["data"]) # ── Step 3: background_chain_planning ── # planner-driven path: 4 group → 4 _plan_one_location LLM call. 각 call마다 LocationGroup # 의 shots를 그대로 anchor_root nodes에 할당해 invariant 만족 plan 반환. def fake_plan_one_location(location_group, *args, **kwargs): shot_ids = [s.shot_id for s in location_group.shots] return { "location_id": location_group.location_id, "rationale_summary": f"plan for {location_group.location_id}", "nodes": [{ "id": f"{location_group.location_id}_node1", "kind": "anchor_root", "label": "wide", "description": "wide shot", "shot_ids": shot_ids, "parent_id": "", "depth": 0, "rationale": "anchor", "shared_visual_anchors_with_parent": [], }], "execution_order": [f"{location_group.location_id}_node1"], "unassigned_shots": [], } monkeypatch.setattr( "app.modules.pipeline.background_chain_planning._plan_one_location", fake_plan_one_location, ) cp_runner = _make_chain_planning_runner(tmp_path) cp_result = cp_runner._execute(mode="resume") assert cp_result["applicable_count"] == 4 assert cp_result["completed_count"] == 4 assert cp_result["failed_count"] == 0 groups_data = cp_result["data"]["groups"] for gid in ("CB_L01_DAY", "CB_L01_NIGHT", "CB_L03_DAY", "CB_L04_DAY"): assert gid in groups_data assert groups_data[gid]["status"] == "ok" # chain_planning 체크포인트 기록 _write_checkpoint(tmp_path, "background_chain_planning", cp_result["data"]) # ── Step 4: background_chain_render ── fake_b64 = _b64_png() client = MagicMock() client.images.edit.return_value.data = [MagicMock(b64_json=fake_b64)] client.images.generate.return_value.data = [MagicMock(b64_json=fake_b64)] monkeypatch.setattr("openai.OpenAI", lambda **kw: client) # generate_node_prompt mock — 실제 LLM 호출 우회 with patch( "app.modules.pipeline.background_chain_render.generate_node_prompt", return_value=("photo prompt", []), ): cr_runner = _make_chain_render_runner(tmp_path) cr_result = cr_runner._execute(mode="resume") assert cr_result["applicable_count"] == 4 assert cr_result["completed_count"] == 4 # 4 group 모두 PNG 생성 (group_id별 file은 location_id로 그룹핑된 v01..) rendered_groups = cr_result["data"]["groups"] for gid in ("CB_L01_DAY", "CB_L01_NIGHT", "CB_L03_DAY", "CB_L04_DAY"): assert rendered_groups[gid]["status"] == "ok" # ────────────────────────────────────────────── # 시나리오 2: mode=off → 4 step 모두 not_applicable # ────────────────────────────────────────────── def test_phase5_mode_off_skips_all(tmp_path, monkeypatch): """mode=off 시 background_planner, location_floor_plan은 immediate skip. background_chain_planning/render는 on_demand이지만 planner cp 없으면 legacy path로 동작 (chain_only 회귀 보장 — 별도 시나리오 3에서 검증). 여기선 background_planner와 location_floor_plan만 검증. """ _write_planner_dependencies(tmp_path) monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) monkeypatch.setattr("app.core.config.settings.openai_api_key", "test-key") monkeypatch.setattr("app.core.config.settings.background_mode", "off") # background_planner — mode=off 가드로 LLM 호출 0회 def _no_call(**kwargs): raise AssertionError("LLM should not be called when mode=off") monkeypatch.setattr("app.modules.llm.llm_client.call_structured", _no_call) planner_runner = _make_planner_runner(tmp_path) p_result = planner_runner._execute(mode="resume") assert p_result["applicable_count"] == 0 assert p_result["completed_count"] == 0 assert p_result["failed_count"] == 0 assert p_result["data"] == {} # location_floor_plan — mode=off 가드로 _empty_result 반환 fp_runner = _make_floor_plan_runner(tmp_path) fp_result = fp_runner._execute(mode="resume") assert fp_result["applicable_count"] == 0 assert fp_result["completed_count"] == 0 assert fp_result["failed_count"] == 0 assert fp_result["data"]["floor_plans"] == {} assert fp_result["data"]["locations"] == [] # ────────────────────────────────────────────── # 시나리오 3: mode=chain_only → planner+floor_plan skip, chain_bg legacy path # ────────────────────────────────────────────── def test_phase5_mode_chain_only_legacy(tmp_path, monkeypatch): """mode=chain_only: - background_planner: applicability=if_floor_plan_mode → mode 가드로 skip. - location_floor_plan: 동일하게 skip. - background_chain_planning: planner cp 없음 → legacy path. - background_chain_render: planner_chain_order=None → legacy path. legacy path가 mode=chain_only에서 정상 작동하는지(회귀 0건) 검증. """ _write_planner_dependencies(tmp_path) # shot_staging도 추가 (chain_planning legacy path 컨텍스트) _write_checkpoint(tmp_path, "shot_staging", {"scenes": []}) monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) monkeypatch.setattr("app.core.config.settings.openai_api_key", "test-key") monkeypatch.setattr("app.core.config.settings.background_mode", "chain_only") # planner — mode 가드 skip monkeypatch.setattr( "app.modules.llm.llm_client.call_structured", lambda **kw: (_ for _ in ()).throw(AssertionError("not called")), ) planner_runner = _make_planner_runner(tmp_path) p_result = planner_runner._execute(mode="resume") assert p_result["applicable_count"] == 0 # floor_plan — mode 가드 skip fp_runner = _make_floor_plan_runner(tmp_path) fp_result = fp_runner._execute(mode="resume") assert fp_result["applicable_count"] == 0 # chain_bg_planning — legacy path. _phase0_group_by_location → ThreadPool. # _plan_one_location helper를 mock — legacy 진입 보장. from app.modules.pipeline import background_chain_planning as bcp fake_group = bcp.LocationGroup( location_id="L01", location_name="loc-1", location_description="indoor", visual_traits=[], shots=[ bcp.ShotInfo(scene_index=5, shot_index=k, shot_id=f"S05_Shot{k}", description=f"shot {k}") for k in range(1, 6) ], ) monkeypatch.setattr( bcp, "_phase0_group_by_location", lambda *a, **k: {"L01": fake_group}, ) def fake_plan_one_location(location_group, *args, **kwargs): shot_ids = [s.shot_id for s in location_group.shots] return { "location_id": location_group.location_id, "rationale_summary": "legacy plan", "nodes": [{ "id": "n1", "kind": "anchor_root", "label": "wide", "description": "wide", "shot_ids": shot_ids, "parent_id": "", "depth": 0, "rationale": "anchor", "shared_visual_anchors_with_parent": [], }], "execution_order": ["n1"], "unassigned_shots": [], } monkeypatch.setattr(bcp, "_plan_one_location", fake_plan_one_location) cp_runner = _make_chain_planning_runner(tmp_path) cp_result = cp_runner._execute(mode="resume") # legacy: data.locations 키 — planner cp 없음이라 group key 미존재 assert "locations" in cp_result["data"] assert "L01" in cp_result["data"]["locations"] # planner-driven path가 활성화되지 않았는지 확인 (groups 키 없거나 빈 dict) assert cp_result["data"].get("groups") is None or cp_result["data"].get("groups") == {} # chain_planning 체크포인트 기록 (render 입력) _write_checkpoint(tmp_path, "background_chain_planning", cp_result["data"]) # chain_bg_render — legacy path. planner_chain_order=None → ThreadPool. fake_b64 = _b64_png() client = MagicMock() client.images.edit.return_value.data = [MagicMock(b64_json=fake_b64)] client.images.generate.return_value.data = [MagicMock(b64_json=fake_b64)] monkeypatch.setattr("openai.OpenAI", lambda **kw: client) with patch( "app.modules.pipeline.background_chain_render.generate_node_prompt", return_value=("photo p", []), ): cr_runner = _make_chain_render_runner(tmp_path) cr_result = cr_runner._execute(mode="resume") # legacy path: data["locations"] 사용. planner-driven groups 키 미존재 assert "locations" in cr_result["data"] assert "L01" in cr_result["data"]["locations"] # ────────────────────────────────────────────── # 시나리오 4: planner.floor_plans=[] → location_floor_plan no-op # ────────────────────────────────────────────── def test_phase5_planner_empty_floor_plans(tmp_path, monkeypatch): """planner가 floor_plans=[] 반환 (모두 outdoor) → location_floor_plan은 graceful no-op (applicable=0). background_chain_render는 planner_chain_order=[] graceful empty. """ _write_planner_dependencies(tmp_path) monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) monkeypatch.setattr("app.core.config.settings.openai_api_key", "test-key") monkeypatch.setattr("app.core.config.settings.background_mode", "floor_plan_anchored") empty_plan = { "rationale_summary": "all outdoor", "floor_plans": [], "floor_plan_order": [], "chain_bg_groups": [], "chain_bg_order": [], "prev_shot_only": [ {"location_id": f"L0{i}", "kind": "outdoor", "rationale": "outdoor"} for i in range(1, 7) ], } monkeypatch.setattr( "app.modules.llm.llm_client.call_structured", lambda **kw: empty_plan, ) planner_runner = _make_planner_runner(tmp_path) p_result = planner_runner._execute(mode="resume") assert p_result["completed_count"] == 1 assert p_result["data"]["floor_plans"] == [] _write_checkpoint(tmp_path, "background_planner", p_result["data"]) # location_floor_plan — empty floor_plans 시 no-op fp_runner = _make_floor_plan_runner(tmp_path) fp_result = fp_runner._execute(mode="resume") assert fp_result["applicable_count"] == 0 assert fp_result["completed_count"] == 0 assert fp_result["failed_count"] == 0 assert fp_result["data"]["floor_plans"] == {} _write_checkpoint(tmp_path, "location_floor_plan", fp_result["data"]) # chain_bg_planning — planner_groups에 빈 chain_bg_groups → None loader → legacy path. # legacy도 location 0개 (모두 outdoor 처리) — 빈 결과. from app.modules.pipeline import background_chain_planning as bcp monkeypatch.setattr(bcp, "_phase0_group_by_location", lambda *a, **k: {}) cp_runner = _make_chain_planning_runner(tmp_path) cp_result = cp_runner._execute(mode="resume") assert cp_result["applicable_count"] == 0 assert cp_result["data"].get("locations") == {} # ────────────────────────────────────────────── # 시나리오 5: floor_plan 1개 실패 → chain_bg_render 그 group skip + 나머지 진행 # ────────────────────────────────────────────── def test_phase5_planner_partial_failure(tmp_path, monkeypatch): """FP_L01은 success, FP_L03은 LLM 실패 → location_floor_plan 1 success / 1 failure. chain_bg_render는 floor_plan_paths에 FP_L01 PNG만 포함되며 4 group 처리는 모두 진행(L03 group은 floor_plan ref 없이 location_ref/legacy로 fallback). """ _write_planner_dependencies(tmp_path) monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) monkeypatch.setattr("app.core.config.settings.openai_api_key", "test-key") monkeypatch.setattr("app.core.config.settings.background_mode", "floor_plan_anchored") plan = _ok_planner_output() monkeypatch.setattr( "app.modules.llm.llm_client.call_structured", lambda **kw: plan, ) planner_runner = _make_planner_runner(tmp_path) p_result = planner_runner._execute(mode="resume") _write_checkpoint(tmp_path, "background_planner", p_result["data"]) # location_floor_plan — FP_L01 success, FP_L03 fail def fake_process_floor_plan(self, *, fp_id, spec, **kw): if fp_id == "FP_L03": return { "status": "failed", "prompt_text": "", "png_path": "", "fp_id": fp_id, "primary_location_id": spec["primary_location_id"], "building_group": spec.get("building_group", ""), "location_ids": list(spec.get("location_ids") or []), "ref_used": "text_only", "failure_reason": "RuntimeError: simulated LLM down", } png_path = kw["image_dir"] / f"{fp_id}.png" png_path.write_bytes(b"x" * 2048) return { "status": "ok", "prompt_text": f"Top-down floor plan {fp_id} " * 30, "png_path": str(png_path), "fp_id": fp_id, "primary_location_id": spec["primary_location_id"], "building_group": spec.get("building_group", ""), "location_ids": list(spec.get("location_ids") or []), "ref_used": "text_only", } monkeypatch.setattr( "app.core.steps.location_floor_plan_step.LocationFloorPlanStep._process_floor_plan", fake_process_floor_plan, ) fp_runner = _make_floor_plan_runner(tmp_path) fp_result = fp_runner._execute(mode="resume") assert fp_result["applicable_count"] == 2 assert fp_result["completed_count"] == 1 assert fp_result["failed_count"] == 1 fps = fp_result["data"]["floor_plans"] assert fps["FP_L01"]["status"] == "ok" assert fps["FP_L03"]["status"] == "failed" _write_checkpoint(tmp_path, "location_floor_plan", fp_result["data"]) # chain_bg_planning — 4 group 모두 OK 가정 (LLM mock) def fake_plan_one_location(location_group, *args, **kwargs): shot_ids = [s.shot_id for s in location_group.shots] return { "location_id": location_group.location_id, "rationale_summary": f"plan {location_group.location_id}", "nodes": [{ "id": f"{location_group.location_id}_n1", "kind": "anchor_root", "label": "x", "description": "x", "shot_ids": shot_ids, "parent_id": "", "depth": 0, "rationale": "x", "shared_visual_anchors_with_parent": [], }], "execution_order": [f"{location_group.location_id}_n1"], "unassigned_shots": [], } monkeypatch.setattr( "app.modules.pipeline.background_chain_planning._plan_one_location", fake_plan_one_location, ) cp_runner = _make_chain_planning_runner(tmp_path) cp_result = cp_runner._execute(mode="resume") assert cp_result["applicable_count"] == 4 _write_checkpoint(tmp_path, "background_chain_planning", cp_result["data"]) # chain_bg_render — FP_L03 PNG 없음. L03/L04 group은 building_group reverse lookup # 도 안 되므로 location_ref/legacy로 fallback. fake_b64 = _b64_png() client = MagicMock() client.images.edit.return_value.data = [MagicMock(b64_json=fake_b64)] client.images.generate.return_value.data = [MagicMock(b64_json=fake_b64)] monkeypatch.setattr("openai.OpenAI", lambda **kw: client) with patch( "app.modules.pipeline.background_chain_render.generate_node_prompt", return_value=("p", []), ): cr_runner = _make_chain_render_runner(tmp_path) cr_result = cr_runner._execute(mode="resume") # 4 group 모두 처리 시도 (L03/L04는 floor_plan 없이 fallback) assert cr_result["applicable_count"] == 4 rendered = cr_result["data"]["groups"] # L01 group은 floor_plan ref 사용 for gid in ("CB_L01_DAY", "CB_L01_NIGHT"): assert rendered[gid]["status"] == "ok" # L03/L04 group은 fallback해서 처리 (graceful) for gid in ("CB_L03_DAY", "CB_L04_DAY"): assert rendered[gid]["status"] in ("ok", "skipped_planning") # ────────────────────────────────────────────── # manifest invariant — depends_on 정합성 # ────────────────────────────────────────────── def test_location_floor_plan_depends_on_background_planner(): """T8 manifest 변경: location_floor_plan.depends_on에 background_planner 포함.""" from app.core.step_manifest import STEP_MANIFEST, get_depends_on deps = get_depends_on("location_floor_plan") assert "background_planner" in deps # 기존 6개 의존성도 그대로 유지 for d in ( "shot_validator", "shot_selection", "scene_save", "entity_merge", "visual_world_rules", "scene_director", ): assert d in deps # background_chain_planning/render는 best-effort 패턴 — manifest depends_on에 미포함. # (Phase 4 I1 동일.) cp_deps = get_depends_on("background_chain_planning") assert "background_planner" not in cp_deps cr_deps = get_depends_on("background_chain_render") assert "background_planner" not in cr_deps