"""W20B: ShotAwareBgRenderPlanStep gate + dispatch tests. LLM / image / VLM API call 0 (default). DB / ImageAsset write 0. Production paths use ``llm_provider=None`` and emit empty plans with diagnostics. Mock provider is injected via ``set_llm_provider_for_testing`` on a per-instance basis — never touches global state. """ from __future__ import annotations import ast from pathlib import Path from typing import Dict from unittest.mock import MagicMock, patch from app.core.steps.shot_aware_bg_render_plan_step import ( PROMPT_VERSION, SCHEMA_VERSION, ShotAwareBgRenderPlanStep, ) _DOSSIER_CP = { "data": { "dossiers": { "fp_a": { "fp_id": "fp_a", "fp_image_path": "/p/c/e/floor_plan_render/fp_a.png", "grid_size": [10, 10], "base_marker_inventory": [ {"number": 1, "label": "u1", "category": "area", "position_hint": "", "base_layer_decision": "base_structural_unit"}, {"number": 2, "label": "u2", "category": "area", "position_hint": "", "base_layer_decision": "base_structural_unit"}, ], "per_bg_render_facts_by_bg_id": { "L01B01": { "bg_id": "L01B01", "fp_id": "fp_a", "target_unit_marker_numbers": [1], "dominant_target_unit_marker_number": 1, "use_numbered_elements": [1], "ignore_numbered_elements": [], "base_marker_numbers_to_reference": [1], "transient_marker_numbers_to_describe": [], "ignored_state_overlay_marker_numbers": [], "clean_background_expected": True, "applies_to_shots": ["S1_Shot1"], "depends_on_bg": [], "diagnostics": [], }, }, # W20E7-C: clean anchor candidate surface. "anchor_selection_metadata": { "candidate_bg_ids": ["L01B01"], "selection_diagnostics": [], "selected_anchor_bg_id": None, }, } } } } _GEOMETRY_CP = { "data": { "per_fp": { "fp_a": { "fp_id": "fp_a", "readback_status": "synthetic_fixture", "geometry": { "fp_id": "fp_a", "grid_size": [10, 10], "readback_status": "synthetic_fixture", "camera_cell_candidates_per_unit": {"1": [[0, 0]]}, "look_at_cell_candidates_per_unit": {"1": [[2, 0]]}, }, } } } } _OVERLAY_CP = {"data": {"overlays": {}}} _MASTER_PLAN_CP = {"data": {"plans": {}}} _SHOT_STAGING_CP = { "data": {"shots": [{"shot_id": "S1_Shot1", "character_angles": []}]} } _CP_MAP = { "base_location_dossier": _DOSSIER_CP, "floor_plan_geometry_readback": _GEOMETRY_CP, "floor_plan_overlay_payload": _OVERLAY_CP, "background_master_plan": _MASTER_PLAN_CP, "shot_staging": _SHOT_STAGING_CP, } def _new_step() -> ShotAwareBgRenderPlanStep: step = ShotAwareBgRenderPlanStep.__new__(ShotAwareBgRenderPlanStep) 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) ) step._llm_provider_override = None return step def _enable_all(**overrides): """Context-manager helper to flip every gate True at once.""" base = dict( background_mode="on", floor_plan_prompt_version="6", base_location_dossier_enabled=True, floor_plan_geometry_readback_enabled=True, shot_aware_bg_render_plan_enabled=True, # W21B-w3 Commit 1: pin the real-provider opt-in so local operator # `.env` (SHOT_AWARE_BG_RENDER_PLAN_REAL_PROVIDER_ENABLED=true) does # not leak into "no provider" tests. Tests that exercise the real # path override this to True explicitly. shot_aware_bg_render_plan_real_provider_enabled=False, ) base.update(overrides) return [ patch(f"app.core.config.settings.{k}", v) for k, v in base.items() ] def _patched(*managers): """Combine multiple `patch(...)` instances into one with-block.""" for m in managers: m.start() try: yield finally: for m in managers: m.stop() # ──────────────────────────── gates ──────────────────────────── def test_default_off_returns_not_applicable(): step = _new_step() managers = _enable_all(shot_aware_bg_render_plan_enabled=False) try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() assert result["applicable_count"] == 0 assert result["data"] == {} def test_not_applicable_when_dossier_disabled(): step = _new_step() managers = _enable_all(base_location_dossier_enabled=False) try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() assert result["applicable_count"] == 0 def test_not_applicable_when_geometry_disabled(): step = _new_step() managers = _enable_all(floor_plan_geometry_readback_enabled=False) try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() assert result["applicable_count"] == 0 def test_not_applicable_when_v5_default(): step = _new_step() managers = _enable_all(floor_plan_prompt_version="5") try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() assert result["applicable_count"] == 0 def test_not_applicable_when_bg_mode_off(): step = _new_step() managers = _enable_all(background_mode="off") try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() assert result["applicable_count"] == 0 # ──────────────────── opt-in (production default — no provider) ──────────────────── def test_opt_in_with_no_llm_provider_runs_zero_llm_calls(): step = _new_step() assert step._llm_provider_override is None # production default managers = _enable_all() try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() assert result["applicable_count"] == 1 # No fp completes (no provider) → completed=0. assert result["completed_count"] == 0 assert result["data"]["real_api_call_counts"] == { "image": 0, "llm": 0, "vlm": 0 } plan = result["data"]["per_fp"]["fp_a"] assert plan["graph"]["nodes"] == [] assert plan["real_api_call_counts"]["llm"] == 0 assert plan["production_clear"] is False # ──────────────────── opt-in with mock LLM provider ──────────────────── _RENDER_GUIDANCE_FIELDS = ( "visible_space_directive", "camera_framing_directive", "subject_position_directive", "state_cue_directive", "negative_continuity_directive", ) def _render_guidance_mock() -> Dict[str, str]: return {f: f"directive for {f}" for f in _RENDER_GUIDANCE_FIELDS} def _valid_mock_output(**_kw): return { "graph": { "nodes": [ { "bg_id": "L01B01", "node_index": 0, "mode": "fp_seeded_anchor", "is_dwelling_identity_anchor": True, "rationale": "anchor", "reference_decision": { "selected_refs": [], "rejected_refs": [], "same_physical_space_dedup_decision": "single_ref", "why_single_ref_or_two_refs": "anchor", "physical_space_id_per_ref": [], "same_physical_space_low_delta_candidate": False, "low_delta_reuse_target_bg_id": "", "low_delta_rationale": "", }, "camera_decision": { "camera_unit": 1, "camera_cell": [0, 0], "look_at_unit": 1, "look_at_cell": [2, 0], "lens_enum": "normal", "fov_deg": 50, "framing_notes": "", }, "render_guidance": _render_guidance_mock(), } ] } } def test_opt_in_with_mock_provider_increments_llm_count_once_per_fp(): step = _new_step() step.set_llm_provider_for_testing(_valid_mock_output) managers = _enable_all() try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() assert result["data"]["real_api_call_counts"]["llm"] == 1 plan = result["data"]["per_fp"]["fp_a"] # Synthetic readback -> production_clear False even on validator pass. assert plan["production_clear"] is False assert plan["validators"]["all_validators_passed"] is True def test_step_passes_projection_card_index_into_plan(): """W21B-w4 C3: the step loads the shot_projection_card checkpoint, builds a per-(bg,shot) card index, and the plan nodes carry the resolved projection-card state/provenance.""" step = _new_step() card_cp = { "data": { "cards": { "L01B01::S1_Shot1": { "bg_id": "L01B01", "shot_id": "S1_Shot1", "fp_id": "fp_a", "card_state": "pass", "validator_state": "ok", "card": {"card_id": "card_L01B01"}, } } } } local_map = dict(_CP_MAP) local_map["shot_projection_card"] = card_cp step._load_prev_checkpoint = MagicMock( side_effect=lambda sid: local_map.get(sid) ) step.set_llm_provider_for_testing(_valid_mock_output) managers = _enable_all() try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() node = result["data"]["per_fp"]["fp_a"]["graph"]["nodes"][0] assert node["bg_id"] == "L01B01" assert node["projection_card_state"] == "pass" assert node["projection_card_id"] == "card_L01B01" assert node["anchor_shot_id"] == "S1_Shot1" def test_step_default_no_card_cp_stamps_not_available(): """No shot_projection_card checkpoint (default / card step OFF) → plan nodes carry not_available projection fields; plan otherwise unchanged and zero extra API calls.""" step = _new_step() # _CP_MAP carries no shot_projection_card entry step.set_llm_provider_for_testing(_valid_mock_output) managers = _enable_all() try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() node = result["data"]["per_fp"]["fp_a"]["graph"]["nodes"][0] assert node["projection_card_state"] == "not_available" assert node["projection_card_source_bg_id"] == "L01B01" assert result["data"]["real_api_call_counts"] == { "image": 0, "llm": 1, "vlm": 0 } def test_build_projection_card_content_index_carries_envelope(): """W21B-w4 3a — the content index keeps the full card envelope (so the router can read base bands), unlike the metadata-only index. Absent cp returns None (router no-op).""" step = _new_step() assert step._build_projection_card_content_index(None) is None card_cp = { "data": { "cards": { "L01B01::S1_Shot1": { "bg_id": "L01B01", "shot_id": "S1_Shot1", "card_state": "pass", "card": { "card_id": "card_L01B01", "vlm_output": { "visible_items": [ {"marker_number": 1, "marker_layer": "base", "horizontal_band": "left", "depth_band": "midground", "visibility": "visible", "confidence": 0.8}, ] }, }, } } } } idx = step._build_projection_card_content_index(card_cp) entry = idx["L01B01"]["S1_Shot1"] assert entry["card_state"] == "pass" assert entry["card_id"] == "card_L01B01" # envelope present + re-joinable (entry card_id == envelope card_id) so # _self_card_base_bands can extract base bands. assert entry["card"]["card_id"] == "card_L01B01" from app.modules.pipeline.shot_aware_bg_render_plan import extract_base_bands bands = extract_base_bands(entry) assert bands == {1: ("left", "midground", "visible", 0.8)} def test_step_stamps_partition_and_dag_fields_on_nodes_card_off(): """W21B-w4 3a/3b — even with the card subsystem OFF (default), every plan node the step writes carries the PLATE_PARTITION_FIELDS / REFERENCE_DAG fields (SCHEMA-9 persisted shape). No extra API calls.""" from app.modules.pipeline.shot_aware_bg_render_plan import ( PLATE_PARTITION_FIELDS, ) step = _new_step() step.set_llm_provider_for_testing(_valid_mock_output) managers = _enable_all() try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() node = result["data"]["per_fp"]["fp_a"]["graph"]["nodes"][0] for f in PLATE_PARTITION_FIELDS: assert f in node, f"missing partition field {f}" assert "needs_new_plate" in node assert result["data"]["real_api_call_counts"] == { "image": 0, "llm": 1, "vlm": 0 } def test_w20e6b_nano_not_applicable_fp_does_not_increment_failed_count(): """W20E6-B nano: when no BG of an fp has any staged consuming shot, the per-fp plan must be marked ``not_applicable`` and the step aggregate must NOT increment failed_count or completed_count for that fp. Provider call count stays 0 because the planner is structurally skipped. """ step = _new_step() # Empty shot_staging cp -> no shot is staged for L01B01's only # applies_to_shot. fp_a has zero renderable BGs. cp_map_empty_staging = dict(_CP_MAP) cp_map_empty_staging["shot_staging"] = {"data": {"shots": []}} step._load_prev_checkpoint = MagicMock( side_effect=lambda sid: cp_map_empty_staging.get(sid) ) # Provider would explode if called -- the nano contract says it # must never be reached when zero BGs are renderable. sentinel = MagicMock(side_effect=AssertionError( "provider was called even though every BG is not_applicable" )) step.set_llm_provider_for_testing(sentinel) managers = _enable_all() try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() sentinel.assert_not_called() plan = result["data"]["per_fp"]["fp_a"] assert plan["shot_aware_bg_render_plan_status"] == "not_applicable" assert plan["graph"]["nodes"] == [] assert plan["real_api_call_counts"]["llm"] == 0 assert result["completed_count"] == 0 # Critical assertion: not_applicable per-fp MUST NOT count as failed. assert result["failed_count"] == 0 assert result["data"]["real_api_call_counts"]["llm"] == 0 assert result["data"]["llm_provider_attempts_per_fp"] == {} # ──────────────────────────── config_hash ──────────────────────────── def test_config_hash_changes_when_selector_flips(): step = _new_step() managers_off = _enable_all(shot_aware_bg_render_plan_enabled=False) try: for m in managers_off: m.start() h_off = step._config_hash() finally: for m in managers_off: m.stop() managers_on = _enable_all() try: for m in managers_on: m.start() h_on = step._config_hash() finally: for m in managers_on: m.stop() assert h_off != h_on def test_schema_and_prompt_version_constants(): # W21B-w4 3a/3b (2026-05-31): plan vNext adds card-aware routing # (route_render_actions_v2 D5 withhold) + two additive deterministic # mirrors of the canonical render_action SOT — PLATE_PARTITION_FIELDS # (mirror_plate_partition) and REFERENCE_DAG_FIELDS (build_reference_dag), # both stamped unconditionally → persisted node shape changed, 8→9. # W21B-w5 STEP4 (2026-05-31): ⑧ apply_space_partition_plan canonical # normalizer consumes bg_space_partition's space_partition_plan; every node # additively carries render_action_source + partition_fallback_reason → # persisted node shape changed, 9→10. # W20F10 (2026-07-23): planner pack v3(graph/anchor 교정 retry 스템) # 승격 — step 은 provider 상수를 단일 SOT 로 소비 (BLOCKING-1). assert SCHEMA_VERSION == 10 assert PROMPT_VERSION == "3.202607231435" def test_prompt_version_single_sot_cross_pin(): """BLOCKING-1 재발 방지: step 이 소비하는 PROMPT_VERSION == provider 상수 (pack 승격 시 config_hash 자동 drift).""" from app.modules.pipeline.shot_aware_bg_render_plan_llm_provider import ( PROMPT_VERSION as provider_version, ) assert PROMPT_VERSION is provider_version def test_config_hash_drifts_on_prompt_version_change(): """pack 승격=완료 CP stale — v2 잔존이 clean-skip 되던 결함 잠금.""" from unittest.mock import patch as _patch step = _new_step() managers = _enable_all() try: for m in managers: m.start() h_now = step._config_hash() with _patch( "app.core.steps.shot_aware_bg_render_plan_step.PROMPT_VERSION", "2.202605291800", ): h_v2 = step._config_hash() finally: for m in managers: m.stop() assert h_now != h_v2 # ──────────────────────────── registry ──────────────────────────── def test_step_registered_in_step_classes(): from app.core.steps import STEP_CLASSES assert "shot_aware_bg_render_plan" in STEP_CLASSES assert ( STEP_CLASSES["shot_aware_bg_render_plan"] is ShotAwareBgRenderPlanStep ) def test_step_registered_in_manifest_between_geometry_and_bg_prompt(): from app.core.step_manifest import STEP_MANIFEST entry = STEP_MANIFEST["shot_aware_bg_render_plan"] assert entry["applicability"] == "if_background_mode" assert entry["step_type"] == "transform" geom_order = STEP_MANIFEST["floor_plan_geometry_readback"]["order"] bgp_order = STEP_MANIFEST["background_prompt"]["order"] assert geom_order < entry["order"] < bgp_order deps = entry["depends_on"] for required in ( "base_location_dossier", "floor_plan_geometry_readback", "floor_plan_overlay_payload", "background_master_plan", "shot_staging", ): assert required in deps # W20D narrow: manifest schema_version must track the step module's # SCHEMA_VERSION so resume / checkpoint shape drift is pinned. assert entry["schema_version"] == SCHEMA_VERSION # ──────────── 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" / "shot_aware_bg_render_plan.py" ) tokens = _import_tokens(mod) leaks = tokens & _BANNED_IMPORT_TOKENS assert not leaks, ( f"shot_aware_bg_render_plan 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" / "shot_aware_bg_render_plan_step.py" ) tokens = _import_tokens(step) leaks = tokens & _BANNED_IMPORT_TOKENS assert not leaks # ───────────── W19B-3 / W20B mutual exclusion guard (shot-aware side) ──────── def _new_shot_aware_step(cp_map=None): """Helper analogous to _new_step but for ShotAwareBgRenderPlanStep.""" step = ShotAwareBgRenderPlanStep.__new__(ShotAwareBgRenderPlanStep) 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 or {}).get(sid) ) return step def test_w19b3_w20b_both_active_fails_closed_before_planner(): """w18j_overlap (W19B-3) AND shot_aware_bg_render_plan_enabled (W20B) 동시 활성 시 shot-aware side step 도 cp / planner / llm call 전에 failed_count=1 + selector_conflict diagnostic 으로 fail-closed.""" step = _new_shot_aware_step() 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), \ patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", True, ), patch( "app.core.config.settings.shot_aware_bg_render_plan_enabled", True ), patch( "app.core.config.settings.background_render_reference_mode", "w18j_overlap", ): result = step._execute() # No cp loads before the gate fires. step._load_prev_checkpoint.assert_not_called() assert result["failed_count"] == 1 assert result["completed_count"] == 0 assert result["applicable_count"] == 1 assert "selector_conflict" in result["data"] assert "w18j_overlap" in result["data"]["selector_conflict"] assert "shot_aware_bg_render_plan_enabled" in ( result["data"]["selector_conflict"] ) assert "mutually exclusive" in result["data"]["selector_conflict"] assert "W19B-3/W20B" in result["error"] assert result["data"]["selector_conflict"] == result["error"] def test_w19b3_w20b_only_shot_aware_active_passes_gate(): """shot_aware_bg_render_plan_enabled True + background_render_reference_mode legacy default 면 conflict 없음 — cp 로드 시작 (test 에서는 빈 cp 라 downstream planner 가 빈 plan + diagnostic emit, 다만 selector_conflict 는 emit 안 됨).""" step = _new_shot_aware_step() 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), \ patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", True, ), patch( "app.core.config.settings.shot_aware_bg_render_plan_enabled", True ), patch( "app.core.config.settings.background_render_reference_mode", "legacy", ): result = step._execute() # Gate 통과 시 step 이 prev cp 로드 호출 (빈 cp_map 이라 downstream # 은 fail / not_ready 이지만 selector_conflict 는 없음). step._load_prev_checkpoint.assert_called() assert "selector_conflict" not in result.get("data", {}) # ─────────────── W20B real-provider selector wire-up tests ──────────────── # # Every test below either stays on the default-OFF selector or uses an # explicit injection helper. NO real litellm completion call is made in # any branch — the closure either falls through to the test-injected # provider, or never resolves (selector OFF), or short-circuits at the # mutex gate. def _patch_real_provider_selector(value: bool): """Patch the W20B real-provider selector. Used to verify the lazy-import branch resolves to the helper symbol (the helper is mock-patched separately, never invoked for real).""" return patch( "app.core.config.settings.shot_aware_bg_render_plan_real_provider_enabled", value, ) def test_resolve_llm_provider_defaults_to_none(): """No override + selector OFF (default) → resolver returns None. build_render_plan_for_fp must not call any provider in this branch.""" step = _new_step() with _patch_real_provider_selector(False): resolved = step._resolve_llm_provider() assert resolved is None def test_resolve_llm_provider_returns_override_when_set(): """Explicit test override bypasses the selector — even when the selector is True the override wins (mock-provider tests must not accidentally invoke the real helper).""" step = _new_step() sentinel = MagicMock(name="test_override_provider") step.set_llm_provider_for_testing(sentinel) with _patch_real_provider_selector(True): resolved = step._resolve_llm_provider() assert resolved is sentinel def test_resolve_llm_provider_returns_real_helper_when_selector_on(): """Selector True + no override → lazy import resolves to the W20B real helper symbol. We compare by identity to the import-time binding; the function body is never invoked here (no completion call ever issued).""" from app.modules.pipeline.shot_aware_bg_render_plan_llm_provider import ( litellm_shot_aware_bg_render_plan_provider, ) step = _new_step() with _patch_real_provider_selector(True): resolved = step._resolve_llm_provider() assert resolved is litellm_shot_aware_bg_render_plan_provider def test_real_provider_selector_changes_config_hash(): """Flipping the W20B real-provider selector must perturb the step's config_hash so previously-stamped checkpoints do not silently shadow a different active configuration.""" step = _new_step() managers_off = _enable_all( shot_aware_bg_render_plan_real_provider_enabled=False, ) try: for m in managers_off: m.start() h_off = step._config_hash() finally: for m in managers_off: m.stop() managers_on = _enable_all( shot_aware_bg_render_plan_real_provider_enabled=True, ) try: for m in managers_on: m.start() h_on = step._config_hash() finally: for m in managers_on: m.stop() assert h_off != h_on def test_default_off_zero_provider_call_when_no_override(): """Default-OFF selector + no override → provider attempts dict is empty + llm count 0 + plan empty. This is the production-safe branch the W20B flag-flip lands without burning an API call.""" step = _new_step() assert step._llm_provider_override is None managers = _enable_all( shot_aware_bg_render_plan_real_provider_enabled=False, ) try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() assert result["data"]["real_api_call_counts"]["llm"] == 0 assert result["data"]["llm_provider_attempts_per_fp"] == {} def test_mock_provider_via_override_records_one_attempt_per_fp(): """Mock provider injected via the test override is invoked exactly once per fp by the step's counted closure. The per-fp attempt counter and the per-fp plan ``real_api_call_counts.llm`` must agree (closure increment ↔ build_render_plan_for_fp bookkeeping).""" step = _new_step() mock_provider = MagicMock(side_effect=_valid_mock_output) step.set_llm_provider_for_testing(mock_provider) managers = _enable_all() try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() # Fixture has exactly one fp. assert mock_provider.call_count == 1 assert result["data"]["llm_provider_attempts_per_fp"] == {"fp_a": 1} assert result["data"]["real_api_call_counts"]["llm"] == 1 # And the closure observed kwargs identical to what # build_render_plan_for_fp passes to the provider. call_kwargs = mock_provider.call_args.kwargs assert call_kwargs["fp_id"] == "fp_a" assert "dossier" in call_kwargs assert "geometry" in call_kwargs assert "shot_readiness" in call_kwargs assert "candidate_catalog" in call_kwargs def test_mock_provider_raise_marks_fp_failed_and_records_attempt(): """A provider that raises must (a) mark the fp 'failed' in the per-fp plan, (b) still record exactly one provider attempt in the counter (we tried once, the call shape itself is the failure), and (c) keep llm count == 1 in the aggregate.""" step = _new_step() def boom(**_kw): raise RuntimeError("upstream LLM down") step.set_llm_provider_for_testing(boom) managers = _enable_all() try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() plan = result["data"]["per_fp"]["fp_a"] assert plan["shot_aware_bg_render_plan_status"] == "failed" assert plan["real_api_call_counts"]["llm"] == 1 assert result["data"]["llm_provider_attempts_per_fp"] == {"fp_a": 1} assert result["data"]["real_api_call_counts"]["llm"] == 1 def test_mutex_short_circuits_before_resolve_or_provider_call(): """w18j_overlap + W20B both ON → mutex error fires BEFORE the resolver runs and BEFORE any provider call. We patch the resolver to a sentinel that would raise if invoked, and assert it is not called (the closure also is not built).""" step = _new_shot_aware_step() # Sentinel resolver: if the gate fails to short-circuit, this # explodes and the test fails with a clear stack trace. sentinel = MagicMock(side_effect=AssertionError( "resolver was called even though mutex should have fired" )) step._resolve_llm_provider = sentinel # type: ignore[assignment] 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), \ patch( "app.core.config.settings.floor_plan_geometry_readback_enabled", True, ), patch( "app.core.config.settings.shot_aware_bg_render_plan_enabled", True ), patch( "app.core.config.settings.background_render_reference_mode", "w18j_overlap", ), patch( "app.core.config.settings.shot_aware_bg_render_plan_real_provider_enabled", True, # would normally resolve to the real helper, but # mutex must fire first ): result = step._execute() sentinel.assert_not_called() assert result["failed_count"] == 1 assert "selector_conflict" in result["data"] def test_mock_provider_return_non_dict_records_attempt_count(): """Provider closure가 호출됐다면 (실제 completion 후 malformed return) attempt counter 와 per-fp ``real_api_call_counts.llm`` 가 모두 1 이어야 함. build_render_plan_for_fp 는 non-dict 반환을 ``ShotAwareBgRenderPlanError`` 로 변환하므로 step 의 expected except branch 가 attempts 를 보존해야 한다.""" step = _new_step() mock_provider = MagicMock(return_value=["not a dict"]) step.set_llm_provider_for_testing(mock_provider) managers = _enable_all() try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() assert mock_provider.call_count == 1 plan = result["data"]["per_fp"]["fp_a"] assert plan["shot_aware_bg_render_plan_status"] == "failed" assert plan["real_api_call_counts"]["llm"] == 1, ( f"non-dict provider return must preserve attempt count, got " f"{plan['real_api_call_counts']}" ) assert result["data"]["llm_provider_attempts_per_fp"] == {"fp_a": 1} assert result["data"]["real_api_call_counts"]["llm"] == 1 assert result["failed_count"] == 1 def test_selector_true_does_not_invoke_litellm_when_no_planner_inputs(): """When opt-in is True but cp loads return nothing (planner_inputs is empty) the step short-circuits to not_applicable WITHOUT resolving / calling the provider. Confirms the resolver runs only after the planner-input gate clears.""" step = ShotAwareBgRenderPlanStep.__new__(ShotAwareBgRenderPlanStep) step.project_id = "p" step.episode_id = "e" step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) step._load_prev_checkpoint = MagicMock(return_value=None) # cp empty step._llm_provider_override = None sentinel_resolver = MagicMock(side_effect=AssertionError( "resolver invoked despite empty planner_inputs" )) step._resolve_llm_provider = sentinel_resolver # type: ignore[assignment] managers = _enable_all( shot_aware_bg_render_plan_real_provider_enabled=True, ) try: for m in managers: m.start() result = step._execute() finally: for m in managers: m.stop() sentinel_resolver.assert_not_called() assert result["applicable_count"] == 0 # ──────────────── W21B-w5 STEP4: bg_space_partition cp wiring ──────────────── def _partition_cp(*, schema=10, fp_id="fp_a", bg="L01B01"): """A bg_space_partition checkpoint that marks the planner's single bg a solo render_new plate — enough to prove the ⑧ normalizer consumed it.""" return { "schema_version": schema, "data": {"per_fp": {fp_id: { "status": "ok", "space_partition_plan": { "render_actions": { bg: {"render_action": "render_new_plate", "reuse_target_bg_id": ""}}, "ref_tree_parents": {bg: []}, "node_assignments": {bg: f"grp::{bg}"}, "plate_groups": [{"plate_group_id": f"grp::{bg}", "anchor_bg_id": bg, "member_bg_ids": []}], }, }}}, } def _run_with_cp(step, extra_cp): local_map = dict(_CP_MAP) local_map.update(extra_cp) step._load_prev_checkpoint = MagicMock(side_effect=lambda sid: local_map.get(sid)) step.set_llm_provider_for_testing(_valid_mock_output) managers = _enable_all() try: for m in managers: m.start() return step._execute() finally: for m in managers: m.stop() def test_partition_cp_present_drives_canonical_source(): step = _new_step() # the step's expected partition schema (BG_SPACE_PARTITION_SCHEMA) is 1. from app.core.steps.shot_aware_bg_render_plan_step import BG_SPACE_PARTITION_SCHEMA result = _run_with_cp( step, {"bg_space_partition": _partition_cp(schema=BG_SPACE_PARTITION_SCHEMA)} ) plan = result["data"]["per_fp"]["fp_a"] node = plan["graph"]["nodes"][0] assert node["render_action_source"] == "space_partition_plan" assert node["partition_fallback_reason"] == "" assert plan["validators"]["space_partition_applied"] is True def test_partition_cp_absent_falls_back_to_geometry(): step = _new_step() # _CP_MAP carries no bg_space_partition entry result = _run_with_cp(step, {}) plan = result["data"]["per_fp"]["fp_a"] node = plan["graph"]["nodes"][0] assert node["render_action_source"] == "geometry_route" assert node["partition_fallback_reason"] == "partition_plan_missing" assert plan["validators"]["space_partition_applied"] is False def test_partition_cp_schema_mismatch_fails_closed(): step = _new_step() # a partition cp whose schema does NOT match BG_SPACE_PARTITION_SCHEMA is # ignored entirely (fail-closed) → geometry fallback, never half-consumed. result = _run_with_cp(step, {"bg_space_partition": _partition_cp(schema=999)}) plan = result["data"]["per_fp"]["fp_a"] node = plan["graph"]["nodes"][0] assert node["render_action_source"] == "geometry_route" assert plan["validators"]["space_partition_applied"] is False # ──────────────── W21B Phase 2: dwelling_zone_map cp lookup ──────────────── def test_zone_map_lookup_filters_synthetic_and_missing(): # The lookup keeps only NON-synthetic per-dwelling zone_maps; a missing cp # or a synthetic fixture yields no entry → zone_plan=None → legacy partition # path (byte-identical). Multi-zone consolidation is exercised end-to-end on # real data (visual gate), not in this single-bg unit fixture. assert ShotAwareBgRenderPlanStep._build_zone_map_lookup(None) == {} cp = {"data": {"per_dwelling": { "fp_a": {"synthetic": False, "bg_zone_assignments": {"L01B01": {"zone_id": "Z01"}}}, "fp_b": {"synthetic": True, "bg_zone_assignments": {}}, }}} lookup = ShotAwareBgRenderPlanStep._build_zone_map_lookup(cp) assert set(lookup) == {"fp_a"}