"""W21B (2026-06-08): DwellingZoneMapStep gate + path tests. opt-in (default OFF). Covers gating → byte-stable not_applicable, the synthetic plumbing path (enabled + real_provider OFF), the real path via injected overrides (no network / spend), and the per-dwelling synthetic fallback on a provider miss (failed_count held at 0). The real LLM/VLM/image clients are bypassed via the override slots. """ from __future__ import annotations from contextlib import contextmanager from pathlib import Path from unittest.mock import MagicMock from app.core.config import settings from app.core.steps.dwelling_zone_map_step import ( SCHEMA_VERSION, DwellingZoneMapStep, ) @contextmanager def _settings(**kw): old = {k: getattr(settings, k) for k in kw} for k, v in kw.items(): setattr(settings, k, v) try: yield finally: for k, v in old.items(): setattr(settings, k, v) def _bg(bg_id, fp, role="interior_room", shots=None, sub="", state=""): return { "bg_id": bg_id, "depends_on_fp": [fp], "surface_role": role, "applies_to_shots": list(shots or []), "sub_location_label": sub, "state_label_raw": state, } def _mp_cp(catalog): return {"data": {"background_catalog": catalog}} def _new_step(cp_map): step = DwellingZoneMapStep.__new__(DwellingZoneMapStep) step.project_id = "p" step.episode_id = "e" step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) step._analyze_override = None step._fp_prompt_override = None step._render_override = None step._vlm_override = None step._annotate_override = None step._client = None step._load_prev_checkpoint = MagicMock(side_effect=lambda sid: cp_map.get(sid)) return step _TWO_BG = { "L05B01": _bg("L05B01", "fp_int", shots=["S5_Shot1"], sub="거실"), "L05B02": _bg("L05B02", "fp_int", shots=["S25_Shot4"], sub="침실"), "L05B03": _bg("L05B03", "fp_int", shots=["S18_Shot1"], sub="거실"), } # ── gating ──────────────────────────────────────────────────────────── def test_disabled_returns_not_applicable_byte_stable(): step = _new_step({"background_master_plan": _mp_cp(_TWO_BG)}) with _settings(background_mode="on", dwelling_zone_map_enabled=False, floor_plan_prompt_version="6"): res = step._execute() assert res["applicable_count"] == 0 assert res["completed_count"] == 0 assert res["failed_count"] == 0 assert res["data"] == {} assert res["schema_version"] == SCHEMA_VERSION def test_background_mode_off_not_applicable(): step = _new_step({"background_master_plan": _mp_cp(_TWO_BG)}) with _settings(background_mode="off", dwelling_zone_map_enabled=True, floor_plan_prompt_version="6"): res = step._execute() assert res["data"] == {} # ── synthetic plumbing path ──────────────────────────────────────────── def test_enabled_synthetic_path_emits_fixtures(): step = _new_step({"background_master_plan": _mp_cp(_TWO_BG)}) with _settings(background_mode="on", dwelling_zone_map_enabled=True, dwelling_zone_map_real_provider_enabled=False, floor_plan_prompt_version="6"): res = step._execute() assert res["applicable_count"] == 1 assert res["failed_count"] == 0 per = res["data"]["per_dwelling"] assert "fp_int" in per assert per["fp_int"]["synthetic"] is True assert per["fp_int"]["applicable"] is False assert set(per["fp_int"]["bg_zone_assignments"]) == {"L05B01", "L05B02", "L05B03"} assert res["data"]["synthetic_count"] == 1 assert res["data"]["real_count"] == 0 # ── real path via overrides ──────────────────────────────────────────── def _analyze(**_): return {"space_type": "x", "summary": "s", "rooms": [ {"name": "living", "same_space_as": None, "fixtures": ["TV"]}, {"name": "bedroom", "same_space_as": None, "fixtures": ["침대"]}, ], "openings": []} def _vlm(**_): return [ {"bg_id": "L05B01", "room": "living", "grid": {"x": 40, "y": 80}, "zone_group": "living", "confidence": 0.93, "evidence": "e"}, {"bg_id": "L05B02", "room": "bedroom", "grid": {"x": 80, "y": 70}, "zone_group": "bedroom", "confidence": 0.88, "evidence": "e"}, {"bg_id": "L05B03", "room": "living", "grid": {"x": 38, "y": 60}, "zone_group": "living", "confidence": 0.9, "evidence": "e"}, ] def _render_ok(*, prompt, out_path, fp_id): Path(out_path).parent.mkdir(parents=True, exist_ok=True) Path(out_path).write_bytes(b"PNGDATA") return {"status": "ok", "png_path": str(out_path), "error": ""} def _annotate_ok(*, clean_png, out_path, zone_map): return str(out_path) def _real_step(tmp_path, cp_map): step = _new_step(cp_map) step.set_overrides_for_testing( analyze=_analyze, fp_prompt=lambda **_: "draw a clean plan", render=_render_ok, vlm=_vlm, annotate=_annotate_ok) return step def test_real_path_produces_zone_map(tmp_path): step = _real_step(tmp_path, {"background_master_plan": _mp_cp(_TWO_BG)}) with _settings(background_mode="on", dwelling_zone_map_enabled=True, dwelling_zone_map_real_provider_enabled=True, floor_plan_prompt_version="6", projects_dir=str(tmp_path)): res = step._execute() assert res["data"]["real_count"] == 1 assert res["failed_count"] == 0 zm = res["data"]["per_dwelling"]["fp_int"] assert zm["synthetic"] is False assert len(zm["zones"]) == 2 assert zm["bg_zone_assignments"]["L05B01"]["zone_id"] == "Z01" assert zm["bg_zone_assignments"]["L05B02"]["zone_id"] == "Z02" # clean vs annotated separation + render-forbidden flag. assert zm["clean_fp_ref"] assert zm["annotated_fp_ref"]["must_not_be_used_for_render"] is True assert zm["clean_fp_ref"] != zm["annotated_fp_ref"]["ref"] assert zm["diagnostics"]["edge_judge_comparison"]["edge_judge_available"] is False def test_real_path_provider_failure_falls_back_synthetic(tmp_path): def _boom(**_): raise RuntimeError("vlm exploded") step = _real_step(tmp_path, {"background_master_plan": _mp_cp(_TWO_BG)}) step.set_overrides_for_testing( analyze=_analyze, fp_prompt=lambda **_: "p", render=_render_ok, vlm=_boom, annotate=_annotate_ok) with _settings(background_mode="on", dwelling_zone_map_enabled=True, dwelling_zone_map_real_provider_enabled=True, floor_plan_prompt_version="6", projects_dir=str(tmp_path)): res = step._execute() assert res["failed_count"] == 0 # provider miss is never a step failure assert res["data"]["fallback_count"] == 1 zm = res["data"]["per_dwelling"]["fp_int"] assert zm["synthetic"] is True assert "vlm exploded" in zm["diagnostics"]["fallback_reason"] def test_single_bg_dwelling_excluded_from_processing(): one = {"L05B01": _bg("L05B01", "fp_solo", shots=["S1_Shot1"])} step = _new_step({"background_master_plan": _mp_cp(one)}) with _settings(background_mode="on", dwelling_zone_map_enabled=True, dwelling_zone_map_real_provider_enabled=False, floor_plan_prompt_version="6"): res = step._execute() # the only dwelling is single-bg → not applicable → empty not_applicable. assert res["data"] == {} # ── input-SOT boundaries (Codex W21B re-review NARROW_2) ────────────── def _space_model_sm(): return {"space_type": "interior_room", "zones": [ {"zone_id": "living_dining", "zone_type": "living", "essential_elements": ["tv"], "openings": ["open_passage_to_sink"]}, {"zone_id": "bedroom_a", "zone_type": "bedroom", "essential_elements": ["bed"], "openings": ["door_to_living"]}, ]} def test_fresh_t2i_gate_only_when_hash_matches(): cp = {"data": {"consumed_bg_catalog_hash": "H1", "backgrounds": {"B1": {"t2i_prompt": "p1"}}}} assert DwellingZoneMapStep._fresh_t2i_by_bg(cp, "H1") == {"B1": "p1"} # fresh assert DwellingZoneMapStep._fresh_t2i_by_bg(cp, "H2") == {} # stale hash assert DwellingZoneMapStep._fresh_t2i_by_bg(cp, None) == {} # no master hash assert DwellingZoneMapStep._fresh_t2i_by_bg(None, "H1") == {} # absent def test_structural_facts_keeps_numeric_only(): facts = {"target_units": [3, 7], "dominant": 5, "clean": True, "label": "prose text", "nested": {"x": 1}, "mixed": [1, "a"]} out = DwellingZoneMapStep._structural_facts(facts) assert out == {"target_units": [3, 7], "dominant": 5, "clean": True} assert DwellingZoneMapStep._structural_facts(None) is None assert DwellingZoneMapStep._structural_facts({"only_prose": "x"}) is None def test_bg_description_priority_and_dossier_enrichment(): step = _new_step({}) proj = {"data": {"cards": {"c1": {"bg_id": "B1", "vlm_output": { "bg_plate_visible_description": "plate prose B1"}}}}} catalog = {"B1": _bg("B1", "fp"), "B2": _bg("B2", "fp", sub="subB2"), "B3": _bg("B3", "fp", sub="subB3")} blocks = step._assemble_bg_descriptions( bg_ids=["B1", "B2", "B3"], catalog=catalog, projection_cp=proj, fresh_t2i_by_bg={"B2": "t2i B2"}, dossier_facts={"B1": {"target_units": [3]}}) by = {b["bg_id"]: b for b in blocks} # priority: projection_card > fresh background_prompt > catalog. assert by["B1"]["description"] == "plate prose B1" assert by["B2"]["description"] == "t2i B2" assert by["B2"]["source"] == "background_prompt_fresh" assert by["B3"]["source"] == "catalog_label" # dossier facts attached only where present (real enrichment, not primary). assert by["B1"]["structure_facts"] == {"target_units": [3]} assert by["B1"]["source"] == "projection_card+dossier" assert "structure_facts" not in by["B2"] def test_real_path_feeds_space_model_to_fp_and_dossier_to_vlm(tmp_path): captured = {} def _cap_fp(**kw): captured["space_model"] = kw.get("space_model") return "clean plan prompt" def _cap_vlm(**kw): captured["bg_blocks"] = kw.get("bg_blocks") return _vlm() fpp = {"data": {"floor_plans": {"fp_int": {"space_model": _space_model_sm()}}}} dossier = {"data": {"dossiers": {"fp_int": { "per_bg_render_facts_by_bg_id": {"L05B01": {"target_units": [9]}}}}}} cp = {"background_master_plan": _mp_cp(_TWO_BG), "floor_plan_prompt": fpp, "base_location_dossier": dossier} step = _new_step(cp) step.set_overrides_for_testing( analyze=_analyze, fp_prompt=_cap_fp, render=_render_ok, vlm=_cap_vlm, annotate=_annotate_ok) with _settings(background_mode="on", dwelling_zone_map_enabled=True, dwelling_zone_map_real_provider_enabled=True, floor_plan_prompt_version="6", projects_dir=str(tmp_path)): res = step._execute() # floor_plan_prompt.space_model actually reaches the FP prompt builder. assert captured["space_model"] is not None assert captured["space_model"]["zones"] # dossier structural facts are merged into the VLM bg_blocks (not just provenance). b01 = next(b for b in captured["bg_blocks"] if b["bg_id"] == "L05B01") assert b01.get("structure_facts") == {"target_units": [9]} prov = res["data"]["provenance"]["fp_int"] assert prov["structure_source"] == "space_model" assert "L05B01" in prov["dossier_enriched_bgs"] def test_real_path_numbered_elements_is_structure_source(tmp_path): # The production reality: no fp prompt pack emits space_model — it emits # numbered_elements. The adapter must fire so structure_source is # 'numbered_elements', NOT 'bg_blocks_fallback' (the P0 bug that collapsed # 투룸 → 1침실 + split the living room). fpp = {"data": {"floor_plans": {"fp_int": {"numbered_elements": [ {"number": 1, "category": "area", "base_layer_decision": "base_structural_unit", "label": "living dining"}, {"number": 2, "category": "area", "base_layer_decision": "base_structural_unit", "label": "bedroom"}, ]}}}} cp = {"background_master_plan": _mp_cp(_TWO_BG), "floor_plan_prompt": fpp} step = _new_step(cp) step.set_overrides_for_testing( analyze=_analyze, fp_prompt=lambda **kw: "clean plan prompt", render=_render_ok, vlm=lambda **kw: _vlm(), annotate=_annotate_ok) with _settings(background_mode="on", dwelling_zone_map_enabled=True, dwelling_zone_map_real_provider_enabled=True, floor_plan_prompt_version="6", projects_dir=str(tmp_path)): res = step._execute() prov = res["data"]["provenance"]["fp_int"] assert prov["structure_source"] == "numbered_elements" def test_real_path_space_model_wins_over_numbered(tmp_path): # both present → space_model stays the higher-priority structure source # (byte-identical to the space_model-only path). fpp = {"data": {"floor_plans": {"fp_int": { "space_model": _space_model_sm(), "numbered_elements": [{"number": 1, "category": "area", "base_layer_decision": "base_structural_unit", "label": "ignored"}]}}}} cp = {"background_master_plan": _mp_cp(_TWO_BG), "floor_plan_prompt": fpp} step = _new_step(cp) step.set_overrides_for_testing( analyze=_analyze, fp_prompt=lambda **kw: "clean plan prompt", render=_render_ok, vlm=lambda **kw: _vlm(), annotate=_annotate_ok) with _settings(background_mode="on", dwelling_zone_map_enabled=True, dwelling_zone_map_real_provider_enabled=True, floor_plan_prompt_version="6", projects_dir=str(tmp_path)): res = step._execute() prov = res["data"]["provenance"]["fp_int"] assert prov["structure_source"] == "space_model"