"""W20A2: FloorPlanGeometryReadbackStep gate + happy-path tests. LLM / image / VLM API call 0. DB / ImageAsset write 0. Production-side, focused. The step writes a sidecar HTML file under a tmp_path-backed checkpoint dir; tests monkeypatch settings.projects_dir to a tmp path so production checkpoints stay untouched. """ from __future__ import annotations import ast import json from pathlib import Path from unittest.mock import MagicMock, patch from app.core.steps.floor_plan_geometry_readback_step import ( PROMPT_VERSION, SCHEMA_VERSION, FloorPlanGeometryReadbackStep, ) from app.modules.pipeline.floor_plan_geometry_readback import ( BASE_OPENING, BASE_PERSISTENT_FIXTURE, BASE_PERSISTENT_FURNITURE, BASE_STRUCTURAL_UNIT, ) _DOSSIER = { "fp_id": "fp_a", "fp_image_path": "/p/c/e/floor_plan_render/fp_a.png", "grid_size": [10, 10], "dwelling_identity": { "standard_of_living_band": "modest_residential", }, "base_marker_inventory": [ {"number": 1, "label": "primary unit", "category": "area", "position_hint": "", "base_layer_decision": BASE_STRUCTURAL_UNIT}, {"number": 2, "label": "secondary unit", "category": "area", "position_hint": "", "base_layer_decision": BASE_STRUCTURAL_UNIT}, {"number": 3, "label": "interior opening", "category": "opening", "position_hint": "", "base_layer_decision": BASE_OPENING}, {"number": 4, "label": "service counter", "category": "furniture", "position_hint": "", "base_layer_decision": BASE_PERSISTENT_FIXTURE}, {"number": 5, "label": "anchor seating", "category": "furniture", "position_hint": "", "base_layer_decision": BASE_PERSISTENT_FURNITURE}, ], "per_bg_render_facts_by_bg_id": {}, "diagnostics": [], } def _cp_map_with_dossier(): return { "base_location_dossier": {"data": {"dossiers": {"fp_a": _DOSSIER}}}, } def _new_step(*, tmp_path: Path, cp_map: dict) -> FloorPlanGeometryReadbackStep: step = FloorPlanGeometryReadbackStep.__new__( FloorPlanGeometryReadbackStep ) 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 ──────────────────────────── def test_default_off_returns_not_applicable(tmp_path): step = _new_step(tmp_path=tmp_path, cp_map=_cp_map_with_dossier()) with patch("app.core.config.settings.projects_dir", str(tmp_path)), \ 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), \ patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", False, ): result = step._execute() assert result["applicable_count"] == 0 assert result["data"] == {} def test_not_applicable_when_dossier_selector_disabled(tmp_path): step = _new_step(tmp_path=tmp_path, cp_map=_cp_map_with_dossier()) with patch("app.core.config.settings.projects_dir", str(tmp_path)), \ 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), \ patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", True, ): result = step._execute() assert result["applicable_count"] == 0 def test_not_applicable_when_background_mode_off(tmp_path): step = _new_step(tmp_path=tmp_path, cp_map=_cp_map_with_dossier()) with patch("app.core.config.settings.projects_dir", str(tmp_path)), \ 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), \ patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", True, ): result = step._execute() assert result["applicable_count"] == 0 def test_not_applicable_when_floor_plan_prompt_version_is_v5(tmp_path): step = _new_step(tmp_path=tmp_path, cp_map=_cp_map_with_dossier()) with patch("app.core.config.settings.projects_dir", str(tmp_path)), \ 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), \ patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", True, ): result = step._execute() assert result["applicable_count"] == 0 def test_not_applicable_when_dossier_checkpoint_missing(tmp_path): step = _new_step(tmp_path=tmp_path, cp_map={}) # no dossier cp at all with patch("app.core.config.settings.projects_dir", str(tmp_path)), \ 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), \ patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", True, ): result = step._execute() assert result["applicable_count"] == 0 assert result["data"] == {} # ──────────────────────────── opt-in happy path ──────────────────────────── # ──────────────────────── W20A2.5 vlm provider resolution ─────────────────────── def test_resolve_provider_defaults_to_none(tmp_path): step = _new_step(tmp_path=tmp_path, cp_map=_cp_map_with_dossier()) with patch( "app.core.config.settings.floor_plan_vlm_readback_real_provider_enabled", False, ): assert step._resolve_vlm_provider() is None def test_resolve_provider_returns_litellm_helper_when_selector_true(tmp_path): """W20A2.5: selector True → step resolves to the provider helper. The helper itself raises NotImplementedError when invoked; this test only asserts the resolution wiring, not the invocation.""" from app.modules.pipeline.floor_plan_vlm_provider import ( gemini_vlm_provider, ) step = _new_step(tmp_path=tmp_path, cp_map=_cp_map_with_dossier()) with patch( "app.core.config.settings.floor_plan_vlm_readback_real_provider_enabled", True, ): # ★#92: 단일 경로도 gemini 다 (Router 경유 어댑터). assert step._resolve_vlm_provider() is gemini_vlm_provider def test_test_only_provider_override_bypasses_selector(tmp_path): step = _new_step(tmp_path=tmp_path, cp_map=_cp_map_with_dossier()) fake = lambda **kw: {"status": "ok"} step.set_vlm_provider_for_testing(fake) with patch( "app.core.config.settings.floor_plan_vlm_readback_real_provider_enabled", False, ): assert step._resolve_vlm_provider() is fake def test_opt_in_mock_provider_invokes_once_per_fp_and_records_counter(tmp_path): """Mock provider returning a valid ``ok`` readback → counter=1 per fp. Counter aggregated to ``real_vlm_call_count`` at result level.""" step = _new_step(tmp_path=tmp_path, cp_map=_cp_map_with_dossier()) def mock_provider(*, dossier, fp_image_path, grid_size): return { "status": "ok", "fp_id": dossier["fp_id"], "grid_size": list(grid_size), "observed_markers": [ {"number": 1, "row": 1, "col": 1, "kind": "base_structural_unit"}, ], "missing_markers": [], "extra_markers": [], "confidence": 0.9, "diagnostics": [], } step.set_vlm_provider_for_testing(mock_provider) with patch("app.core.config.settings.projects_dir", str(tmp_path)), \ 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), \ patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", True, ): result = step._execute() assert result["applicable_count"] == 1 assert result["data"]["real_vlm_call_count"] == 1 entry = result["data"]["per_fp"]["fp_a"] assert entry["real_vlm_call"] is True assert entry["real_vlm_call_count"] == 1 assert entry["readback_status"] == "ok" def test_opt_in_selector_true_helper_not_implemented_marks_fp_failed(tmp_path): """Flipping the real-provider selector before its follow-up wave must fail closed at the call site (provider error → fp entry 'failed', counter reflects the attempted call). ``max_retries=0`` isolates this single-attempt fail-closed intent from the TASK3-A retry behaviour (covered by its own tests).""" step = _new_step(tmp_path=tmp_path, cp_map=_cp_map_with_dossier()) with patch("app.core.config.settings.projects_dir", str(tmp_path)), \ 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), \ patch( "app.core.config.settings.floor_plan_geometry_readback_max_retries", 0, ), patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", True, ), patch( "app.core.config.settings.floor_plan_vlm_readback_real_provider_enabled", True, ): result = step._execute() assert result["failed_count"] == 1 entry = result["data"]["per_fp"]["fp_a"] assert "error" in entry assert entry["real_vlm_call_count"] == 1 assert result["data"]["real_vlm_call_count"] == 1 def test_default_off_real_provider_selector_keeps_synthetic_path(tmp_path): """W20A2.5 default OFF — selector False → resolved provider None → synthetic fixture readback emitted, counter 0.""" step = _new_step(tmp_path=tmp_path, cp_map=_cp_map_with_dossier()) # No injection, selector default False. with patch("app.core.config.settings.projects_dir", str(tmp_path)), \ 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), \ patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", True, ), patch( "app.core.config.settings.floor_plan_vlm_readback_real_provider_enabled", False, ): result = step._execute() assert result["data"]["real_vlm_call_count"] == 0 entry = result["data"]["per_fp"]["fp_a"] assert entry["readback_status"] == "synthetic_fixture" assert entry["real_vlm_call"] is False assert entry["real_vlm_call_count"] == 0 # ──────────────────────── original happy path (kept) ─────────────────────── def test_opt_in_completes_writes_html_and_records_zero_vlm_calls(tmp_path): cp_map = _cp_map_with_dossier() step = _new_step(tmp_path=tmp_path, cp_map=cp_map) with patch("app.core.config.settings.projects_dir", str(tmp_path)), \ 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), \ patch( "app.core.config.settings." "floor_plan_vlm_readback_real_provider_enabled", False, ), \ patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", True, ): result = step._execute() assert result["applicable_count"] == 1 assert result["completed_count"] == 1 assert result["failed_count"] == 0 per_fp = result["data"]["per_fp"] assert "fp_a" in per_fp entry = per_fp["fp_a"] assert entry["readback_status"] == "synthetic_fixture" assert entry["real_vlm_call"] is False rel = entry["review_html_relative_path"] assert rel.endswith("/fp_a.html") assert "floor_plan_geometry_readback/review_html" in rel # HTML file actually written. html_path = ( Path(tmp_path) / "p" / "checkpoints" / "episodes" / "e" / rel ) assert html_path.exists() content = html_path.read_text(encoding="utf-8") assert content.startswith("") assert "fp_id=fp_a" in content # Step-level counters. assert result["data"]["real_vlm_call_count"] == 0 assert result["data"]["image_api_call_count"] == 0 assert result["data"]["llm_call_count"] == 0 def test_opt_in_failed_dossier_produces_failed_entry_not_crash(tmp_path): """Empty base inventory → fixture error per fp_id, but step itself returns failed_count=1 + structured error entry, no exception out.""" bad_dossier = dict(_DOSSIER) bad_dossier["base_marker_inventory"] = [] cp_map = { "base_location_dossier": { "data": {"dossiers": {"fp_a": bad_dossier}} } } step = _new_step(tmp_path=tmp_path, cp_map=cp_map) with patch("app.core.config.settings.projects_dir", str(tmp_path)), \ 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), \ patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", True, ): result = step._execute() assert result["applicable_count"] == 1 assert result["completed_count"] == 0 assert result["failed_count"] == 1 assert "error" in result["data"]["per_fp"]["fp_a"] # ──────────────────────────── config_hash ──────────────────────────── def test_config_hash_changes_when_geometry_selector_flips(tmp_path): cp_map = _cp_map_with_dossier() step = _new_step(tmp_path=tmp_path, cp_map=cp_map) with patch("app.core.config.settings.projects_dir", str(tmp_path)), \ 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), \ patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", False, ): h_off = step._config_hash() with patch("app.core.config.settings.projects_dir", str(tmp_path)), \ 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), \ patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", True, ): h_on = step._config_hash() assert h_off != h_on def test_schema_and_prompt_version_constants(): # W20F7-A bumped production SCHEMA_VERSION 1→2 (VLM max_completion_tokens # 2000→16000). 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 "floor_plan_geometry_readback" in STEP_CLASSES assert ( STEP_CLASSES["floor_plan_geometry_readback"] is FloorPlanGeometryReadbackStep ) def test_step_registered_in_manifest_between_dossier_and_bg_prompt(): from app.core.step_manifest import STEP_MANIFEST entry = STEP_MANIFEST["floor_plan_geometry_readback"] assert entry["applicability"] == "if_background_mode" assert entry["step_type"] == "transform" dossier_order = STEP_MANIFEST["base_location_dossier"]["order"] bg_prompt_order = STEP_MANIFEST["background_prompt"]["order"] assert dossier_order < entry["order"] < bg_prompt_order assert "base_location_dossier" in entry["depends_on"] # ──────────── 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_geometry_readback.py" ) tokens = _import_tokens(mod) leaks = tokens & _BANNED_IMPORT_TOKENS assert not leaks, ( f"floor_plan_geometry_readback must not import LLM/image/VLM APIs; " f"found: {sorted(leaks)}" ) def test_html_emitter_has_no_llm_image_vlm_imports(): mod = ( Path(__file__).resolve().parent.parent.parent / "app" / "modules" / "pipeline" / "floor_plan_review_html.py" ) tokens = _import_tokens(mod) leaks = tokens & _BANNED_IMPORT_TOKENS assert not leaks def test_step_has_no_llm_image_vlm_imports(): step = ( Path(__file__).resolve().parent.parent.parent / "app" / "core" / "steps" / "floor_plan_geometry_readback_step.py" ) tokens = _import_tokens(step) leaks = tokens & _BANNED_IMPORT_TOKENS assert not leaks # ─────────────────── TASK3-A: per-fp bounded retry ─────────────────── def _ok_readback(dossier, grid_size): return { "status": "ok", "fp_id": dossier["fp_id"], "grid_size": list(grid_size), "observed_markers": [ {"number": 1, "row": 1, "col": 1, "kind": "base_structural_unit"}, ], "missing_markers": [], "extra_markers": [], "confidence": 0.9, "diagnostics": [], } def _readback_settings_patch(tmp_path, max_retries=2): """Common settings context for the retry tests (real-provider lane).""" return [ patch("app.core.config.settings.projects_dir", str(tmp_path)), 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), patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", True, ), patch( "app.core.config.settings.floor_plan_geometry_readback_max_retries", max_retries, ), ] def _run_with(patches, step): import contextlib with contextlib.ExitStack() as stack: for p in patches: stack.enter_context(p) return step._execute() def test_retry_recovers_transient_readback_failure(tmp_path): """A provider that raises on the first attempt but succeeds on the second yields an ``ok`` fp once retry is enabled. The retry catches a NON-GeometryReadbackError (mirrors the real VlmProviderError path), so a broad transient catch is required. Each attempt is a fresh counted call → ``real_vlm_call_count`` reflects all attempts honestly.""" step = _new_step(tmp_path=tmp_path, cp_map=_cp_map_with_dossier()) calls = {"n": 0} def flaky_provider(*, dossier, fp_image_path, grid_size): calls["n"] += 1 if calls["n"] == 1: raise RuntimeError("validator_failed: marker #3 duplicated") return _ok_readback(dossier, grid_size) step.set_vlm_provider_for_testing(flaky_provider) result = _run_with(_readback_settings_patch(tmp_path, 2), step) entry = result["data"]["per_fp"]["fp_a"] assert entry["readback_status"] == "ok" assert entry["readback_attempts"] == 2 assert entry["real_vlm_call_count"] == 2 assert result["data"]["real_vlm_call_count"] == 2 assert result["completed_count"] == 1 assert result["failed_count"] == 0 def test_retry_exhausted_keeps_failed_shape(tmp_path): """When every attempt fails the fp stays failed with the original error semantics + attempt count recorded. With max_retries=2 the provider is invoked 3 times.""" step = _new_step(tmp_path=tmp_path, cp_map=_cp_map_with_dossier()) calls = {"n": 0} def always_fail(*, dossier, fp_image_path, grid_size): calls["n"] += 1 raise RuntimeError("validator_failed: persistent") step.set_vlm_provider_for_testing(always_fail) result = _run_with(_readback_settings_patch(tmp_path, 2), step) assert calls["n"] == 3 entry = result["data"]["per_fp"]["fp_a"] assert "error" in entry assert entry["readback_attempts"] == 3 assert entry["real_vlm_call_count"] == 3 assert result["failed_count"] == 1 assert result["completed_count"] == 0 def test_retry_zero_is_single_attempt(tmp_path): """max_retries=0 = legacy single-attempt: a failing provider is called exactly once and the fp fails (no retry).""" step = _new_step(tmp_path=tmp_path, cp_map=_cp_map_with_dossier()) calls = {"n": 0} def always_fail(*, dossier, fp_image_path, grid_size): calls["n"] += 1 raise RuntimeError("boom") step.set_vlm_provider_for_testing(always_fail) result = _run_with(_readback_settings_patch(tmp_path, 0), step) assert calls["n"] == 1 assert result["failed_count"] == 1 def test_synthetic_path_skips_retry_field_byte_identical(tmp_path): """Provider None (synthetic / default-OFF) is deterministic: no retry is attempted even with a high max_retries, and the per_fp entry carries NO ``readback_attempts`` key (payload byte-identical to legacy).""" step = _new_step(tmp_path=tmp_path, cp_map=_cp_map_with_dossier()) # No provider injected + real-provider selector OFF → resolved provider # None → synthetic fixture (deterministic, retry skipped). patches = _readback_settings_patch(tmp_path, 5) + [ patch( "app.core.config.settings." "floor_plan_vlm_readback_real_provider_enabled", False, ), ] result = _run_with(patches, step) entry = result["data"]["per_fp"]["fp_a"] assert entry["readback_status"] == "synthetic_fixture" assert "readback_attempts" not in entry assert result["data"]["real_vlm_call_count"] == 0