"""W-L (2026-07-03) — 야외 기준 재편: 실사형 aerial establishing SOT 결정론 테스트. 야외 loc 의 bg plate 가 '야외 평면도'(추상 다이어그램)를 1순위 ref 로 받아 i2i 재해석 변동이 크고, 같은 loc 에 fp 가 여러 장이면 plate 끼리 서로 다른 위상으로 갈라지는 결함(fp 분열)의 근본 대응 검증. LLM / image / VLM API call 0 (render_one_background + render_location_aerial mock), DB write 0 (_register_image_assets mock 또는 fake db). 커버: - build_loc_aerial_context 구조 조인(순수): label/summary 운반, is_indoor bool 계약, group_mixed, first-wins. - build_aerial_establishing_prompt: location 데이터 주입 + 마커/텍스트 금지 계약(마커형 aerial 과 구분). - render_one_background: aerial_ref_path 가 ref **최우선(첫 번째)** + aerial_ref_attached + None(default) byte-identical. - 큐: 야외 loc 만 aerial 생성(loc당 1회, 정렬 순), 야외 bg = fp ref 제외 + aerial 1순위 + AERIAL_SITE_GUIDANCE, 실내 bg 불변, direct lane 야외 bg 도 aerial 첨부, mixed 그룹 = indoor fp 를 producer 의 I2I ref 로 전달. - fail-safe: aerial 생성 실패 loc 은 기존 fp 경로 그대로(치환 없음). - 캐시: 기존 aerial png + resume = producer 미호출 재사용 / force = 재생성. - flag OFF = byte-identical (producer 미호출·kwargs None·entry 필드 부재· config_hash 불변). - config_hash: shot_aware_plan + ON 일 때만 스탬프. - _register_image_assets: location_aerial UPSERT + aerial_ref bg 의 input_image_ids 1순위 = aerial UUID(fp 대체), 미해결이면 unresolved 구조키. """ from __future__ import annotations from pathlib import Path from types import SimpleNamespace from typing import Any, Dict, List, Optional from unittest.mock import MagicMock, patch from app.modules.pipeline.location_aerial import ( AERIAL_FP_COHERENCE_GUIDANCE, AERIAL_PROMPT_VERSION, AERIAL_SITE_GUIDANCE, build_aerial_establishing_prompt, build_loc_aerial_context, compute_aerial_context_hash, ) def _grp1_ctx_hash(*, fp_sot: str = "") -> str: """픽스처 grp1 의 구조 컨텍스트 해시(캐시 hit 조건 주입용).""" return compute_aerial_context_hash( group_id="grp1", members=[ {"loc_id": "L01", "label": "indoor room", "summary": "a small room", "is_indoor": True}, {"loc_id": "L02", "label": "outdoor yard", "summary": "an open yard", "is_indoor": False}, ], anchor_loc="L01", fp_sot=fp_sot, ) def _touch_png(p: Path) -> Path: p.parent.mkdir(parents=True, exist_ok=True) p.write_bytes(b"\x89PNG\r\n\x1a\n") return p def _bg_spec_fp(*, bg_id: str, fp_id: str, loc_id: str) -> Dict[str, Any]: return { "bg_id": bg_id, "loc_id": loc_id, "sub_location": "primary_unit", "state_label": "day", "applies_to_shots": [f"{bg_id}_Shot1"], "depends_on_fp": [fp_id], "depends_on_bg": [], } def _bg_spec_direct(*, bg_id: str, loc_id: str) -> Dict[str, Any]: return { "bg_id": bg_id, "loc_id": loc_id, "sub_location": "primary_unit", "state_label": "day", "applies_to_shots": [f"{bg_id}_Shot1"], "depends_on_fp": [], "depends_on_bg": [], "surface_role": "transition_zone", } _RENDER_GUIDANCE_FIELDS = ( "visible_space_directive", "camera_framing_directive", "subject_position_directive", "state_cue_directive", "negative_continuity_directive", ) def _plan_node(*, bg_id: str, node_index: int) -> Dict[str, Any]: return { "bg_id": bg_id, "node_index": node_index, "mode": "fp_seeded_anchor", "is_dwelling_identity_anchor": True, "rationale": f"rationale for {bg_id}", "render_action": "render_new_plate", "reuse_target_bg_id": "", "reference_decision": { "selected_refs": [], "rejected_refs": [], "same_physical_space_dedup_decision": "single_ref", "why_single_ref_or_two_refs": "single ref reason", "physical_space_id_per_ref": [], }, "camera_decision": { "camera_unit": 1, "camera_cell": [0, 0], "look_at_unit": 2, "look_at_cell": [2, 0], "lens_enum": "normal", "fov_deg": 50, "framing_notes": f"framing for {bg_id}", }, "render_guidance": { f: f"directive {f} for {bg_id}" for f in _RENDER_GUIDANCE_FIELDS }, } def _plan_for_fp(*, fp_id: str, nodes: List[Dict[str, Any]]) -> Dict[str, Any]: return { "fp_id": fp_id, "shot_aware_bg_render_plan_status": "ok", "graph": {"nodes": list(nodes)}, "production_clear": True, "real_api_call_counts": {"image": 0, "llm": 1, "vlm": 0}, "readback_status": "ok", "validators": {"all_validators_passed": True, "diagnostics": []}, "diagnostics": [], } def _build_cp_map( *, tmp_path: Path, groups: List[Dict[str, Any]], building_groups: List[Dict[str, Any]], ) -> Dict[str, Any]: plans: Dict[str, Any] = {} fp_renders: Dict[str, Any] = {} fp_prompts: Dict[str, Any] = {} bg_prompts: Dict[str, Any] = {} per_fp: Dict[str, Any] = {} for g in groups: plans[g["gid"]] = { "status": "ok", "plan": { "group_id": g["gid"], "rationale_summary": "", "floor_plans": [], "backgrounds": g["bg_specs"], "gen_order": [b["bg_id"] for b in g["bg_specs"]], }, } if g.get("fp_id"): fp_png = _touch_png(tmp_path / "fp_render" / f"{g['fp_id']}.png") fp_renders[g["fp_id"]] = { "status": "ok", "png_path": str(fp_png)} fp_prompts[g["fp_id"]] = { "status": "ok", "numbered_elements": [], "camera_recommendations": [], } per_fp[g["fp_id"]] = g["plan"] for b in g["bg_specs"]: bg_prompts[b["bg_id"]] = { "status": "ok", "t2i_prompt": f"prompt body for {b['bg_id']}", "shot_guides": [], "objects_owned_by_background": [], "spec": b, "group_id": g["gid"], } return { "background_master_plan": { "data": { "plans": plans, "bg_catalog_hash": "", "shot_binding_hash": "", } }, "background_prompt": {"data": {"backgrounds": bg_prompts}}, "floor_plan_render": {"data": {"floor_plans": fp_renders}}, "floor_plan_prompt": {"data": {"floor_plans": fp_prompts}}, "shot_aware_bg_render_plan": {"data": {"per_fp": per_fp}}, "background_classify": { "data": {"building_groups": building_groups}}, "floor_plan_overlay_payload": None, } def _new_step(cp_map: Dict[str, Any]): from app.core.steps.background_render_step import BackgroundRenderStep step = BackgroundRenderStep.__new__(BackgroundRenderStep) step.project_id = "p" step.episode_id = "e" step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) step._load_prev_checkpoint = MagicMock( side_effect=lambda sid: cp_map.get(sid) ) step.db = MagicMock() step._register_image_assets = MagicMock() return step def _apply_settings(monkeypatch, tmp_path, *, wl_on=True, wk_on=False): for key, val in ( ("projects_dir", str(tmp_path)), ("openai_api_key", "sk-test"), ("background_mode", "on"), ("background_render_reference_mode", "shot_aware_plan"), ("shot_aware_bg_render_plan_enabled", False), ("bg_render_substrate_enabled", False), # W-L 단독 동작 검증 — W-G/W-I/W-K 는 자기 스위트가 커버. ("outdoor_building_fp_ref_enabled", False), ("outdoor_building_anchor_ref_enabled", False), ("same_place_render_chain_enabled", wk_on), ("outdoor_aerial_reference_enabled", wl_on), # W-M: plate 단계화 체인은 자기 스위트가 커버 — .env ON 누수 차단 # 명시 OFF (이 파일은 W-M OFF 일 때의 W-L 계약을 검증). ("outdoor_plate_stage_chain_enabled", False), ): monkeypatch.setattr(f"app.core.config.settings.{key}", val) def _capturing_renderer(captured: List[Dict[str, Any]]): def _side(*, openai_client, image_model, prompt, out_path, fp_path, prior_bg_paths, bg_id, building_fp_path=None, building_anchor_path=None, same_place_anchor_path=None, aerial_ref_path=None, **kwargs): _touch_png(Path(out_path)) captured.append({ "bg_id": bg_id, "prompt": prompt, "fp_path": str(fp_path) if fp_path is not None else None, "prior_bg_paths": [str(p) for p in prior_bg_paths], "aerial_ref_path": ( str(aerial_ref_path) if aerial_ref_path is not None else None ), "same_place_anchor_path": ( str(same_place_anchor_path) if same_place_anchor_path is not None else None ), }) return { "status": "ok", "attempts": 1, "strategies": [], "ref_used": "refs_1", "final_block_reason": None, "building_fp_attached": ( building_fp_path is not None and building_fp_path.exists() ), "building_anchor_attached": ( building_anchor_path is not None and building_anchor_path.exists() ), "same_place_anchor_attached": ( same_place_anchor_path is not None and same_place_anchor_path.exists() ), "aerial_ref_attached": ( aerial_ref_path is not None and aerial_ref_path.exists() ), "png_path": str(out_path), } return _side def _capturing_aerial_producer( calls: List[Dict[str, Any]], *, ok: bool = True, ): def _side(*, openai_client, image_model, place_id, members, out_path, building_fp_path=None, **kwargs): calls.append({ "place_id": place_id, "members": list(members or []), "out_path": str(out_path), "building_fp_path": ( str(building_fp_path) if building_fp_path is not None else None ), }) if not ok: return { "status": "failed", "attempts": 2, "strategies": [], "building_fp_used": building_fp_path is not None, "prompt_used": "p", "prompt_version": AERIAL_PROMPT_VERSION, "final_block_reason": "boom", } _touch_png(Path(out_path)) return { "status": "ok", "attempts": 1, "strategies": [], "building_fp_used": building_fp_path is not None, "prompt_used": f"aerial prompt for {place_id}", "prompt_version": AERIAL_PROMPT_VERSION, "final_block_reason": None, "png_path": str(out_path), } return _side def _run(step, captured, aerial_calls, *, aerial_ok=True, mode="resume"): with patch( "app.core.steps.background_render_step._resolve_openai_client", return_value=MagicMock(), ), patch( "app.modules.pipeline.background_render.render_one_background", side_effect=_capturing_renderer(captured), ), patch( "app.modules.pipeline.location_aerial.render_location_aerial", side_effect=_capturing_aerial_producer(aerial_calls, ok=aerial_ok), ): return step._execute(mode=mode) def _fixture(tmp_path): """mixed 그룹 grp1 = 실내 L01(fp lane 1 + direct 1) + 야외 L02(fp lane 2bg, 같은 loc 공유 aerial 검증). 단독 야외 그룹 grp2 = L03(direct lane).""" groups = [ { "gid": "g_l01", "fp_id": "fp_l01", "bg_specs": [ _bg_spec_fp(bg_id="L01B01", fp_id="fp_l01", loc_id="L01")], "plan": _plan_for_fp(fp_id="fp_l01", nodes=[ _plan_node(bg_id="L01B01", node_index=0)]), }, { "gid": "g_l02", "fp_id": "fp_l02", "bg_specs": [ _bg_spec_fp(bg_id="L02B01", fp_id="fp_l02", loc_id="L02"), _bg_spec_fp(bg_id="L02B02", fp_id="fp_l02", loc_id="L02"), ], "plan": _plan_for_fp(fp_id="fp_l02", nodes=[ _plan_node(bg_id="L02B01", node_index=0), _plan_node(bg_id="L02B02", node_index=1), ]), }, { "gid": "g_direct", "bg_specs": [ _bg_spec_direct(bg_id="L01B02", loc_id="L01"), _bg_spec_direct(bg_id="L03B01", loc_id="L03"), ], }, ] building_groups = [ { "group_id": "grp1", "anchor_loc": "L01", "members": [ {"loc_id": "L01", "is_indoor": True, "shot_count": 5, "label": "indoor room", "summary": "a small room"}, {"loc_id": "L02", "is_indoor": False, "shot_count": 2, "label": "outdoor yard", "summary": "an open yard"}, ], }, { "group_id": "grp2", "anchor_loc": "L03", "members": [ {"loc_id": "L03", "is_indoor": False, "shot_count": 1, "label": "outdoor lane", "summary": "a narrow lane"}, ], }, ] return _build_cp_map( tmp_path=tmp_path, groups=groups, building_groups=building_groups) # ─── 순수 함수 ─── def test_build_loc_aerial_context_structure_join(): ctx = build_loc_aerial_context([ { "group_id": "g1", "members": [ {"loc_id": "L01", "is_indoor": True, "label": "a", "summary": "s1"}, {"loc_id": "L02", "is_indoor": False, "label": "b", "summary": "s2"}, {"loc_id": "L09", "is_indoor": "no"}, # bool 계약 위반 제외 ], }, {"group_id": "", "members": [ {"loc_id": "L07", "is_indoor": False}]}, # group_id 없음 제외 { "group_id": "g2", "members": [ {"loc_id": "L02", "is_indoor": True}, # first-wins 로 무시 {"loc_id": "L03", "is_indoor": False, "label": "c", "summary": "s3"}, ], }, { "group_id": "g3", "members": [ {"loc_id": "L05", "is_indoor": False, "label": "d", "summary": "s5"}, ], }, ]) assert set(ctx) == {"L01", "L02", "L03", "L05"} assert ctx["L01"]["group_id"] == "g1" assert ctx["L01"]["is_indoor"] is True assert ctx["L01"]["group_mixed"] is True assert ctx["L01"]["label"] == "a" and ctx["L01"]["summary"] == "s1" # 8차 정정: 그룹 통합 배치도용 전 멤버 데이터 + anchor_loc 운반. assert [m["loc_id"] for m in ctx["L01"]["group_members"]] == [ "L01", "L02"] assert ctx["L02"]["group_members"] == ctx["L01"]["group_members"] assert ctx["L02"]["is_indoor"] is False assert ctx["L02"]["group_mixed"] is True # first-wins 는 출력 entry 만 — 그룹 mixed 는 그룹 멤버 전체로 판정. assert ctx["L03"]["group_mixed"] is True # 단독 야외 그룹은 mixed 아님. assert ctx["L05"]["group_mixed"] is False def test_build_aerial_establishing_prompt_contract(): p = build_aerial_establishing_prompt(members=[ {"label": "outdoor yard", "summary": "an open yard", "is_indoor": False}, {"label": "inner room", "summary": "a small room", "is_indoor": True}, ]) # 그룹 전 멤버가 (indoor/outdoor part) 태그로 place 데이터에 주입된다. assert "(outdoor part) outdoor yard an open yard" in p assert "(indoor part) inner room a small room" in p # 도면형 배치도 계약(7차 정정: 실사 사진 아님 — 실내 fp 와 같은 도면 # 언어) + 마커/텍스트 금지 (마커형 aerial 과 구분). assert "TOP-DOWN SITE PLAN" in p assert "floor plan" in p.lower() assert "NOT a photograph" in p # 8차 정정: 통합(한 장) + 지붕뷰 + 탈것 전체 + 간판 판 규칙(전부 generic). assert "TOGETHER as one coherent whole" in p assert "roof plane" in p assert "ENTIRE craft" in p assert "lettering panels" in p assert "NO people" in p # W-M(9차): 텍스트/화살표/로고 금지는 유지하되 개구부 동그라미 기호는 # 허용(실내 fp 정합점) — 기존 무조건 "markers 금지"에서 계약 정밀화. assert "NO text, letters, numbers" in p assert "hollow-circle opening" in p.lower() or "hollow CIRCLE" in p # 데이터 없으면 중립 문구. p2 = build_aerial_establishing_prompt(members=[]) assert "an ordinary real-world outdoor site" in p2 # ─── render_one_background: aerial 1순위 계약 ─── def _capture_ref_paths(tmp_path, **render_kwargs): from app.modules.pipeline.background_render import render_one_background seen: Dict[str, Any] = {} def _fake_call(client, *, mode, prompt, ref_paths, call_kwargs, **kw): seen["mode"] = mode seen["ref_paths"] = [str(p) for p in (ref_paths or [])] return b"\x89PNG\r\n\x1a\n" * 200 with patch( "app.modules.pipeline.background_render.call_gpt_image_bytes", side_effect=_fake_call, ): info = render_one_background( openai_client=MagicMock(), image_model="gpt-image-2", prompt="p", out_path=tmp_path / "out.png", **render_kwargs, ) return info, seen def test_render_one_background_aerial_first(tmp_path): aerial = _touch_png(tmp_path / "aerial_L02.png") prior = _touch_png(tmp_path / "L02B01.png") bfp = _touch_png(tmp_path / "fp_l01.png") info, seen = _capture_ref_paths( tmp_path, fp_path=None, prior_bg_paths=[prior], aerial_ref_path=aerial, building_fp_path=bfp, ) assert info["status"] == "ok" assert info["aerial_ref_attached"] is True assert seen["ref_paths"] == [str(aerial), str(prior), str(bfp)] def test_render_one_background_aerial_default_none_byte_identical(tmp_path): fp = _touch_png(tmp_path / "fp_l02.png") info, seen = _capture_ref_paths( tmp_path, fp_path=fp, prior_bg_paths=[], ) assert info["ref_used"] == "fp_only" assert info["aerial_ref_attached"] is False assert seen["ref_paths"] == [str(fp)] # ─── 큐 배선 ─── def test_outdoor_bg_fp_replaced_by_aerial(tmp_path, monkeypatch): _apply_settings(monkeypatch, tmp_path) cp_map = _fixture(tmp_path) step = _new_step(cp_map) captured: List[Dict[str, Any]] = [] aerial_calls: List[Dict[str, Any]] = [] result = _run(step, captured, aerial_calls) # aerial 은 야외 loc 보유 그룹당 1회, 그룹 정렬 순 — grp1(L02)/grp2(L03). assert [c["place_id"] for c in aerial_calls] == ["grp1", "grp2"] # 통합 배치도 데이터 = 그룹 전 멤버(실내 L01 포함). assert [m["loc_id"] for m in aerial_calls[0]["members"]] == [ "L01", "L02"] by_bid = {c["bg_id"]: c for c in captured} # 야외 fp lane: fp ref 제외 + aerial 1순위 + guidance. for bid in ("L02B01", "L02B02"): c = by_bid[bid] assert c["fp_path"] is None assert (c["aerial_ref_path"] or "").endswith("aerial_grp1.png") assert AERIAL_SITE_GUIDANCE in c["prompt"] # 야외 direct lane: aerial 첨부 (fp 는 원래 없음). c3 = by_bid["L03B01"] assert (c3["aerial_ref_path"] or "").endswith("aerial_grp2.png") assert AERIAL_SITE_GUIDANCE in c3["prompt"] # 실내 loc 불변: fp 유지 + aerial 없음 + guidance 없음. c1 = by_bid["L01B01"] assert (c1["fp_path"] or "").endswith("fp_l01.png") assert c1["aerial_ref_path"] is None assert AERIAL_SITE_GUIDANCE not in c1["prompt"] assert by_bid["L01B02"]["aerial_ref_path"] is None # entry 구조 필드 + 라벨 (aerial 첫 번째, fp 라벨 부재) + 진단. g = result["data"]["groups"] e = g["L02B01"] assert e["aerial_ref"] == {"loc_id": "L02", "group_id": "grp1"} assert e["fp_ref_replaced_by_aerial"] is True assert e["floor_plan_used"] is False labels = e["attached_reference_lineage"]["attached_ref_labels"] assert labels[0] == "aerial:grp1" assert not [x for x in labels if x.startswith("fp:")] assert "aerial_ref" not in g["L01B01"] assert g["L03B01"]["aerial_ref"] == { "loc_id": "L03", "group_id": "grp2"} # direct lane 은 fp 치환 개념이 없다 — 마커 부재. assert "fp_ref_replaced_by_aerial" not in g["L03B01"] la = result["data"]["location_aerials"] assert la["grp1"]["status"] == "ok" and la["grp1"]["cached"] is False assert la["grp1"]["locs"] == ["L02"] assert la["grp1"]["anchor_loc"] == "L01" assert la["grp2"]["status"] == "ok" # _register_image_assets 에 진단 전달. _, reg_kwargs = step._register_image_assets.call_args assert set(reg_kwargs["location_aerials"]) == {"grp1", "grp2"} def test_mixed_group_aerial_uses_indoor_fp_ref(tmp_path, monkeypatch): _apply_settings(monkeypatch, tmp_path) cp_map = _fixture(tmp_path) indoor_fp_png = _touch_png(tmp_path / "fp_render" / "fp_l01.png") step = _new_step(cp_map) step._load_building_fp_by_loc = MagicMock(return_value={ "L02": { "fp_id": "fp_l01", "indoor_loc_sid": "L01", "group_id": "grp1", "png_path": str(indoor_fp_png), "fp_asset_id": "fp-uuid-1", }, }) captured: List[Dict[str, Any]] = [] aerial_calls: List[Dict[str, Any]] = [] result = _run(step, captured, aerial_calls) by_gid = {c["place_id"]: c for c in aerial_calls} # mixed 그룹(grp1) aerial 만 indoor fp 를 I2I ref 로 받는다. assert (by_gid["grp1"]["building_fp_path"] or "").endswith("fp_l01.png") assert by_gid["grp2"]["building_fp_path"] is None la = result["data"]["location_aerials"] assert la["grp1"]["building_fp_used"] is True assert la["grp1"]["fp_asset_id"] == "fp-uuid-1" def test_aerial_failure_keeps_fp_path_fail_safe(tmp_path, monkeypatch): _apply_settings(monkeypatch, tmp_path) cp_map = _fixture(tmp_path) step = _new_step(cp_map) captured: List[Dict[str, Any]] = [] aerial_calls: List[Dict[str, Any]] = [] result = _run(step, captured, aerial_calls, aerial_ok=False) by_bid = {c["bg_id"]: c for c in captured} # 실패 loc 은 기존 fp 경로 그대로 (치환/첨부/guidance 전부 없음). for bid in ("L02B01", "L02B02"): c = by_bid[bid] assert (c["fp_path"] or "").endswith("fp_l02.png") assert c["aerial_ref_path"] is None assert AERIAL_SITE_GUIDANCE not in c["prompt"] g = result["data"]["groups"] assert "aerial_ref" not in g["L02B01"] assert "fp_ref_replaced_by_aerial" not in g["L02B01"] la = result["data"]["location_aerials"] assert la["grp1"]["status"] == "failed" assert la["grp1"]["error"] == "boom" def test_aerial_cache_reuse_on_resume_and_regen_on_force( tmp_path, monkeypatch, ): _apply_settings(monkeypatch, tmp_path) cp_map = _fixture(tmp_path) image_dir = ( tmp_path / "p" / "episodes" / "e" / "images" / "background_chain" ) _touch_png(image_dir / "aerial_grp1.png") # Codex NARROW: cached 재사용은 이전 run 의 prompt_version 이 현재와 # 일치할 때만 — 자기 cp 에 그룹 키로 버전 기록 주입. cp_map["background_render"] = {"data": {"location_aerials": { "grp1": {"status": "ok", "prompt_version": AERIAL_PROMPT_VERSION, "context_hash": _grp1_ctx_hash()}, }}} # resume + 버전·컨텍스트 일치: grp1 재사용(producer 미호출), grp2 만 생성. step = _new_step(cp_map) captured: List[Dict[str, Any]] = [] aerial_calls: List[Dict[str, Any]] = [] result = _run(step, captured, aerial_calls, mode="resume") assert [c["place_id"] for c in aerial_calls] == ["grp2"] by_bid = {c["bg_id"]: c for c in captured} assert (by_bid["L02B01"]["aerial_ref_path"] or "").endswith( "aerial_grp1.png") assert result["data"]["location_aerials"]["grp1"]["cached"] is True assert result["data"]["location_aerials"]["grp1"][ "prompt_version"] == AERIAL_PROMPT_VERSION # force: 둘 다 재생성. step2 = _new_step(cp_map) captured2: List[Dict[str, Any]] = [] aerial_calls2: List[Dict[str, Any]] = [] _run(step2, captured2, aerial_calls2, mode="force") assert [c["place_id"] for c in aerial_calls2] == ["grp1", "grp2"] def test_aerial_cache_regen_on_prompt_version_mismatch(tmp_path, monkeypatch): """Codex NARROW — 이전 run 의 aerial 이 다른 prompt_version(예: 구 실사형) 산출이면 파일이 있어도 resume 에서 재생성한다.""" _apply_settings(monkeypatch, tmp_path) cp_map = _fixture(tmp_path) image_dir = ( tmp_path / "p" / "episodes" / "e" / "images" / "background_chain" ) _touch_png(image_dir / "aerial_grp1.png") # 구버전 기록 → mismatch. (버전 기록 자체가 없던 이전 cp 도 동일 경로.) cp_map["background_render"] = {"data": {"location_aerials": { "grp1": {"status": "ok", "prompt_version": "1.000000000000"}, }}} step = _new_step(cp_map) captured: List[Dict[str, Any]] = [] aerial_calls: List[Dict[str, Any]] = [] result = _run(step, captured, aerial_calls, mode="resume") assert [c["place_id"] for c in aerial_calls] == ["grp1", "grp2"] assert result["data"]["location_aerials"]["grp1"]["cached"] is False # 버전 기록이 아예 없는 이전 cp(구 실사형 시절) → 역시 재생성. cp_map["background_render"] = {"data": {"location_aerials": { "grp1": {"status": "ok"}, }}} step2 = _new_step(cp_map) captured2: List[Dict[str, Any]] = [] aerial_calls2: List[Dict[str, Any]] = [] _run(step2, captured2, aerial_calls2, mode="resume") assert [c["place_id"] for c in aerial_calls2] == ["grp1", "grp2"] # Codex 8차 NARROW: 버전은 같아도 그룹 구조 컨텍스트(멤버 summary 등)가 # 바뀌면 재생성 — 낡은 그룹 배치도 부활 차단. cp_map["background_render"] = {"data": {"location_aerials": { "grp1": {"status": "ok", "prompt_version": AERIAL_PROMPT_VERSION, "context_hash": "0" * 16}, }}} step3 = _new_step(cp_map) captured3: List[Dict[str, Any]] = [] aerial_calls3: List[Dict[str, Any]] = [] result3 = _run(step3, captured3, aerial_calls3, mode="resume") assert [c["place_id"] for c in aerial_calls3] == ["grp1", "grp2"] assert result3["data"]["location_aerials"]["grp1"][ "context_hash"] == _grp1_ctx_hash() def test_aerial_cached_diag_reflects_structural_fp_sot(tmp_path, monkeypatch): """Codex NARROW — cached 재사용 run 의 진단 fp 필드는 이 run 의 렌더 여부가 아니라 현재 구조 SOT(mixed 그룹 fp 링크) 기준으로 채운다.""" _apply_settings(monkeypatch, tmp_path) cp_map = _fixture(tmp_path) image_dir = ( tmp_path / "p" / "episodes" / "e" / "images" / "background_chain" ) _touch_png(image_dir / "aerial_grp1.png") cp_map["background_render"] = {"data": {"location_aerials": { "grp1": {"status": "ok", "prompt_version": AERIAL_PROMPT_VERSION, "context_hash": _grp1_ctx_hash(fp_sot="fp_l01")}, }}} indoor_fp_png = _touch_png(tmp_path / "fp_render" / "fp_l01.png") step = _new_step(cp_map) step._load_building_fp_by_loc = MagicMock(return_value={ "L02": { "fp_id": "fp_l01", "indoor_loc_sid": "L01", "group_id": "grp1", "png_path": str(indoor_fp_png), "fp_asset_id": "fp-uuid-1", }, }) captured: List[Dict[str, Any]] = [] aerial_calls: List[Dict[str, Any]] = [] result = _run(step, captured, aerial_calls, mode="resume") # grp1 은 cached 재사용(producer 미호출) — 그래도 구조 SOT 필드 유지. assert [c["place_id"] for c in aerial_calls] == ["grp2"] la = result["data"]["location_aerials"] assert la["grp1"]["cached"] is True assert la["grp1"]["building_fp_used"] is True assert la["grp1"]["fp_asset_id"] == "fp-uuid-1" def test_flag_off_byte_identical(tmp_path, monkeypatch): _apply_settings(monkeypatch, tmp_path, wl_on=False) cp_map = _fixture(tmp_path) step = _new_step(cp_map) captured: List[Dict[str, Any]] = [] aerial_calls: List[Dict[str, Any]] = [] result = _run(step, captured, aerial_calls) assert aerial_calls == [] by_bid = {c["bg_id"]: c for c in captured} for bid in ("L02B01", "L02B02"): assert (by_bid[bid]["fp_path"] or "").endswith("fp_l02.png") for c in captured: assert c["aerial_ref_path"] is None assert AERIAL_SITE_GUIDANCE not in c["prompt"] for e in result["data"]["groups"].values(): assert "aerial_ref" not in e assert "fp_ref_replaced_by_aerial" not in e labels = (e.get("attached_reference_lineage") or {}).get( "attached_ref_labels") or [] assert not [x for x in labels if x.startswith("aerial:")] assert "location_aerials" not in result["data"] def test_wl_with_wk_anchor_order_preserved(tmp_path, monkeypatch): """W-L + W-K 동시 ON — aerial 1순위와 same_loc anchor 공존(순서 계약은 render_one_background 가 보장, 큐는 양 kwargs 전달만 검증).""" _apply_settings(monkeypatch, tmp_path, wk_on=True) cp_map = _fixture(tmp_path) step = _new_step(cp_map) captured: List[Dict[str, Any]] = [] aerial_calls: List[Dict[str, Any]] = [] _run(step, captured, aerial_calls) by_bid = {c["bg_id"]: c for c in captured} # W-K 재배열로 mixed outdoor(L02) fp lane 이 먼저 — L02B02 는 loc anchor # (L02B01)와 aerial 둘 다 받는다. c = by_bid["L02B02"] assert (c["aerial_ref_path"] or "").endswith("aerial_grp1.png") assert (c["same_place_anchor_path"] or "").endswith("L02B01.png") assert AERIAL_SITE_GUIDANCE in c["prompt"] # guidance 순서: aerial(1순위) 이 same-place 보다 앞. from app.modules.pipeline.background_render import ( SAME_PLACE_PLATE_GUIDANCE, ) assert SAME_PLACE_PLATE_GUIDANCE in c["prompt"] assert c["prompt"].index(AERIAL_SITE_GUIDANCE) < c["prompt"].index( SAME_PLACE_PLATE_GUIDANCE) # ─── config_hash ─── def test_config_hash_stamps_flag_only_when_on(tmp_path, monkeypatch): _apply_settings(monkeypatch, tmp_path, wl_on=True) cp_map = _fixture(tmp_path) h_on = _new_step(cp_map)._config_hash() _apply_settings(monkeypatch, tmp_path, wl_on=False) h_off = _new_step(cp_map)._config_hash() assert h_on != h_off # legacy 모드에서는 ON 이어도 스탬프하지 않는다(shot_aware 전용). _apply_settings(monkeypatch, tmp_path, wl_on=True) monkeypatch.setattr( "app.core.config.settings.background_render_reference_mode", "legacy") h_legacy_on = _new_step(cp_map)._config_hash() _apply_settings(monkeypatch, tmp_path, wl_on=False) monkeypatch.setattr( "app.core.config.settings.background_render_reference_mode", "legacy") h_legacy_off = _new_step(cp_map)._config_hash() assert h_legacy_on == h_legacy_off # ─── location_aerial producer 단위 ─── def test_render_location_aerial_t2i_and_fp_edit(tmp_path): from app.modules.pipeline.location_aerial import render_location_aerial seen: List[Dict[str, Any]] = [] def _fake_call(client, *, mode, prompt, ref_paths, call_kwargs, **kw): seen.append({ "mode": mode, "prompt": prompt, "ref_paths": [str(p) for p in (ref_paths or [])], }) return b"\x89PNG\r\n\x1a\n" * 200 _members = [ {"label": "outdoor yard", "summary": "an open yard", "is_indoor": False}, {"label": "inner room", "summary": "a small room", "is_indoor": True}, ] with patch( "app.modules.pipeline.location_aerial.call_gpt_image_bytes", side_effect=_fake_call, ): info = render_location_aerial( openai_client=MagicMock(), image_model="gpt-image-2", place_id="grp1", members=_members, out_path=tmp_path / "aerial_grp1.png", ) assert info["status"] == "ok" assert info["prompt_version"] == AERIAL_PROMPT_VERSION assert seen[-1]["mode"] == "generate" assert "outdoor yard" in seen[-1]["prompt"] assert "inner room" in seen[-1]["prompt"] assert AERIAL_FP_COHERENCE_GUIDANCE not in seen[-1]["prompt"] assert (tmp_path / "aerial_grp1.png").exists() fp = _touch_png(tmp_path / "fp_l01.png") info2 = render_location_aerial( openai_client=MagicMock(), image_model="gpt-image-2", place_id="grp1", members=_members, out_path=tmp_path / "aerial_grp1b.png", building_fp_path=fp, ) assert info2["status"] == "ok" assert info2["building_fp_used"] is True assert seen[-1]["mode"] == "edit" assert seen[-1]["ref_paths"] == [str(fp)] assert AERIAL_FP_COHERENCE_GUIDANCE in seen[-1]["prompt"] def test_render_location_aerial_moderation_retry_and_failure(tmp_path): from app.modules.pipeline.location_aerial import render_location_aerial calls = {"n": 0} def _fake_call(client, **kw): calls["n"] += 1 if calls["n"] == 1: raise RuntimeError("blocked by moderation") return b"\x89PNG\r\n\x1a\n" * 200 sanitizer = MagicMock() sanitizer.sanitize.return_value = { "sanitized_prompt": "safe prompt", "strategy": "soften"} with patch( "app.modules.pipeline.location_aerial.call_gpt_image_bytes", side_effect=_fake_call, ): info = render_location_aerial( openai_client=MagicMock(), image_model="gpt-image-2", place_id="grp1", members=[{"label": "l", "summary": "s", "is_indoor": False}], out_path=tmp_path / "a.png", sanitizer=sanitizer, ) assert info["status"] == "ok" assert info["attempts"] == 2 assert sanitizer.sanitize.called # 비-moderation 오류는 재시도 없이 실패 dict (raise 금지 계약). with patch( "app.modules.pipeline.location_aerial.call_gpt_image_bytes", side_effect=RuntimeError("connection reset"), ): info2 = render_location_aerial( openai_client=MagicMock(), image_model="gpt-image-2", place_id="grp1", members=[{"label": "l", "summary": "s", "is_indoor": False}], out_path=tmp_path / "b.png", sanitizer=sanitizer, max_attempts=2, ) assert info2["status"] == "failed" assert "connection reset" in (info2["final_block_reason"] or "") # ─── _register_image_assets: aerial UPSERT + 1순위 lineage 교체 ─── class _FakeQuery: def __init__(self, rows): self._rows = rows def filter(self, *a, **k): return self def filter_by(self, **k): return self def order_by(self, *a, **k): return self def all(self): return self._rows def first(self): return None def _fake_db(canon_rows): db = MagicMock() def _query(*args): name = getattr(args[0], "__name__", "") if name == "EntityCanon": return _FakeQuery(canon_rows) return _FakeQuery([]) db.query = MagicMock(side_effect=_query) return db def _run_register( tmp_path, monkeypatch, groups, order, location_aerials: Optional[Dict[str, Any]] = None, ): from app.core.steps.background_render_step import BackgroundRenderStep monkeypatch.setattr( "app.core.config.settings.projects_dir", str(tmp_path / "proj")) monkeypatch.setattr( "app.core.config.settings." "background_render_record_prior_bg_lineage_enabled", True, ) step = BackgroundRenderStep.__new__(BackgroundRenderStep) step.project_id = "p" step.episode_id = "e" step.project_config = {} canon_rows = [ SimpleNamespace(short_id="L01", id="canon-l01"), SimpleNamespace(short_id="L02", id="canon-l02"), ] step.db = _fake_db(canon_rows) added = [] step.db.add = MagicMock(side_effect=added.append) calls = [] with patch( "app.core.steps.background_render_step.annotate_generated_asset", side_effect=lambda row, **kw: calls.append((row, kw)), ): step._register_image_assets( groups, order, location_aerials=location_aerials) return added, calls def _register_groups_fixture(tmp_path): p1 = _touch_png(tmp_path / "bgc" / "L02B01.png") p2 = _touch_png(tmp_path / "bgc" / "L02B02.png") groups = { "L02B01": { "status": "ok", "png_path": str(p1), "location_id": "L02", "shot_guides": [], "shot_ids": [], "t2i_prompt": "t", "aerial_ref": {"loc_id": "L02", "group_id": "grp1"}, "fp_ref_replaced_by_aerial": True, "attached_reference_lineage": { "prior_bg_ids": [], "ref_used": "refs_1", "attached_ref_labels": ["aerial:grp1"]}, }, "L02B02": { "status": "ok", "png_path": str(p2), "location_id": "L02", "shot_guides": [], "shot_ids": [], "t2i_prompt": "t", "aerial_ref": {"loc_id": "L02", "group_id": "grp1"}, "fp_ref_replaced_by_aerial": True, "attached_reference_lineage": { "prior_bg_ids": ["L02B01"], "ref_used": "refs_2", "attached_ref_labels": ["aerial:grp1", "bg:L02B01"]}, }, } return groups, ["L02B01", "L02B02"] def test_register_aerial_asset_and_first_slot_lineage(tmp_path, monkeypatch): groups, order = _register_groups_fixture(tmp_path) aerial_png = _touch_png(tmp_path / "bgc" / "aerial_grp1.png") location_aerials = { "grp1": { "status": "ok", "png_path": str(aerial_png), "cached": False, "building_fp_used": True, "fp_asset_id": "fp-uuid-1", "prompt_used": "aerial prompt", "attempts": 1, "prompt_version": AERIAL_PROMPT_VERSION, "locs": ["L02"], "anchor_loc": "L01", }, } added, calls = _run_register( tmp_path, monkeypatch, groups, order, location_aerials=location_aerials) row_by_vt = {r.variant_type: r for r in added} assert set(row_by_vt) == {"aerial_grp1", "L02B01", "L02B02"} a_row = row_by_vt["aerial_grp1"] assert a_row.asset_type == "location_aerial" # entity = 그룹 anchor_loc canon (classify 구조 필드). assert a_row.entity_id == "canon-l01" assert a_row.prompt_used == "aerial prompt" # aerial row lineage = 생성에 쓴 indoor fp UUID. a_kw = [kw for row, kw in calls if row is a_row][-1] assert a_kw["pipeline_role"] == "location_aerial" assert a_kw["input_image_ids"] == ["fp-uuid-1"] # bg 1순위 lineage = aerial UUID (loc floor_plan 아님). b01_kw = [kw for row, kw in calls if row is row_by_vt["L02B01"]][-1] assert b01_kw["input_image_ids"] == [a_row.id] # phase2(prior 보유): [aerial, prior] 순서. b02_kw = [kw for row, kw in calls if row is row_by_vt["L02B02"]][-1] assert b02_kw["input_image_ids"] == [a_row.id, row_by_vt["L02B01"].id] def test_register_unresolved_aerial_marks_structural_key( tmp_path, monkeypatch, ): """aerial 산출 진단 없이 aerial_ref 만 있으면 unresolved 구조키.""" groups, order = _register_groups_fixture(tmp_path) added, calls = _run_register( tmp_path, monkeypatch, groups, order, location_aerials=None) row_by_vt = {r.variant_type: r for r in added} assert set(row_by_vt) == {"L02B01", "L02B02"} b01_kw = [kw for row, kw in calls if row is row_by_vt["L02B01"]][-1] meta = b01_kw.get("pipeline_metadata") or {} assert "location_aerial:L02" in (meta.get("unresolved_inputs") or []) # phase2 row 도 unresolved 이월 + prior 는 정상 resolve. b02_kw = [kw for row, kw in calls if row is row_by_vt["L02B02"]][-1] meta2 = b02_kw.get("pipeline_metadata") or {} assert "location_aerial:L02" in (meta2.get("unresolved_inputs") or []) assert row_by_vt["L02B01"].id in (b02_kw.get("input_image_ids") or [])