"""W19B-1: FloorPlanOverlayPayloadStep gate + cp-driven dispatch tests. production-side, minimal. LLM / image / VLM API call 0, DB / ImageAsset write 0. """ from __future__ import annotations import ast from pathlib import Path from unittest.mock import MagicMock, patch from app.core.steps.floor_plan_overlay_payload_step import ( PROMPT_VERSION, SCHEMA_VERSION, FloorPlanOverlayPayloadStep, ) _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, facing east", "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"], }, } } } _CP_MAP = { "floor_plan_prompt": {"data": _FP_PROMPT_DATA}, "background_master_plan": {"data": _MASTER_PLAN_DATA}, } def _new_step() -> FloorPlanOverlayPayloadStep: step = FloorPlanOverlayPayloadStep.__new__(FloorPlanOverlayPayloadStep) step.project_id = "p" step.episode_id = "e" step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) # not used, but keep API parity step._load_prev_checkpoint = MagicMock(side_effect=lambda sid: _CP_MAP.get(sid)) return step # ----- gates ----- def test_step_not_applicable_when_background_mode_off(): with patch("app.core.config.settings.background_mode", "off"), \ patch("app.core.config.settings.floor_plan_prompt_version", "6"): result = _new_step()._execute() assert result["applicable_count"] == 0 assert result["completed_count"] == 0 assert result["failed_count"] == 0 assert result["data"] == {} def test_step_not_applicable_when_chain_only_mode(): with patch("app.core.config.settings.background_mode", "chain_only"), \ patch("app.core.config.settings.floor_plan_prompt_version", "6"): result = _new_step()._execute() assert result["applicable_count"] == 0 assert result["data"] == {} def test_step_not_applicable_when_floor_plan_prompt_version_is_v5_default(): """default v5 path: overlay payload step skips → v5 caller 회귀 0.""" with patch("app.core.config.settings.background_mode", "on"), \ patch("app.core.config.settings.floor_plan_prompt_version", "5"): result = _new_step()._execute() assert result["applicable_count"] == 0 assert result["data"] == {} def test_step_applicable_when_mode_on_and_v6_selector(): with patch("app.core.config.settings.background_mode", "on"), \ patch("app.core.config.settings.floor_plan_prompt_version", "6"): result = _new_step()._execute() assert result["applicable_count"] == 1 assert result["completed_count"] == 1 assert result["failed_count"] == 0 assert "L01B01" in result["data"]["overlays"] overlay = result["data"]["overlays"]["L01B01"] assert overlay["fp_id"] == "fp_a" assert overlay["use_numbered_elements"] == [1] assert overlay["ignore_numbered_elements"] == [2] assert overlay["clean_background_expected"] is True assert overlay["dominant_target_unit_marker_number"] == 1 def test_step_applicable_with_floor_plan_anchored_alias(): with patch("app.core.config.settings.background_mode", "floor_plan_anchored"), \ patch("app.core.config.settings.floor_plan_prompt_version", "6"): result = _new_step()._execute() assert result["applicable_count"] == 1 assert result["completed_count"] == 1 # ----- fail-closed at step level ----- def test_step_fails_when_cp_payload_inconsistent(): """payload module 이 raise → step 이 failed=1 + data.error set, 예외 propagate X.""" bad_master_plan = { "plans": { "g_main": { "status": "ok", "plan": { "group_id": "g_main", "floor_plans": [], "backgrounds": [ { "bg_id": "L01B01", "loc_id": "L01", "sub_location": "primary_unit", "applies_to_shots": ["S1_Shot1"], "depends_on_fp": ["fp_unknown"], "depends_on_bg": [], } ], "gen_order": ["L01B01"], }, } } } cp_map = { "floor_plan_prompt": {"data": _FP_PROMPT_DATA}, "background_master_plan": {"data": bad_master_plan}, } step = FloorPlanOverlayPayloadStep.__new__(FloorPlanOverlayPayloadStep) 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"): result = step._execute() assert result["applicable_count"] == 1 assert result["completed_count"] == 0 assert result["failed_count"] == 1 assert "fp_unknown" in result["data"]["error"] assert result["data"]["overlays"] == {} # ----- config_hash ----- def test_config_hash_changes_with_selector(): with patch("app.core.config.settings.background_mode", "on"), \ patch("app.core.config.settings.floor_plan_prompt_version", "5"): h_v5 = _new_step()._config_hash() with patch("app.core.config.settings.background_mode", "on"), \ patch("app.core.config.settings.floor_plan_prompt_version", "6"): h_v6 = _new_step()._config_hash() assert h_v5 != h_v6 def test_schema_and_prompt_version_constants(): assert SCHEMA_VERSION == 1 assert PROMPT_VERSION == "1" # ----- registry ----- def test_step_registered_in_step_classes(): from app.core.steps import STEP_CLASSES assert "floor_plan_overlay_payload" in STEP_CLASSES assert STEP_CLASSES["floor_plan_overlay_payload"] is FloorPlanOverlayPayloadStep # ----- 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" / "floor_plan_overlay_payload.py" ) tokens = _import_tokens(mod) leaks = tokens & _BANNED_IMPORT_TOKENS assert not leaks, ( f"floor_plan_overlay_payload 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" / "floor_plan_overlay_payload_step.py" ) tokens = _import_tokens(step) leaks = tokens & _BANNED_IMPORT_TOKENS assert not leaks, ( f"floor_plan_overlay_payload_step must not import LLM/image/VLM APIs; " f"found: {sorted(leaks)}" )