"""W20F narrow wave focused tests (Codex 2026-05-28 directive). 3 deterministic minimal tests — F1 / F2 / F3. No broad regression here. - F1: `_index_shot_staging` 가 shot_id/id 부재 entry 를 scene_index+shot_index 로 합성해 `S{si}_Shot{sh}` 키로 인덱싱하고, bool 은 reject 한다. - F2: `validate_visible_entities_contract` 의 entity_canon.name ±60 char window 분기는 raise 가 아니라 audit_warnings 리스트에 push 한다. - F3: `merge_owned_object_usage` 의 unknown owned_token 은 hard fail 대신 drop + diagnostics["rejected_owned_tokens"] 로 carry, 나머지 token 은 정상 merge / 누락은 absent skeleton 으로 채워진다. """ from __future__ import annotations import pytest from app.core.steps._owned_helpers import merge_owned_object_usage from app.core.steps.background_master_plan_step import BackgroundMasterPlanStep from app.core.visible_entities_validator import ( validate_visible_entities_contract, ) from app.modules.pipeline.base_location_dossier import ( _anchor_candidate_surface, ) from app.modules.pipeline.shot_aware_bg_render_plan import _index_shot_staging # ───────────────────────────────────────────────────────────────────── # F1 — synthesized shot_id from scene_index + shot_index # ───────────────────────────────────────────────────────────────────── def test_f1_index_shot_staging_synthesizes_id_from_scene_shot_index(): cp = { "shots": [ {"scene_index": 12, "shot_index": 5, "framing_scale": "wide"}, {"scene_index": 22, "shot_index": 2, "framing_scale": "wide"}, {"shot_id": "explicit_A", "scene_index": 9, "shot_index": 9}, {"scene_index": True, "shot_index": 1}, # bool reject {"scene_index": 3, "shot_index": "nope"}, # bad type reject "not_a_dict", # non-dict reject ] } diag: dict = {} out = _index_shot_staging(cp, diagnostics=diag) assert set(out.keys()) == {"S12_Shot5", "S22_Shot2", "explicit_A"} assert out["S12_Shot5"]["scene_index"] == 12 assert diag["shot_staging_index_explicit_id"] == 1 assert diag["shot_staging_index_synthesized_id"] == 2 assert len(diag["shot_staging_index_dropped_no_id"]) == 1 assert len(diag["shot_staging_index_dropped_bad_index_type"]) == 2 assert diag["shot_staging_index_collisions"] == [] # ───────────────────────────────────────────────────────────────────── # F2 — entity_canon.name ±60 char window 는 audit warning # ───────────────────────────────────────────────────────────────────── def _make_shot_name_with_wrong_window_id() -> dict: # EP1 S29_Shot8 패턴 재현 — visible {C01,C03}. prompt 안에 '수리영'(C01) # 가 등장하지만 그 위치 ±60자 window 에는 C03O01 만 있고 C01O02 는 멀리 # 떨어져 있어 옛 contract 라면 raise 분기. 새 contract: raise 없이 audit # warnings 에 1 건 push. prompt = ( "수리영 watches as C03O01 walks past the open door. " + " " * 60 + " Further away C01O02 sits on the chair." ) return { "scene_index": 29, "shot_index": 8, "visible_entities": ["C01O02", "C03O01"], "t2i_variations": [{"t2i_prompt": prompt}], "render_prompt_card": { "asset_requirements": {"required_refs": []}, "id_policy": { "allowed_outlook_pairs": [ { "base_id": "C01", "composite_id": "C01O02", "outlook_id": "C01O02", }, { "base_id": "C03", "composite_id": "C03O01", "outlook_id": "C03O01", }, ], "subject_reference_policy": [ { "subject_id": "C01", "policy_type": "identity_reference", "policy": "id_and_outlook_required", "reason": "test fixture", }, { "subject_id": "C03", "policy_type": "identity_reference", "policy": "id_and_outlook_required", "reason": "test fixture", }, ], }, }, } def test_f2_entity_name_no_specific_id_emits_audit_warning_not_raise(): shot = _make_shot_name_with_wrong_window_id() audit: list = [] # 옛 동작이면 AppError raise — 새 동작은 raise 없이 audit 에 1 건 push. validate_visible_entities_contract( shot, {"C01": "수리영", "C03": "인우"}, audit_warnings=audit, ) codes = [w.get("code") for w in audit] assert ( "step.scene_detail.audit_warning_entity_name_no_specific_id" in codes ), f"expected audit warning, got audit={audit!r}" one = next( w for w in audit if w["code"] == ( "step.scene_detail.audit_warning_entity_name_no_specific_id" ) ) assert one["base_id"] == "C01" assert one["entity_name"] == "수리영" assert "C01O02" in one["entity_id_candidates"] # ───────────────────────────────────────────────────────────────────── # F3 — unknown owned_token → drop + diagnostics carry # ───────────────────────────────────────────────────────────────────── def test_f3_unknown_owned_token_dropped_into_diagnostics(): owned = ["TV", "bed", "blood", "chair", "cup"] llm_emit = [ { "owned_token": "TV", "usage_kind": "anchor", "source_phrase": "TV stays off in the corner", }, { "owned_token": "dark red paint", # ← unknown, drop expected "usage_kind": "redraw", "source_phrase": "dark red paint on the wall", }, { "owned_token": "blood", "usage_kind": "redraw", "source_phrase": "fresh blood on the floor", }, ] diag: dict = {} merged = merge_owned_object_usage( llm_emit, owned, is_close_framing=False, where="S12_Shot5", diagnostics=diag, ) # skeleton cardinality == owned list len, 순서도 owned 순서. assert [e["owned_token"] for e in merged] == owned by_tok = {e["owned_token"]: e for e in merged} assert by_tok["TV"]["usage_kind"] == "anchor" assert by_tok["blood"]["usage_kind"] == "redraw" assert by_tok["bed"]["usage_kind"] == "absent" # 누락 → absent default assert by_tok["chair"]["usage_kind"] == "absent" assert by_tok["cup"]["usage_kind"] == "absent" rejected = diag.get("rejected_owned_tokens") or [] assert len(rejected) == 1 assert rejected[0]["owned_token"] == "dark red paint" assert rejected[0]["where"] == "S12_Shot5" def test_f5_anchor_candidate_surface_ramps_when_no_clean(): # Clean path: 1+ overlay with clean_background_expected=True 가 있으면 # 그대로 candidate, fallback 안 탄다. overlays_clean = { "L09B03": { "clean_background_expected": True, "transient_markers_to_describe": [], }, "L09B01": { "clean_background_expected": False, "transient_markers_to_describe": [16, 17], }, } out_clean = _anchor_candidate_surface(overlays_clean) assert out_clean["candidate_bg_ids"] == ["L09B03"] assert "ramped_fallback" not in out_clean assert all("W20F5" not in d for d in out_clean["selection_diagnostics"]) # Ramped fallback path: 모든 overlay 가 clean_background_expected=False. # transient marker 가 가장 적은 BG 만 ramped candidate 로 surface 된다. overlays_ramped = { "L04B01": { "clean_background_expected": False, "transient_markers_to_describe": [16, 17, 18], }, "L04B02": { "clean_background_expected": False, "transient_markers_to_describe": [16], # ← min }, "L04B05": { "clean_background_expected": False, "transient_markers_to_describe": [16, 18], }, } out_ramped = _anchor_candidate_surface(overlays_ramped) assert out_ramped["candidate_bg_ids"] == ["L04B02"] assert out_ramped.get("ramped_fallback") is True assert any("W20F5" in d for d in out_ramped["selection_diagnostics"]) # Empty overlays — 여전히 empty candidate (fail-closed downstream). out_empty = _anchor_candidate_surface({}) assert out_empty["candidate_bg_ids"] == [] assert any("no overlays" in d for d in out_empty["selection_diagnostics"]) def test_f6_floor_plan_scope_filter_drops_orphan_bg_and_cascades(): """W21B-wave-2 replacement policy — surface_role aware trim + cascade. W20F6 의 indoor hard filter 는 exterior plate 를 drop 했다. 새 정책은 exterior/transition/site background 를 keep 하되, consuming shot 이 없는 background 와 그 background 를 depends_on_bg 로 참조하는 BG 만 drop 한다. """ step = BackgroundMasterPlanStep.__new__(BackgroundMasterPlanStep) groups = [{ "group_id": "g_test", "kind": "chain_bg", "anchor_loc": "L_IN", "members": [ {"loc_id": "L_IN", "is_indoor": True, "shot_count": 5}, {"loc_id": "L_OUT", "is_indoor": False, "shot_count": 3}, {"loc_id": "L_LOW", "is_indoor": True, "shot_count": 1}, ], }] plan_payload = { "group_id": "g_test", "floor_plans": [ {"fp_id": "fp_in", "loc_id": "L_IN", "space_key_hint": "main", "sub_location": "x", "scope": "y", "depends_on_fp": []}, {"fp_id": "fp_out", "loc_id": "L_OUT", "space_key_hint": "main", "sub_location": "x", "scope": "y", "depends_on_fp": []}, {"fp_id": "fp_low", "loc_id": "L_LOW", "space_key_hint": "main", "sub_location": "x", "scope": "y", "depends_on_fp": []}, ], "backgrounds": [ {"loc_id": "L_IN", "space_key_hint": "main", "time_phase": "day", "state_class": "quiet", "applies_to_shots": ["S1_Shot1"], "surface_role": "interior_room", "sub_location_label": "a", "state_label_raw": "a", "depends_on_fp": ["fp_in"], "depends_on_bg": []}, {"loc_id": "L_OUT", "space_key_hint": "main", "time_phase": "day", "state_class": "quiet", "applies_to_shots": ["S2_Shot1"], "surface_role": "exterior_plate", "sub_location_label": "b", "state_label_raw": "b", "depends_on_fp": [], "depends_on_bg": []}, {"loc_id": "L_OUT", "space_key_hint": "main", "time_phase": "dusk", "state_class": "quiet", "applies_to_shots": ["S2_Shot2"], "surface_role": "transition_zone", "sub_location_label": "b2", "state_label_raw": "b2", "depends_on_fp": ["fp_out"], "depends_on_bg": []}, {"loc_id": "L_IN", "space_key_hint": "main", "time_phase": "day", "state_class": "ransacked", "applies_to_shots": ["S3_Shot1"], "surface_role": "interior_room", "sub_location_label": "c", "state_label_raw": "c", "depends_on_fp": ["fp_in"], # exact raw key of the dropped low-shot/no-consuming background below. "depends_on_bg": ["L_LOW|main|day|quiet"]}, {"loc_id": "L_LOW", "space_key_hint": "main", "time_phase": "day", "state_class": "quiet", "applies_to_shots": [], "surface_role": "interior_room", "sub_location_label": "d", "state_label_raw": "d", "depends_on_fp": ["fp_low"], "depends_on_bg": []}, ], } ordered_plans = {"g_test": {"status": "ok", "plan": plan_payload}} diag = step._apply_w21b_surface_role_scope_filter(ordered_plans, groups) # kept floor_plans + backgrounds: exterior fp-less plate survives, and # transition_zone keeps its referenced fp. kept_plan = ordered_plans["g_test"]["plan"] assert [fp["fp_id"] for fp in kept_plan["floor_plans"]] == ["fp_in", "fp_out"] kept_bg_locs = [(bg["loc_id"], bg["state_class"]) for bg in kept_plan["backgrounds"]] assert kept_bg_locs == [("L_IN", "quiet"), ("L_OUT", "quiet"), ("L_OUT", "quiet")] # diagnostics assert diag["kept_floor_plans_count"] == 2 assert diag["kept_backgrounds_count"] == 3 assert diag["dropped_floor_plans_count"] == 1 # fp_low assert diag["dropped_backgrounds_count"] == 2 # bgD + bgC cascade dropped_fp_ids = {d["fp_id"] for d in diag["dropped_floor_plans"]} assert dropped_fp_ids == {"fp_low"} reasons_by_fp = {d["fp_id"]: d["reasons"] for d in diag["dropped_floor_plans"]} assert "no_kept_background_ref" in reasons_by_fp["fp_low"] # transitive cascade — bgC 가 depends_on_dropped_bg reason 으로 drop reasons_for_bgs = [d["reasons"] for d in diag["dropped_backgrounds"]] flat = {r for rs in reasons_for_bgs for r in rs} assert "no_consuming_shot" in flat assert "depends_on_dropped_bg" in flat # orphan depends_on_fp 없음 (hard assertion — helper 안에서도 raise 함) for bg in kept_plan["backgrounds"]: for ref in bg.get("depends_on_fp", []): assert ref in {"fp_in", "fp_out"} def test_f9_camera_only_failure_classifier(): """W20F9 helper — camera_in_candidates 만 fail 인지 분류기.""" from app.modules.pipeline.shot_aware_bg_render_plan_llm_provider import ( _is_camera_only_failure, ) keys_ok = { "required_shape_ok": True, "graph_completeness_ok": True, "node_index_order_ok": True, "dag_ok": True, "max_refs_per_bg_ok": True, "two_refs_distinct_spaces_ok": True, "same_fp_only_ok": True, "anchor_exactly_one_ok": True, "anchor_in_clean_candidate_set_ok": True, "rationale_and_mode_ok": True, "synthetic_readback_production_clear": True, } # camera fail + 다른 모두 ok → retry 가능. v_camera = dict(keys_ok, camera_in_candidates_ok=False) assert _is_camera_only_failure(v_camera) is True # camera ok → retry 불필요 (False). v_pass = dict(keys_ok, camera_in_candidates_ok=True) assert _is_camera_only_failure(v_pass) is False # camera fail + 다른 곳 도 fail → retry 금지 (Codex 명시). v_mixed = dict(keys_ok, camera_in_candidates_ok=False, dag_ok=False) assert _is_camera_only_failure(v_mixed) is False v_mixed_graph = dict( keys_ok, camera_in_candidates_ok=False, graph_completeness_ok=False, ) assert _is_camera_only_failure(v_mixed_graph) is False # camera_in_candidates_ok 누락 → retry 금지 (defensive). v_missing = dict(keys_ok) assert _is_camera_only_failure(v_missing) is False def test_f3_malformed_owned_entry_still_hard_fails(): # malformed shape (usage_kind enum 밖) 은 W20F3 강등 대상이 아님. from app.core.errors import AppError owned = ["TV", "blood"] bad = [ { "owned_token": "TV", "usage_kind": "set_on_fire", # enum 밖 "source_phrase": "burned", }, ] with pytest.raises(AppError): merge_owned_object_usage(bad, owned, is_close_framing=False)