"""W-I (2026-07-03) — building group anchor 렌더 체이닝 결정론 테스트. 같은 building 그룹(실내·실외 공존)의 outdoor bg 들이 도면(fp)만 공유하고 렌더를 상호 참조하지 않아 plate 마다 건물 외관이 재발명되는 결함의 근본 대응 검증. LLM / image / VLM API call 0 (render_one_background mock), DB write 0 (_register_image_assets mock 또는 fake db). 커버: - first-ok-render-wins: 그룹 첫 ok 렌더가 anchor 로 등록되고, 이후 같은 그룹 **다른 location** 의 bg 렌더에 building_anchor_path 로 첨부된다. - 프롬프트에 BUILDING_ANCHOR_PLATE_GUIDANCE 가 덧붙는다 (첨부시에만). - entry.building_anchor_ref 구조 필드 + lineage 라벨. - flag OFF = byte-identical (anchor kwargs None / entry 필드 부재 / config_hash 불변). - W-G(fp ref) flag OFF 여도 W-I 단독으로 링크 맵 재사용해 동작 (fp 첨부는 안 함). - dedup: substrate prior 로 anchor png 가 이미 첨부되는 렌더에는 중복 첨부 안 함. - 링크 밖 location 은 등록/첨부 대상 아님. - _register_image_assets: building_anchor_ref → anchor UUID 가 input_image_ids 에 병합(캔버스 bg→bg 엣지), 미해결이면 unresolved_inputs 구조키. """ from __future__ import annotations from pathlib import Path from types import SimpleNamespace from typing import Any, Dict, List from unittest.mock import MagicMock, patch from app.modules.pipeline.background_render import ( BUILDING_ANCHOR_PLATE_GUIDANCE, BUILDING_FP_PLATE_GUIDANCE, ) 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(*, 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": [], } _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, mode: str = "fp_seeded_anchor", selected_refs=None, is_anchor: bool = True, ) -> Dict[str, Any]: if selected_refs is None: selected_refs = [] return { "bg_id": bg_id, "node_index": node_index, "mode": mode, "is_dwelling_identity_anchor": is_anchor, "rationale": f"rationale for {bg_id}", "render_action": "render_new_plate", "reuse_target_bg_id": "", "reference_decision": { "selected_refs": list(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": [ r.get("physical_space_id", "") for r in selected_refs ], }, "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_two_groups( *, tmp_path: Path, groups: List[Dict[str, Any]], ) -> Dict[str, Any]: """groups = [{"gid", "fp_id", "bg_specs", "plan"}] — 멀티 fp 그룹 cp 조립.""" 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: fp_png = _touch_png(tmp_path / "fp_render" / f"{g['fp_id']}.png") 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"]], }, } fp_renders[g["fp_id"]] = {"status": "ok", "png_path": str(fp_png)} fp_prompts[g["fp_id"]] = { "status": "ok", "numbered_elements": [], "camera_recommendations": [], } 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"], } per_fp[g["fp_id"]] = g["plan"] 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}}, "floor_plan_overlay_payload": None, } def _new_step(cp_map: Dict[str, Any], *, building_link=None): 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() if building_link is not None: step._load_building_fp_by_loc = MagicMock(return_value=building_link) return step def _apply_settings( monkeypatch, tmp_path, *, anchor_on=True, wg_fp_on=True, ): 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), ("outdoor_building_anchor_ref_enabled", anchor_on), ("outdoor_building_fp_ref_enabled", wg_fp_on), # W-K 는 별도 스위트 — 이 파일은 W-K OFF 일 때의 W-I 계약을 검증 # (.env 의 실환경 ON 값이 새어들지 않게 명시 고정). ("same_place_render_chain_enabled", False), # W-L: aerial 은 자기 스위트가 커버 — .env ON 누수 차단 명시 OFF. ("outdoor_aerial_reference_enabled", False), # W-M: plate 단계화 체인도 자기 스위트가 커버 — 명시 OFF. ("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, **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], "building_fp_path": ( str(building_fp_path) if building_fp_path is not None else None ), "building_anchor_path": ( str(building_anchor_path) if building_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() ), "png_path": str(out_path), } return _side def _two_loc_fixture(tmp_path, *, link=True): """L03(fp_l03)·L10(fp_l10) 두 outdoor loc, 같은 building 그룹 bgrp1.""" groups = [ { "gid": "g_l03", "fp_id": "fp_l03", "bg_specs": [_bg_spec(bg_id="L03B01", fp_id="fp_l03", loc_id="L03")], "plan": _plan_for_fp(fp_id="fp_l03", nodes=[ _plan_node(bg_id="L03B01", node_index=0)]), }, { "gid": "g_l10", "fp_id": "fp_l10", "bg_specs": [_bg_spec(bg_id="L10B01", fp_id="fp_l10", loc_id="L10")], "plan": _plan_for_fp(fp_id="fp_l10", nodes=[ _plan_node(bg_id="L10B01", node_index=0)]), }, ] cp_map = _build_cp_map_two_groups(tmp_path=tmp_path, groups=groups) building_link = None if link: indoor_fp_png = _touch_png(tmp_path / "fp_render" / "fp_in.png") building_link = { loc: { "fp_id": "fp_in", "indoor_loc_sid": "L04", "group_id": "bgrp1", "png_path": str(indoor_fp_png), "fp_asset_id": "uuid-fp-in", } for loc in ("L03", "L10") } return cp_map, building_link def _run(step, captured): 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), ): return step._execute() # ─── first-ok-render-wins + cross-location 첨부 ─── def test_anchor_attached_across_locations_same_group(tmp_path, monkeypatch): _apply_settings(monkeypatch, tmp_path) cp_map, link = _two_loc_fixture(tmp_path) step = _new_step(cp_map, building_link=link) captured: List[Dict[str, Any]] = [] result = _run(step, captured) assert [c["bg_id"] for c in captured] == ["L03B01", "L10B01"] # 첫 렌더(그룹 anchor 자신) — anchor 첨부 없음, W-G fp 는 첨부. first = captured[0] assert first["building_anchor_path"] is None assert first["building_fp_path"] is not None assert BUILDING_ANCHOR_PLATE_GUIDANCE not in first["prompt"] # 두 번째 렌더(같은 그룹 다른 loc) — 첫 렌더 png 가 anchor 로 첨부. second = captured[1] assert second["building_anchor_path"] is not None assert second["building_anchor_path"].endswith("L03B01.png") assert BUILDING_ANCHOR_PLATE_GUIDANCE in second["prompt"] # 프롬프트 순서: anchor 지시 → building fp 지시 (첨부 순서와 정합). assert second["prompt"].index(BUILDING_ANCHOR_PLATE_GUIDANCE) < ( second["prompt"].index(BUILDING_FP_PLATE_GUIDANCE)) a = result["data"]["groups"]["L03B01"] b = result["data"]["groups"]["L10B01"] assert "building_anchor_ref" not in a assert b["building_anchor_ref"] == { "anchor_bg_id": "L03B01", "group_id": "bgrp1", } labels = b["attached_reference_lineage"]["attached_ref_labels"] assert "building_anchor:L03B01" in labels # anchor 라벨은 building_fp 라벨보다 앞 (첨부 순서 계약). assert labels.index("building_anchor:L03B01") < labels.index( "building_fp:fp_in") # prior_bg lineage 채널에는 섞이지 않는다. assert "L03B01" not in b["attached_reference_lineage"]["prior_bg_ids"] def test_anchor_flag_off_byte_identical(tmp_path, monkeypatch): _apply_settings(monkeypatch, tmp_path, anchor_on=False) cp_map, link = _two_loc_fixture(tmp_path) step = _new_step(cp_map, building_link=link) captured: List[Dict[str, Any]] = [] result = _run(step, captured) assert all(c["building_anchor_path"] is None for c in captured) assert all( BUILDING_ANCHOR_PLATE_GUIDANCE not in c["prompt"] for c in captured ) for entry in result["data"]["groups"].values(): assert "building_anchor_ref" not in entry for label in entry["attached_reference_lineage"]["attached_ref_labels"]: assert not label.startswith("building_anchor:") def test_anchor_without_wg_fp_ref_still_chains(tmp_path, monkeypatch): """W-G(fp 첨부) OFF 여도 W-I 는 링크 맵을 재사용해 anchor 만 체인한다.""" _apply_settings(monkeypatch, tmp_path, wg_fp_on=False) cp_map, link = _two_loc_fixture(tmp_path) step = _new_step(cp_map, building_link=link) captured: List[Dict[str, Any]] = [] result = _run(step, captured) # fp 첨부는 전부 없음 (W-G OFF). assert all(c["building_fp_path"] is None for c in captured) assert all( BUILDING_FP_PLATE_GUIDANCE not in c["prompt"] for c in captured ) # anchor 체인은 동작. assert captured[1]["building_anchor_path"].endswith("L03B01.png") b = result["data"]["groups"]["L10B01"] assert b["building_anchor_ref"]["anchor_bg_id"] == "L03B01" assert "building_fp_ref" not in b def test_no_link_no_anchor(tmp_path, monkeypatch): """링크 맵이 비면(그룹 없음) 등록/첨부 모두 없음 — 기존 경로.""" _apply_settings(monkeypatch, tmp_path) cp_map, _ = _two_loc_fixture(tmp_path, link=False) step = _new_step(cp_map, building_link={}) captured: List[Dict[str, Any]] = [] result = _run(step, captured) assert all(c["building_anchor_path"] is None for c in captured) for entry in result["data"]["groups"].values(): assert "building_anchor_ref" not in entry def test_anchor_dedup_when_prior_already_has_anchor_png( tmp_path, monkeypatch, ): """같은 loc 파생 노드의 substrate/catalog prior 에 anchor png 가 이미 있으면 중복 첨부하지 않는다 (경로 dedup).""" _apply_settings(monkeypatch, tmp_path) groups = [ { "gid": "g_l03", "fp_id": "fp_l03", "bg_specs": [ _bg_spec(bg_id="L03B01", fp_id="fp_l03", loc_id="L03"), _bg_spec(bg_id="L03B02", fp_id="fp_l03", loc_id="L03"), ], "plan": _plan_for_fp(fp_id="fp_l03", nodes=[ _plan_node(bg_id="L03B01", node_index=0), _plan_node( bg_id="L03B02", node_index=1, mode="reference_derived", is_anchor=False, selected_refs=[{ "ref_bg_id": "L03B01", "physical_space_id": "primary", "space_description": None, }], ), ]), }, ] cp_map = _build_cp_map_two_groups(tmp_path=tmp_path, groups=groups) indoor_fp_png = _touch_png(tmp_path / "fp_render" / "fp_in.png") link = {"L03": { "fp_id": "fp_in", "indoor_loc_sid": "L04", "group_id": "bgrp1", "png_path": str(indoor_fp_png), "fp_asset_id": "uuid-fp-in", }} step = _new_step(cp_map, building_link=link) captured: List[Dict[str, Any]] = [] result = _run(step, captured) second = captured[1] assert second["bg_id"] == "L03B02" # catalog prior 로 anchor png 가 이미 첨부됨 → anchor 채널 미사용. assert any(p.endswith("L03B01.png") for p in second["prior_bg_paths"]) assert second["building_anchor_path"] is None assert "building_anchor_ref" not in result["data"]["groups"]["L03B02"] def test_config_hash_stamps_flag_only_when_on(tmp_path, monkeypatch): _apply_settings(monkeypatch, tmp_path, anchor_on=True) monkeypatch.setattr( "app.core.config.settings.floor_plan_light_sidecar_enabled", False) cp_map, link = _two_loc_fixture(tmp_path) step_on = _new_step(cp_map, building_link=link) h_on = step_on._config_hash() monkeypatch.setattr( "app.core.config.settings.outdoor_building_anchor_ref_enabled", False) step_off = _new_step(cp_map, building_link=link) h_off = step_off._config_hash() assert h_on != h_off # ─── _register_image_assets: anchor UUID 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 _register_groups_fixture(tmp_path): p1 = _touch_png(tmp_path / "bgc" / "L03B01.png") p2 = _touch_png(tmp_path / "bgc" / "L10B01.png") groups = { "L03B01": { "status": "ok", "png_path": str(p1), "location_id": "L03", "shot_guides": [], "shot_ids": [], "t2i_prompt": "t", "attached_reference_lineage": { "prior_bg_ids": [], "ref_used": "refs_1", "attached_ref_labels": []}, }, "L10B01": { "status": "ok", "png_path": str(p2), "location_id": "L10", "shot_guides": [], "shot_ids": [], "t2i_prompt": "t", "building_anchor_ref": { "anchor_bg_id": "L03B01", "group_id": "bgrp1"}, "attached_reference_lineage": { "prior_bg_ids": [], "ref_used": "refs_2", "attached_ref_labels": ["building_anchor:L03B01"]}, }, } return groups, ["L03B01", "L10B01"] def _run_register(tmp_path, monkeypatch, groups, order): 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="L03", id="canon-l03"), SimpleNamespace(short_id="L10", id="canon-l10"), ] 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) return added, calls def test_register_merges_anchor_uuid_into_input_image_ids( tmp_path, monkeypatch, ): groups, order = _register_groups_fixture(tmp_path) added, calls = _run_register(tmp_path, monkeypatch, groups, order) row_by_vt = {r.variant_type: r for r in added} assert set(row_by_vt) == {"L03B01", "L10B01"} anchor_uuid = row_by_vt["L03B01"].id # L10B01 의 마지막 annotate(phase2 재기록)에 anchor UUID 가 병합된다. l10_calls = [kw for row, kw in calls if row is row_by_vt["L10B01"]] assert l10_calls, "L10B01 row 에 annotate 호출이 없다" final = l10_calls[-1] assert anchor_uuid in (final.get("input_image_ids") or []) def test_register_unresolved_anchor_marks_structural_key( tmp_path, monkeypatch, ): """anchor bg 렌더가 실패해 row 가 없으면 unresolved_inputs 구조키.""" groups, order = _register_groups_fixture(tmp_path) groups["L03B01"]["status"] = "failed" groups["L03B01"]["png_path"] = "" added, calls = _run_register(tmp_path, monkeypatch, groups, order) row_by_vt = {r.variant_type: r for r in added} assert set(row_by_vt) == {"L10B01"} l10_calls = [kw for row, kw in calls if row is row_by_vt["L10B01"]] final = l10_calls[-1] meta = final.get("pipeline_metadata") or {} assert "building_anchor:L03B01" in (meta.get("unresolved_inputs") or [])