"""W21B (2026-06-08) — dwelling_zone_map pure-module tests. Locks the DETERMINISTIC core of the dwelling zone map (the FP-image + VLM zone grouping that bypasses the edge-judge): dwelling target selection, VLM-output strict validation, the deterministic zone-map assembly / join, the edge-judge diagnostic comparison, and the synthetic fixture. The FP image generation and the VLM vision call are EXPERIMENTAL (provider, visual gate) and are NOT exercised here — only the deterministic join/contract is. Scenario leakage guard: every fixture below is synthetic structural data (roles / fp ids / grid points). No L05-specific lexicon drives any code path under test; the rooftop ids are used only as opaque join keys. """ from app.modules.pipeline.dwelling_zone_map import ( CONFIDENCE_FLOOR, GRID_SIZE, INTERIOR_SURFACE_ROLES, assemble_zone_map, build_analysis_from_numbered_elements, build_analysis_from_space_model, compare_to_edge_judge, compute_synthetic_zone_map, select_dwelling_targets, validate_vlm_zone_output, ) def _bg(bg_id, fp, role="interior_room", shots=None): return { "bg_id": bg_id, "depends_on_fp": [fp] if fp else [], "surface_role": role, "applies_to_shots": list(shots or []), } def _catalog(*entries): return {e["bg_id"]: e for e in entries} # ── dwelling target selection ──────────────────────────────────────── def test_select_groups_interior_bgs_by_fp(): catalog = _catalog( _bg("D1B01", "fp_a", shots=["S1_Shot1"]), _bg("D1B02", "fp_a", shots=["S2_Shot1", "S2_Shot2"]), _bg("D2B01", "fp_b", shots=["S3_Shot1"]), _bg("D2B02", "fp_b", shots=["S3_Shot2"]), ) out = select_dwelling_targets(background_catalog=catalog) assert set(out) == {"fp_a", "fp_b"} assert out["fp_a"]["bg_ids"] == ["D1B01", "D1B02"] assert out["fp_a"]["shot_ids"] == ["S1_Shot1", "S2_Shot1", "S2_Shot2"] assert out["fp_a"]["applicable"] is True assert out["fp_b"]["bg_ids"] == ["D2B01", "D2B02"] def test_select_excludes_exterior_and_site_bgs(): catalog = _catalog( _bg("L05B01", "fp_int", role="interior_room"), _bg("L05B02", "fp_int", role="interior_room"), _bg("L04B01", "fp_ext", role="exterior_plate"), _bg("L11B01", "fp_site", role="site_surface"), ) out = select_dwelling_targets(background_catalog=catalog) assert set(out) == {"fp_int"} assert "fp_ext" not in out assert "fp_site" not in out def test_select_single_bg_dwelling_not_applicable(): # a dwelling with one bg has nothing to cross-reference → not_applicable. catalog = _catalog(_bg("D1B01", "fp_solo")) out = select_dwelling_targets(background_catalog=catalog) assert out["fp_solo"]["applicable"] is False assert out["fp_solo"]["not_applicable_reason"] == "single_bg" def test_select_fp_less_bg_ignored(): catalog = _catalog( _bg("D1B01", None, role="interior_room"), _bg("D1B02", None, role="interior_room"), ) out = select_dwelling_targets(background_catalog=catalog) assert out == {} def test_interior_roles_constant_is_conservative(): # exterior / site / transition are NOT dwelling interiors. assert "interior_room" in INTERIOR_SURFACE_ROLES assert "exterior_plate" not in INTERIOR_SURFACE_ROLES assert "site_surface" not in INTERIOR_SURFACE_ROLES # ── VLM zone-output strict validation ──────────────────────────────── def _vlm_entry(bg_id, group, x, y, conf=0.9): return { "bg_id": bg_id, "room": "r", "zone_group": group, "grid": {"x": x, "y": y}, "confidence": conf, "evidence": "e", } def test_validate_vlm_output_accepts_well_formed(): raw = [ _vlm_entry("D1B01", "living", 40, 60), _vlm_entry("D1B02", "bedroom", 80, 70), ] res = validate_vlm_zone_output(raw=raw, expected_bg_ids=["D1B01", "D1B02"]) assert res["ok"] is True assert res["blockers"] == [] assert set(res["normalized"]) == {"D1B01", "D1B02"} assert res["normalized"]["D1B01"]["zone_group"] == "living" assert res["normalized"]["D1B01"]["grid"] == {"x": 40, "y": 60} def test_validate_vlm_output_missing_bg_fails(): raw = [_vlm_entry("D1B01", "living", 40, 60)] res = validate_vlm_zone_output(raw=raw, expected_bg_ids=["D1B01", "D1B02"]) assert res["ok"] is False assert any("D1B02" in b for b in res["blockers"]) def test_validate_vlm_output_unexpected_bg_fails(): raw = [ _vlm_entry("D1B01", "living", 40, 60), _vlm_entry("GHOST", "x", 10, 10), ] res = validate_vlm_zone_output(raw=raw, expected_bg_ids=["D1B01"]) assert res["ok"] is False assert any("GHOST" in b for b in res["blockers"]) def test_validate_vlm_output_grid_out_of_bounds_fails(): raw = [_vlm_entry("D1B01", "living", GRID_SIZE + 5, 60)] res = validate_vlm_zone_output(raw=raw, expected_bg_ids=["D1B01"]) assert res["ok"] is False assert any("grid" in b.lower() for b in res["blockers"]) def test_validate_vlm_output_confidence_out_of_range_fails(): raw = [_vlm_entry("D1B01", "living", 40, 60, conf=1.5)] res = validate_vlm_zone_output(raw=raw, expected_bg_ids=["D1B01"]) assert res["ok"] is False assert any("confidence" in b.lower() for b in res["blockers"]) def test_validate_vlm_output_empty_zone_group_fails(): raw = [_vlm_entry("D1B01", "", 40, 60)] res = validate_vlm_zone_output(raw=raw, expected_bg_ids=["D1B01"]) assert res["ok"] is False assert any("zone_group" in b for b in res["blockers"]) def test_validate_vlm_output_non_list_fails_closed(): res = validate_vlm_zone_output(raw={"not": "a list"}, expected_bg_ids=["D1B01"]) assert res["ok"] is False assert res["normalized"] == {} # ── zone-map assembly (deterministic join) ─────────────────────────── def _norm(bg_id, group, x, y, conf=0.9, room=None): return { "bg_id": bg_id, "room": room or group, "zone_group": group, "grid": {"x": x, "y": y}, "confidence": conf, "evidence": "e", } # Mirrors the PoC: living = B01,B03,B04 ; bedroom = B02,B05,B06. def _rooftop_normalized(): return { "L05B01": _norm("L05B01", "living", 44, 86, 0.93), "L05B02": _norm("L05B02", "bedroom", 78, 74, 0.86), "L05B03": _norm("L05B03", "living", 37, 60, 0.90), "L05B04": _norm("L05B04", "living", 38, 59, 0.95), "L05B05": _norm("L05B05", "bedroom", 79, 75, 0.88), "L05B06": _norm("L05B06", "bedroom", 79, 75, 0.88), } def _rooftop_bg_shots(): return { "L05B01": ["S5_Shot1"], "L05B02": ["S25_Shot4"], "L05B03": ["S18_Shot1"], "L05B04": ["S18_Shot13"], "L05B05": ["S12_Shot6"], "L05B06": ["S14_Shot4"], } def _assemble_rooftop(**over): kw = dict( fp_id="fp_rooftop_room_interior", vlm_normalized=_rooftop_normalized(), bg_shot_map=_rooftop_bg_shots(), clean_fp_ref="clean.png", annotated_fp_ref="annot.png", ) kw.update(over) return assemble_zone_map(**kw) def test_assemble_groups_two_zones_living_and_bedroom(): res = _assemble_rooftop() assert res["ok"] is True zm = res["zone_map"] assert zm["synthetic"] is False zones = zm["zones"] assert len(zones) == 2 # members partition into the PoC living/bedroom split. members = {zid: set(z["member_bg_ids"]) for zid, z in zones.items()} assert {"L05B01", "L05B03", "L05B04"} in members.values() assert {"L05B02", "L05B05", "L05B06"} in members.values() def test_assemble_zone_id_is_first_seen_deterministic(): # first bg in sorted order (L05B01 = living) anchors Z01. zm = _assemble_rooftop()["zone_map"] assert zm["bg_zone_assignments"]["L05B01"]["zone_id"] == "Z01" assert zm["bg_zone_assignments"]["L05B02"]["zone_id"] == "Z02" assert zm["bg_zone_assignments"]["L05B03"]["zone_id"] == "Z01" def test_assemble_bg_assignment_carries_grid_and_shots(): zm = _assemble_rooftop()["zone_map"] a = zm["bg_zone_assignments"]["L05B01"] assert a["grid_focus"] == {"x": 44, "y": 86} assert a["shot_ids"] == ["S5_Shot1"] assert a["assignment_state"] == "ok" def test_assemble_shot_assignments_invert_bg_binding(): zm = _assemble_rooftop()["zone_map"] sa = zm["shot_zone_assignments"] assert sa["S5_Shot1"]["bg_id"] == "L05B01" assert sa["S5_Shot1"]["zone_id"] == "Z01" assert sa["S25_Shot4"]["zone_id"] == "Z02" def test_assemble_zone_bbox_covers_member_points(): zm = _assemble_rooftop()["zone_map"] for z in zm["zones"].values(): bb = z["grid_bbox"] for bg_id in z["member_bg_ids"]: g = zm["bg_zone_assignments"][bg_id]["grid_focus"] assert bb["x"] <= g["x"] <= bb["x"] + bb["w"] assert bb["y"] <= g["y"] <= bb["y"] + bb["h"] def test_assemble_annotated_ref_is_render_forbidden_and_separate(): zm = _assemble_rooftop()["zone_map"] assert zm["clean_fp_ref"] == "clean.png" ann = zm["annotated_fp_ref"] assert ann["ref"] == "annot.png" assert ann["must_not_be_used_for_render"] is True assert ann["asset_role"] == "vlm_mapping_only" # the clean render ref must never equal the annotated (number-leak) ref. assert zm["clean_fp_ref"] != ann["ref"] def test_assemble_single_zone_marks_not_applicable(): # every bg in one zone → no cross-space problem → degenerate. one_zone = { "D1B01": _norm("D1B01", "living", 40, 60), "D1B02": _norm("D1B02", "living", 42, 62), } res = _assemble_rooftop( vlm_normalized=one_zone, bg_shot_map={"D1B01": ["S1_Shot1"], "D1B02": ["S2_Shot1"]}, ) zm = res["zone_map"] assert zm["applicable"] is False assert zm["not_applicable_reason"] == "single_zone" assert zm["diagnostics"]["zone_count"] == 1 def test_assemble_low_confidence_flagged_and_ambiguous(): low = _rooftop_normalized() low["L05B02"]["confidence"] = 0.3 # below floor zm = _assemble_rooftop(vlm_normalized=low)["zone_map"] assert "L05B02" in zm["diagnostics"]["low_confidence"] assert zm["bg_zone_assignments"]["L05B02"]["assignment_state"] == "ambiguous" def test_assemble_structure_cues_exact_match_only(): # structure_cues come from an EXACT room-name join, never a fuzzy match. space = {"rooms": [ {"name": "living", "fixtures": ["TV", "식탁"]}, {"name": "bedroom", "fixtures": ["침대", "창문"]}, ]} zm = _assemble_rooftop(space_analysis=space)["zone_map"] by_label = {z["label"]: z for z in zm["zones"].values()} assert by_label["living"]["structure_cues"] == ["TV", "식탁"] assert by_label["bedroom"]["structure_cues"] == ["침대", "창문"] def test_confidence_floor_is_conservative(): assert 0.0 < CONFIDENCE_FLOOR < 0.75 # ── floor_plan_prompt.space_model → analysis adapter (structure SOT) ── def _rooftop_space_model(): # the real fp_rooftop_room_interior space_model shape: 투룸 + 화장실 + 주방. return { "space_type": "interior_room", "zones": [ {"zone_id": "entry_threshold", "zone_type": "entry", "essential_elements": ["metal_entry_door", "shoe_step"], "openings": ["interior_passage_to_living"]}, {"zone_id": "living_dining", "zone_type": "living", "essential_elements": ["tv", "dining_table", "chairs"], "openings": ["door_to_entry", "doors_to_bedrooms"]}, {"zone_id": "sink_alcove", "zone_type": "sink", "essential_elements": ["sink", "cooktop"], "openings": ["open_passage_to_living"]}, {"zone_id": "main_bedroom", "zone_type": "bedroom", "essential_elements": ["bed", "wardrobe"], "openings": ["door_to_living"]}, {"zone_id": "daughter_bedroom", "zone_type": "bedroom", "essential_elements": ["bed", "curtain"], "openings": ["door_to_living"]}, {"zone_id": "bathroom", "zone_type": "bath", "essential_elements": ["washbasin", "toilet", "shower"], "openings": ["door_to_living"]}, ], } def test_space_model_adapter_preserves_every_zone(): # the bug we fix: a sparse input dropped the bathroom + 2nd bedroom. The # adapter must carry EVERY structural zone through, never collapse them. a = build_analysis_from_space_model(_rooftop_space_model()) names = {r["name"] for r in a["rooms"]} assert names == { "entry_threshold", "living_dining", "sink_alcove", "main_bedroom", "daughter_bedroom", "bathroom", } # two distinct bedrooms are kept distinct. beds = [r for r in a["rooms"] if r["label"] == "bedroom"] assert len(beds) == 2 def test_space_model_adapter_carries_fixtures(): a = build_analysis_from_space_model(_rooftop_space_model()) bath = next(r for r in a["rooms"] if r["name"] == "bathroom") assert bath["fixtures"] == ["washbasin", "toilet", "shower"] assert a["space_type"] == "interior_room" def test_space_model_adapter_empty_or_bad_returns_no_rooms(): assert build_analysis_from_space_model(None)["rooms"] == [] assert build_analysis_from_space_model({"zones": []})["rooms"] == [] assert build_analysis_from_space_model({"zones": "nope"})["rooms"] == [] # ── floor_plan_prompt.numbered_elements → analysis adapter (v6+ structure SOT; # no space_model is ever emitted by the prompt packs, so this is the real # production path) ── def _rooftop_numbered_elements(): # the real fp_l04_rooftop_room_main numbered_elements shape: 7 area zones # (투룸 + 화장실 + 주방 + 거실 + 현관 + 옥상), openings, persistent fixtures, # and state-overlay props. category/base_layer_decision are the closed enums # the adapter reads — never the label prose. return [ {"number": 1, "category": "area", "base_layer_decision": "base_structural_unit", "label": "entry threshold zone"}, {"number": 2, "category": "area", "base_layer_decision": "base_structural_unit", "label": "living dining shared zone"}, {"number": 3, "category": "area", "base_layer_decision": "base_structural_unit", "label": "kitchenette service strip"}, {"number": 4, "category": "area", "base_layer_decision": "base_structural_unit", "label": "main bedroom private zone"}, {"number": 5, "category": "area", "base_layer_decision": "base_structural_unit", "label": "second bedroom private zone"}, {"number": 6, "category": "area", "base_layer_decision": "base_structural_unit", "label": "bathroom service zone"}, {"number": 7, "category": "area", "base_layer_decision": "base_structural_unit", "label": "rooftop exterior landing"}, {"number": 8, "category": "opening", "base_layer_decision": "base_opening", "label": "iron entrance door"}, {"number": 10, "category": "opening", "base_layer_decision": "base_opening", "label": "main bedroom door"}, {"number": 14, "category": "furniture", "base_layer_decision": "base_persistent_fixture", "label": "kitchen sink and faucet"}, {"number": 19, "category": "furniture", "base_layer_decision": "base_persistent_furniture", "label": "main bedroom bed"}, {"number": 28, "category": "prop", "base_layer_decision": "state_overlay_plot_cue", "label": "blood stains and footprint trail"}, {"number": 31, "category": "prop", "base_layer_decision": "state_overlay_plot_cue", "label": "red circle on bedroom wall"}, ] def test_numbered_elements_adapter_keeps_all_structural_areas(): # the bug we fix: code expected space_model.zones (never emitted), so it fell # to bg_blocks_fallback and collapsed 투룸 → 1침실. The adapter must carry # EVERY area/base_structural_unit zone through. a = build_analysis_from_numbered_elements(_rooftop_numbered_elements()) assert len(a["rooms"]) == 7 assert a["source"] == "floor_plan_prompt.numbered_elements" # name is the stable E id; raw label carried verbatim. by_name = {r["name"]: r["label"] for r in a["rooms"]} assert by_name["E4"] == "main bedroom private zone" assert by_name["E5"] == "second bedroom private zone" def test_numbered_elements_adapter_excludes_state_overlays_and_global_fixtures(): a = build_analysis_from_numbered_elements(_rooftop_numbered_elements()) room_labels = {r["label"] for r in a["rooms"]} # transient props / plot cues never enter the clean structure. assert "blood stains and footprint trail" not in room_labels assert "red circle on bedroom wall" not in room_labels # openings only from base_opening; fixtures carried GLOBALLY, not per-room. assert "iron entrance door" in a["openings"] assert "kitchen sink and faucet" in a["global_fixtures"] assert "main bedroom bed" in a["global_fixtures"] assert all(r["fixtures"] == [] for r in a["rooms"]) # no per-room semantic inference def test_numbered_elements_adapter_distinct_rooms_on_duplicate_labels(): ne = [ {"number": 1, "category": "area", "base_layer_decision": "base_structural_unit", "label": "bedroom"}, {"number": 2, "category": "area", "base_layer_decision": "base_structural_unit", "label": "bedroom"}, ] a = build_analysis_from_numbered_elements(ne) # same label, distinct number-based names → two distinct rooms (no merge). assert len(a["rooms"]) == 2 assert {r["name"] for r in a["rooms"]} == {"E1", "E2"} def test_numbered_elements_adapter_empty_or_no_structural_areas(): assert build_analysis_from_numbered_elements(None)["rooms"] == [] assert build_analysis_from_numbered_elements([])["rooms"] == [] # only openings/props, zero structural areas → no rooms (caller falls back). only_non_area = [ {"number": 1, "category": "opening", "base_layer_decision": "base_opening", "label": "door"}, {"number": 2, "category": "prop", "base_layer_decision": "state_overlay_plot_cue", "label": "blood"}, ] assert build_analysis_from_numbered_elements(only_non_area)["rooms"] == [] # ── edge-judge diagnostic comparison ───────────────────────────────── def test_compare_surfaces_pairs_zone_map_merged(): # zone_map unifies B01+B03 (Z01); the edge-judge split them (g1 vs g2). zm = _assemble_rooftop( vlm_normalized={ "B01": _norm("B01", "living", 40, 60), "B02": _norm("B02", "bedroom", 80, 70), "B03": _norm("B03", "living", 42, 62), }, bg_shot_map={"B01": ["S1"], "B02": ["S2"], "B03": ["S3"]}, )["zone_map"] plan = {"node_assignments": {"B01": "g1", "B02": "g3", "B03": "g2"}} diag = compare_to_edge_judge(zone_map=zm, partition_plan=plan) assert diag["edge_judge_available"] is True assert ["B01", "B03"] in diag["merged_by_zone_map"] assert diag["split_by_zone_map"] == [] def test_compare_counts_agreement(): zm = _assemble_rooftop( vlm_normalized={ "B01": _norm("B01", "living", 40, 60), "B02": _norm("B02", "bedroom", 80, 70), }, bg_shot_map={"B01": ["S1"], "B02": ["S2"]}, )["zone_map"] # edge-judge also splits them → both methods agree they are apart. plan = {"node_assignments": {"B01": "g1", "B02": "g2"}} diag = compare_to_edge_judge(zone_map=zm, partition_plan=plan) assert diag["agreement_pair_count"] == 1 assert diag["merged_by_zone_map"] == [] def test_compare_no_plan_marks_unavailable(): zm = _assemble_rooftop()["zone_map"] diag = compare_to_edge_judge(zone_map=zm, partition_plan=None) assert diag["edge_judge_available"] is False assert diag["merged_by_zone_map"] == [] # ── synthetic fixture (non-authoritative) ──────────────────────────── def test_synthetic_fixture_is_non_authoritative_and_shaped(): zm = compute_synthetic_zone_map( fp_id="fp_x", bg_ids=["B01", "B02"], bg_shot_map={"B01": ["S1"]}, ) assert zm["synthetic"] is True assert zm["applicable"] is False # the contract shape is present so downstream code can read it uniformly. assert set(zm["bg_zone_assignments"]) == {"B01", "B02"} assert "zones" in zm and "shot_zone_assignments" in zm assert zm["annotated_fp_ref"]["must_not_be_used_for_render"] is True