"""W20A: base_location_dossier pure-function unit tests. LLM / image / VLM API call 0. DB / ImageAsset write 0. Deterministic. Focused tests prove the dossier contract: - 5 dwelling identity axes present. - FP geometry candidates shape declared with VLM-dependent slots empty (W20A leaves them for the VLM wiring wave). - VLM 10x10 readback emits synthetic_placeholder ONLY — code does not invoke any real VLM in W20A. - Anchor selection is NOT performed — only the candidate surface is surfaced; ``selected_anchor_bg_id`` stays None per W20 spec §4.4. - Exact-ID partition is re-validated on the dossier consumer side. - Generic enum / shape-kind labels carry no scenario-specific tokens in the static contract; runtime labels pass through verbatim as data. """ from __future__ import annotations import pytest from app.modules.pipeline.base_location_dossier import ( ALL_LAYER_DECISIONS, BASE_LAYER_DECISIONS, DEFAULT_GRID_SIZE, STANDARD_OF_LIVING_BAND_DEFAULT, STATE_OVERLAY_DECISIONS, VIEW_CONE_LENS_ENUM_MAP, BaseLocationDossierError, build_dossiers, ) # ---------- fixtures ---------- def _fp_prompt_data() -> dict: return { "floor_plans": { "fp_a": { "status": "ok", "fp_id": "fp_a", "t2i_prompt": "x" * 50, "key_elements": [], "numbered_elements": [ { "number": 1, "label": "primary unit", "category": "area", "position_hint": "center", "base_layer_decision": "base_structural_unit", }, { "number": 2, "label": "secondary unit", "category": "area", "position_hint": "east side", "base_layer_decision": "base_structural_unit", }, { "number": 3, "label": "interior opening", "category": "opening", "position_hint": "between 1 and 2", "base_layer_decision": "base_opening", }, { "number": 4, "label": "service counter", "category": "furniture", "position_hint": "north wall of unit 1", "base_layer_decision": "base_persistent_fixture", }, { "number": 5, "label": "anchor seating", "category": "furniture", "position_hint": "south of unit 1", "base_layer_decision": "base_persistent_furniture", }, { "number": 6, "label": "transient cue", "category": "prop", "position_hint": "north wall", "base_layer_decision": "state_overlay_plot_cue", }, ], "camera_recommendations": [ { "bg_id": "L01B01", "sub_location": "primary_unit", "camera_position": "near number 1, facing east", "camera_height": "eye level", "lens_hint": "35mm", "framing_notes": "", "use_numbered_elements": [1, 3, 4], "ignore_numbered_elements": [6], }, { "bg_id": "L01B02", "sub_location": "secondary_unit", "camera_position": "inside number 2", "camera_height": "eye level", "lens_hint": "35mm", "framing_notes": "", "use_numbered_elements": [2, 6], "ignore_numbered_elements": [], }, ], } } } def _master_plan_data() -> dict: return { "plans": { "g_main": { "status": "ok", "plan": { "group_id": "g_main", "rationale_summary": "", "floor_plans": [], "backgrounds": [ { "bg_id": "L01B01", "loc_id": "L01", "sub_location": "primary_unit", "state_label_raw": "clean", "applies_to_shots": ["S1_Shot1"], "depends_on_fp": ["fp_a"], "depends_on_bg": [], }, { "bg_id": "L01B02", "loc_id": "L01", "sub_location": "secondary_unit", "state_label_raw": "dirty", "applies_to_shots": ["S1_Shot2"], "depends_on_fp": ["fp_a"], "depends_on_bg": ["L01B01"], }, ], "gen_order": ["L01B01", "L01B02"], }, } } } def _overlay_payload_data() -> dict: # Mirrors floor_plan_overlay_payload.build_overlay_payload output. return { "overlays": { "L01B01": { "bg_id": "L01B01", "fp_id": "fp_a", "use_numbered_elements": [1, 3, 4], "ignore_numbered_elements": [6], "base_markers_to_reference": [ { "number": 1, "label": "primary unit", "category": "area", "position_hint": "center", "base_layer_decision": "base_structural_unit", }, { "number": 3, "label": "interior opening", "category": "opening", "position_hint": "between 1 and 2", "base_layer_decision": "base_opening", }, { "number": 4, "label": "service counter", "category": "furniture", "position_hint": "north wall of unit 1", "base_layer_decision": "base_persistent_fixture", }, ], "transient_markers_to_describe": [], "ignored_state_overlay_markers": [ { "number": 6, "label": "transient cue", "category": "prop", "position_hint": "north wall", "base_layer_decision": "state_overlay_plot_cue", }, ], "target_unit_marker_numbers": [1], "dominant_target_unit_marker_number": 1, "clean_background_expected": True, "diagnostics": [], }, "L01B02": { "bg_id": "L01B02", "fp_id": "fp_a", "use_numbered_elements": [2, 6], "ignore_numbered_elements": [], "base_markers_to_reference": [ { "number": 2, "label": "secondary unit", "category": "area", "position_hint": "east side", "base_layer_decision": "base_structural_unit", }, ], "transient_markers_to_describe": [ { "number": 6, "label": "transient cue", "category": "prop", "position_hint": "north wall", "base_layer_decision": "state_overlay_plot_cue", }, ], "ignored_state_overlay_markers": [], "target_unit_marker_numbers": [2], "dominant_target_unit_marker_number": 2, "clean_background_expected": False, "diagnostics": [], }, } } def _fp_render_data() -> dict: return { "floor_plans": { "fp_a": {"png_path": "/projects/p/checkpoints/.../fp_a.png"} } } # ---------- happy path ---------- def test_dossier_emits_five_identity_axes_for_valid_fp(): out = build_dossiers( fp_prompt_data=_fp_prompt_data(), master_plan_data=_master_plan_data(), overlay_payload_data=_overlay_payload_data(), fp_render_data=_fp_render_data(), ) assert "fp_a" in out dossier = out["fp_a"] assert dossier["fp_id"] == "fp_a" assert dossier["fp_image_path"] == "/projects/p/checkpoints/.../fp_a.png" ident = dossier["dwelling_identity"] # All five axes present (W20 spec §4.1). assert set(ident.keys()) == { "structure", "materials", "fixed_elements_inventory", "standard_of_living_band", "lighting_identity", } assert ident["standard_of_living_band"] == STANDARD_OF_LIVING_BAND_DEFAULT # Structure partitions all base_* markers by exact-ID enum. structure = ident["structure"] assert structure["unit_markers"] == [1, 2] assert structure["opening_markers"] == [3] assert structure["fixed_fixture_markers"] == [4] assert structure["anchor_furniture_markers"] == [5] assert structure["marker_count_per_layer"]["base_structural_unit"] == 2 # Fixed elements inventory carries runtime label verbatim # (label="service counter" is generic shape-kind; the test simulates # both generic and any future runtime label passing through). fixed = ident["fixed_elements_inventory"] assert len(fixed) == 1 assert fixed[0]["number"] == 4 assert fixed[0]["label"] == "service counter" # LLM-emit slots empty in W20A. assert ident["materials"]["diagnostics"] # non-empty diagnostic assert ident["lighting_identity"]["diagnostics"] def test_dossier_geometry_candidates_shape_declared_with_vlm_slots_empty(): out = build_dossiers( fp_prompt_data=_fp_prompt_data(), master_plan_data=_master_plan_data(), overlay_payload_data=_overlay_payload_data(), fp_render_data=_fp_render_data(), ) geom = out["fp_a"]["fp_geometry_candidates"] assert geom["grid_size"] == list(DEFAULT_GRID_SIZE) # VLM-dependent slots — empty in W20A by contract. assert geom["camera_cell_candidates_per_unit"] == {} assert geom["look_at_cell_candidates_per_unit"] == {} assert geom["visible_units_candidates"] == {} assert geom["visible_openings_candidates"] == {} assert geom["wall_door_invalidation_diagnostics"] == [] # View cone lens enum map — static, exact equality. assert geom["view_cone_lens_enum_map"] == dict(VIEW_CONE_LENS_ENUM_MAP) # Diagnostic mentions VLM placeholder dependency. assert any("VLM" in d for d in geom["diagnostics"]) def test_dossier_vlm_readback_is_synthetic_placeholder_only(): """W20A must NEVER produce status='ok' on the VLM gate. A real VLM call would flip status to 'ok' / 'failed'. The dossier must stay on 'synthetic_placeholder' with 'synthetic_pass' gate decision until a follow-up wave wires a real provider. """ out = build_dossiers( fp_prompt_data=_fp_prompt_data(), master_plan_data=_master_plan_data(), overlay_payload_data=_overlay_payload_data(), fp_render_data=_fp_render_data(), ) readback = out["fp_a"]["fp_geometry_vlm_readback"] assert readback["status"] == "synthetic_placeholder" assert readback["gate_decision"] == "synthetic_pass" # Observation fields stay None — no fabricated content. assert readback["observed_markers"] is None assert readback["missing_markers"] is None assert readback["extra_markers"] is None assert readback["confidence"] is None assert any("not invoked" in d for d in readback["diagnostics"]) def test_dossier_anchor_selection_is_surface_only_never_picks(): out = build_dossiers( fp_prompt_data=_fp_prompt_data(), master_plan_data=_master_plan_data(), overlay_payload_data=_overlay_payload_data(), fp_render_data=_fp_render_data(), ) anchor = out["fp_a"]["anchor_selection_metadata"] # L01B01 is clean_background_expected=True → candidate. # L01B02 is False → not a candidate. assert anchor["candidate_bg_ids"] == ["L01B01"] # W20A never picks anchor. assert anchor["selected_anchor_bg_id"] is None assert any("W20B" in d for d in anchor["selection_diagnostics"]) def test_dossier_overlay_inventory_carries_bg_id_and_number(): out = build_dossiers( fp_prompt_data=_fp_prompt_data(), master_plan_data=_master_plan_data(), overlay_payload_data=_overlay_payload_data(), fp_render_data=_fp_render_data(), ) overlay_inv = out["fp_a"]["overlay_marker_inventory"] assert len(overlay_inv) == 1 entry = overlay_inv[0] assert entry["bg_id"] == "L01B02" assert entry["number"] == 6 assert entry["base_layer_decision"] == "state_overlay_plot_cue" def test_dossier_emits_per_bg_render_facts_for_each_bg_id(): """per_bg_render_facts_by_bg_id surfaces exact-ID per-bg facts so W20B can plan without re-reading W19 / master_plan checkpoints. Each entry must carry the full overlay-derived integer fact set (target units, dominant unit, use/ignore lists, base marker numbers to reference, transient marker numbers to describe, ignored overlay marker numbers, clean expectation) plus master_plan-derived relationships (applies_to_shots, depends_on_bg). """ out = build_dossiers( fp_prompt_data=_fp_prompt_data(), master_plan_data=_master_plan_data(), overlay_payload_data=_overlay_payload_data(), fp_render_data=_fp_render_data(), ) facts = out["fp_a"]["per_bg_render_facts_by_bg_id"] assert set(facts.keys()) == {"L01B01", "L01B02"} b1 = facts["L01B01"] assert b1["bg_id"] == "L01B01" assert b1["fp_id"] == "fp_a" assert b1["target_unit_marker_numbers"] == [1] assert b1["dominant_target_unit_marker_number"] == 1 assert b1["use_numbered_elements"] == [1, 3, 4] assert b1["ignore_numbered_elements"] == [6] assert b1["base_marker_numbers_to_reference"] == [1, 3, 4] assert b1["transient_marker_numbers_to_describe"] == [] assert b1["ignored_state_overlay_marker_numbers"] == [6] assert b1["clean_background_expected"] is True assert b1["diagnostics"] == [] b2 = facts["L01B02"] assert b2["target_unit_marker_numbers"] == [2] assert b2["dominant_target_unit_marker_number"] == 2 assert b2["use_numbered_elements"] == [2, 6] assert b2["base_marker_numbers_to_reference"] == [2] assert b2["transient_marker_numbers_to_describe"] == [6] assert b2["clean_background_expected"] is False assert b2["diagnostics"] == [] def test_dossier_per_bg_facts_carry_applies_to_shots_and_depends_on_bg(): out = build_dossiers( fp_prompt_data=_fp_prompt_data(), master_plan_data=_master_plan_data(), overlay_payload_data=_overlay_payload_data(), fp_render_data=_fp_render_data(), ) facts = out["fp_a"]["per_bg_render_facts_by_bg_id"] assert facts["L01B01"]["applies_to_shots"] == ["S1_Shot1"] assert facts["L01B01"]["depends_on_bg"] == [] assert facts["L01B02"]["applies_to_shots"] == ["S1_Shot2"] assert facts["L01B02"]["depends_on_bg"] == ["L01B01"] def test_dossier_per_bg_facts_diagnostic_when_master_plan_spec_missing(): """overlay present + master_plan spec absent → diagnostic + empty spec-derived fields. No inference about shot/bg relationships.""" out = build_dossiers( fp_prompt_data=_fp_prompt_data(), master_plan_data={"plans": {}}, # no bg specs overlay_payload_data=_overlay_payload_data(), fp_render_data=_fp_render_data(), ) facts = out["fp_a"]["per_bg_render_facts_by_bg_id"] assert "L01B01" in facts b1 = facts["L01B01"] # overlay-derived fields preserved. assert b1["target_unit_marker_numbers"] == [1] assert b1["clean_background_expected"] is True # spec-derived fields default empty. assert b1["applies_to_shots"] == [] assert b1["depends_on_bg"] == [] # diagnostic surfaces the gap. assert any( "background_master_plan" in d for d in b1["diagnostics"] ) def test_dossier_per_bg_facts_diagnostic_when_overlay_missing(): """master_plan spec present + overlay absent → diagnostic + empty overlay-derived fields. clean defaults False; no inference.""" out = build_dossiers( fp_prompt_data=_fp_prompt_data(), master_plan_data=_master_plan_data(), overlay_payload_data={"overlays": {}}, # overlay step n/a upstream fp_render_data=_fp_render_data(), ) facts = out["fp_a"]["per_bg_render_facts_by_bg_id"] assert "L01B01" in facts and "L01B02" in facts b1 = facts["L01B01"] assert b1["target_unit_marker_numbers"] == [] assert b1["dominant_target_unit_marker_number"] is None assert b1["use_numbered_elements"] == [] assert b1["ignore_numbered_elements"] == [] assert b1["base_marker_numbers_to_reference"] == [] assert b1["transient_marker_numbers_to_describe"] == [] assert b1["ignored_state_overlay_marker_numbers"] == [] assert b1["clean_background_expected"] is False # spec-derived fields still flow through. assert b1["applies_to_shots"] == ["S1_Shot1"] assert any( "floor_plan_overlay_payload" in d for d in b1["diagnostics"] ) def test_dossier_base_marker_inventory_excludes_overlay_markers(): out = build_dossiers( fp_prompt_data=_fp_prompt_data(), master_plan_data=_master_plan_data(), overlay_payload_data=_overlay_payload_data(), fp_render_data=_fp_render_data(), ) base_inv = out["fp_a"]["base_marker_inventory"] numbers = sorted(entry["number"] for entry in base_inv) # 1, 2, 3, 4, 5 are base_*; 6 is overlay → excluded. assert numbers == [1, 2, 3, 4, 5] for entry in base_inv: assert entry["base_layer_decision"] in BASE_LAYER_DECISIONS # ---------- fail-closed ---------- def test_dossier_fails_when_no_floor_plans_with_status_ok(): with pytest.raises(BaseLocationDossierError): build_dossiers( fp_prompt_data={"floor_plans": {}}, master_plan_data=_master_plan_data(), overlay_payload_data=_overlay_payload_data(), ) def test_dossier_fails_on_invalid_base_layer_decision_enum(): fp = _fp_prompt_data() fp["floor_plans"]["fp_a"]["numbered_elements"][0]["base_layer_decision"] = ( "not_in_enum" ) with pytest.raises(BaseLocationDossierError) as exc: build_dossiers( fp_prompt_data=fp, master_plan_data=_master_plan_data(), overlay_payload_data=_overlay_payload_data(), ) assert "not_in_enum" in str(exc.value) def test_dossier_fails_when_no_base_structural_unit(): """A dossier with zero unit markers cannot be built — fail-closed.""" fp = _fp_prompt_data() # Reclassify all units as fixtures so unit_markers becomes empty. for ne in fp["floor_plans"]["fp_a"]["numbered_elements"]: if ne.get("base_layer_decision") == "base_structural_unit": ne["base_layer_decision"] = "base_persistent_fixture" with pytest.raises(BaseLocationDossierError) as exc: build_dossiers( fp_prompt_data=fp, master_plan_data=_master_plan_data(), overlay_payload_data=_overlay_payload_data(), ) assert "zero base_structural_unit" in str(exc.value) def test_dossier_fails_on_bad_grid_size(): with pytest.raises(BaseLocationDossierError): build_dossiers( fp_prompt_data=_fp_prompt_data(), master_plan_data=_master_plan_data(), overlay_payload_data=_overlay_payload_data(), grid_size=(10,), # not a 2-tuple ) # ---------- missing-but-non-fatal inputs ---------- def test_dossier_records_diagnostic_when_fp_render_missing(): out = build_dossiers( fp_prompt_data=_fp_prompt_data(), master_plan_data=_master_plan_data(), overlay_payload_data=_overlay_payload_data(), fp_render_data=None, # render cp absent ) dossier = out["fp_a"] assert dossier["fp_image_path"] is None assert any( "floor_plan_render" in d for d in dossier["diagnostics"] ) def test_dossier_records_diagnostic_when_overlay_payload_missing(): out = build_dossiers( fp_prompt_data=_fp_prompt_data(), master_plan_data=_master_plan_data(), overlay_payload_data={}, # v5 default upstream → overlay step n/a fp_render_data=_fp_render_data(), ) dossier = out["fp_a"] assert dossier["overlay_marker_inventory"] == [] assert any( "floor_plan_overlay_payload" in d for d in dossier["diagnostics"] ) # Anchor candidate set is empty because no overlay supplied # clean_background_expected — code must not invent candidates. assert dossier["anchor_selection_metadata"]["candidate_bg_ids"] == [] # ---------- partition enum is the locked W19 contract ---------- def test_partition_enums_match_w19b1_locked_set(): """BASE / STATE_OVERLAY enum sets must mirror floor_plan_overlay_payload.""" from app.modules.pipeline.floor_plan_overlay_payload import ( BASE_LAYER_DECISIONS as W19_BASE, STATE_OVERLAY_DECISIONS as W19_STATE, ALL_LAYER_DECISIONS as W19_ALL, ) assert BASE_LAYER_DECISIONS == W19_BASE assert STATE_OVERLAY_DECISIONS == W19_STATE assert ALL_LAYER_DECISIONS == W19_ALL