"""W20A: BaseLocationDossierStep gate + cp-driven dispatch tests. production-side, focused. LLM / image / VLM API call 0, DB / ImageAsset write 0. Default-OFF contract: - ``settings.base_location_dossier_enabled = False`` (default) → step returns ``not_applicable`` regardless of other settings. - Opt-in requires both ``base_location_dossier_enabled = True`` AND ``floor_plan_prompt_version = "6"``. - Background mode must be ``on`` or ``floor_plan_anchored``. When opt-in, the step delegates to ``build_dossiers`` and serializes the result under ``data.dossiers``. No retry, no LLM/VLM call. """ from __future__ import annotations import ast from pathlib import Path from unittest.mock import MagicMock, patch from app.core.steps.base_location_dossier_step import ( PROMPT_VERSION, SCHEMA_VERSION, BaseLocationDossierStep, ) _FP_PROMPT_DATA = { "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": "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", "camera_height": "eye level", "lens_hint": "35mm", "framing_notes": "", "use_numbered_elements": [1], "ignore_numbered_elements": [2], } ], } } } _MASTER_PLAN_DATA = { "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": [], } ], "gen_order": ["L01B01"], }, } } } _OVERLAY_DATA = { "overlays": { "L01B01": { "bg_id": "L01B01", "fp_id": "fp_a", "use_numbered_elements": [1], "ignore_numbered_elements": [2], "base_markers_to_reference": [ { "number": 1, "label": "primary unit", "category": "area", "position_hint": "center", "base_layer_decision": "base_structural_unit", } ], "transient_markers_to_describe": [], "ignored_state_overlay_markers": [ { "number": 2, "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": [], } } } _FP_RENDER_DATA = { "floor_plans": {"fp_a": {"png_path": "/p/c/e/floor_plan_render/fp_a.png"}} } _CP_MAP = { "floor_plan_prompt": {"data": _FP_PROMPT_DATA}, "background_master_plan": {"data": _MASTER_PLAN_DATA}, "floor_plan_overlay_payload": {"data": _OVERLAY_DATA}, "floor_plan_render": {"data": _FP_RENDER_DATA}, } def _new_step() -> BaseLocationDossierStep: step = BaseLocationDossierStep.__new__(BaseLocationDossierStep) 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) ) return step # ---------- gates: default-off ---------- def test_default_off_returns_not_applicable_even_when_v6_and_bg_on(): """Highest-priority contract: dossier is opt-in. With the default ``base_location_dossier_enabled=False`` no step output is produced even when every other gate would have passed. """ with patch("app.core.config.settings.background_mode", "on"), \ patch("app.core.config.settings.floor_plan_prompt_version", "6"), \ patch("app.core.config.settings.base_location_dossier_enabled", False): result = _new_step()._execute() assert result["applicable_count"] == 0 assert result["completed_count"] == 0 assert result["failed_count"] == 0 assert result["data"] == {} def test_not_applicable_when_background_mode_off_even_when_enabled(): with patch("app.core.config.settings.background_mode", "off"), \ patch("app.core.config.settings.floor_plan_prompt_version", "6"), \ patch("app.core.config.settings.base_location_dossier_enabled", True): result = _new_step()._execute() assert result["applicable_count"] == 0 assert result["data"] == {} def test_not_applicable_when_v5_default_even_when_enabled(): """v5 path lacks base_layer_decision → dossier partition fails. The step must skip cleanly, not try to build and fail-closed. """ with patch("app.core.config.settings.background_mode", "on"), \ patch("app.core.config.settings.floor_plan_prompt_version", "5"), \ patch("app.core.config.settings.base_location_dossier_enabled", True): result = _new_step()._execute() assert result["applicable_count"] == 0 assert result["data"] == {} # ---------- opt-in happy path ---------- def test_opt_in_completes_with_dossier_under_data(): with patch("app.core.config.settings.background_mode", "on"), \ patch("app.core.config.settings.floor_plan_prompt_version", "6"), \ patch("app.core.config.settings.base_location_dossier_enabled", True): result = _new_step()._execute() assert result["applicable_count"] == 1 assert result["completed_count"] == 1 assert result["failed_count"] == 0 dossiers = result["data"]["dossiers"] assert "fp_a" in dossiers dossier = dossiers["fp_a"] assert dossier["fp_id"] == "fp_a" assert dossier["fp_image_path"] == "/p/c/e/floor_plan_render/fp_a.png" # Anchor never picked by code. assert dossier["anchor_selection_metadata"]["selected_anchor_bg_id"] is None # VLM gate is synthetic only. assert ( dossier["fp_geometry_vlm_readback"]["status"] == "synthetic_placeholder" ) def test_opt_in_with_floor_plan_anchored_alias_also_active(): with patch("app.core.config.settings.background_mode", "floor_plan_anchored"), \ patch("app.core.config.settings.floor_plan_prompt_version", "6"), \ patch("app.core.config.settings.base_location_dossier_enabled", True): result = _new_step()._execute() assert result["applicable_count"] == 1 assert result["completed_count"] == 1 # ---------- fail-closed ---------- def test_opt_in_fails_when_marker_partition_invalid(): bad_fp = { "floor_plans": { "fp_a": { "status": "ok", "fp_id": "fp_a", "t2i_prompt": "x" * 50, "numbered_elements": [ { "number": 1, "label": "u", "category": "area", "position_hint": "", "base_layer_decision": "not_in_enum", } ], "camera_recommendations": [], } } } cp_map = dict(_CP_MAP) cp_map["floor_plan_prompt"] = {"data": bad_fp} step = BaseLocationDossierStep.__new__(BaseLocationDossierStep) 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)) with patch("app.core.config.settings.background_mode", "on"), \ patch("app.core.config.settings.floor_plan_prompt_version", "6"), \ patch("app.core.config.settings.base_location_dossier_enabled", True): result = step._execute() assert result["applicable_count"] == 1 assert result["completed_count"] == 0 assert result["failed_count"] == 1 assert "not_in_enum" in result["data"]["error"] assert result["data"]["dossiers"] == {} # ---------- config_hash ---------- def test_config_hash_changes_when_selector_flips(): with patch("app.core.config.settings.background_mode", "on"), \ patch("app.core.config.settings.floor_plan_prompt_version", "6"), \ patch("app.core.config.settings.base_location_dossier_enabled", False): h_off = _new_step()._config_hash() with patch("app.core.config.settings.background_mode", "on"), \ patch("app.core.config.settings.floor_plan_prompt_version", "6"), \ patch("app.core.config.settings.base_location_dossier_enabled", True): h_on = _new_step()._config_hash() assert h_off != h_on def test_schema_and_prompt_version_constants(): # W20F5 bumped production SCHEMA_VERSION 1→2 (anchor fallback). Lock # synced to the production value (W21B-wave-4 deterministic hygiene). assert SCHEMA_VERSION == 2 assert PROMPT_VERSION == "1" # ---------- registry ---------- def test_step_registered_in_step_classes(): from app.core.steps import STEP_CLASSES assert "base_location_dossier" in STEP_CLASSES assert STEP_CLASSES["base_location_dossier"] is BaseLocationDossierStep def test_step_registered_in_manifest_with_order_between_overlay_and_prompt(): from app.core.step_manifest import STEP_MANIFEST assert "base_location_dossier" in STEP_MANIFEST entry = STEP_MANIFEST["base_location_dossier"] assert entry["applicability"] == "if_background_mode" assert entry["step_type"] == "transform" overlay_order = STEP_MANIFEST["floor_plan_overlay_payload"]["order"] prompt_order = STEP_MANIFEST["background_prompt"]["order"] assert overlay_order < entry["order"] < prompt_order # ---------- safety: no LLM/image/VLM imports in module/step source ---------- _BANNED_IMPORT_TOKENS = { "litellm", "openai", "anthropic", "google.generativeai", "fal", "call_structured", "call_multiturn", "images.edit", "ImageAsset", } def _import_tokens(path: Path) -> set[str]: tokens: set[str] = set() tree = ast.parse(path.read_text(encoding="utf-8")) for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: tokens.add(alias.name) elif isinstance(node, ast.ImportFrom): if node.module: tokens.add(node.module) for alias in node.names: tokens.add(alias.name) return tokens def test_module_has_no_llm_image_vlm_imports(): mod = ( Path(__file__).resolve().parent.parent.parent / "app" / "modules" / "pipeline" / "base_location_dossier.py" ) tokens = _import_tokens(mod) leaks = tokens & _BANNED_IMPORT_TOKENS assert not leaks, ( f"base_location_dossier module must not import LLM/image/VLM APIs; " f"found: {sorted(leaks)}" ) def test_step_has_no_llm_image_vlm_imports(): step = ( Path(__file__).resolve().parent.parent.parent / "app" / "core" / "steps" / "base_location_dossier_step.py" ) tokens = _import_tokens(step) leaks = tokens & _BANNED_IMPORT_TOKENS assert not leaks, ( f"base_location_dossier_step must not import LLM/image/VLM APIs; " f"found: {sorted(leaks)}" )