"""W21B-wave-4 C2: shot_projection_card deterministic core (TDD). LLM/VLM/image/DB I/O 0 — pure deterministic functions only. The real VLM provider is exercised separately (default OFF). These tests pin the input-assembly (overlay union registry), source-hash, envelope, and the v0 deterministic gate (validator) contract that the wiring brief §4/§5/§9 and the design brief §4/§5/§9 locked. """ from __future__ import annotations import pytest from app.modules.pipeline.shot_projection_card import ( ProjectionCardError, build_card_envelope, build_union_registry, compute_card_cache_key, compute_projection_card, compute_source_hashes, compute_synthetic_fixture, validate_shot_projection_card, validate_vlm_output, ) def _hash_inputs(**over): base = dict( fp_render={"png": "abc"}, substrate={"raster": "sub"}, overlay={"ov": 1}, geometry={"geo": 1}, semantic=None, camera_rec={"cam": 1}, shot_context={"scene_text": "방", "shot_guides": ["a"], "visible_entities": [3, 7]}, prompt_version="1", schema_version=1, model="gpt", provider="openai", pack_version="1.202605302212", dossier={"fp_id": "fp_x", "base_marker_inventory": []}, ) base.update(over) return base class TestBuildUnionRegistry: def test_union_of_base_transient_ignored_with_camera_ref(self): ov_entry = { "base_markers_to_reference": [ {"number": 3, "label": "sink"}, {"number": 7, "label": "sliding door"}, ], "transient_markers_to_describe": [ {"number": 18, "label": "temporary cartons"}, ], "ignored_state_overlay_markers": [ {"number": 25, "label": "blood pool overlay"}, ], "use_numbered_elements": [3, 7, 18], "ignore_numbered_elements": [], } inventory, registry = build_union_registry(ov_entry) assert registry["base_numbers"] == [3, 7] assert registry["transient_numbers"] == [18] assert registry["ignored_numbers"] == [25] assert registry["union_numbers"] == [3, 7, 18, 25] assert registry["camera_referenced_numbers"] == [3, 7, 18] assert registry["out_of_union_referenced"] == [] # inventory carries each marker's layer verbatim layers = {m["number"]: m["marker_layer"] for m in inventory} assert layers == {3: "base", 7: "base", 18: "transient", 25: "ignored_state_overlay"} def test_out_of_union_when_camera_refs_unknown_marker(self): ov_entry = { "base_markers_to_reference": [{"number": 3, "label": "sink"}], "transient_markers_to_describe": [], "ignored_state_overlay_markers": [], "use_numbered_elements": [3, 99], "ignore_numbered_elements": [], } _inventory, registry = build_union_registry(ov_entry) assert registry["out_of_union_referenced"] == [99] def test_ignore_numbered_elements_subtracted_from_camera_ref(self): ov_entry = { "base_markers_to_reference": [{"number": 3, "label": "sink"}], "transient_markers_to_describe": [], "ignored_state_overlay_markers": [], "use_numbered_elements": [3, 7], "ignore_numbered_elements": [7], } _inventory, registry = build_union_registry(ov_entry) assert registry["camera_referenced_numbers"] == [3] assert registry["out_of_union_referenced"] == [] class TestSourceHashes: def test_deterministic_same_inputs_same_hashes(self): a = compute_source_hashes(**_hash_inputs()) b = compute_source_hashes(**_hash_inputs()) assert a == b def test_semantic_none_records_not_available_sentinel(self): a = compute_source_hashes(**_hash_inputs(semantic=None)) assert a["semantic_hash"] == "not_available" def test_semantic_present_is_hashed(self): a = compute_source_hashes(**_hash_inputs(semantic={"gate": "pass"})) assert a["semantic_hash"] != "not_available" assert len(a["semantic_hash"]) >= 8 def test_shot_context_hash_is_key_order_normalized(self): # same content, different dict key order → same shot_context_hash a = compute_source_hashes( **_hash_inputs(shot_context={"a": 1, "b": 2}) ) b = compute_source_hashes( **_hash_inputs(shot_context={"b": 2, "a": 1}) ) assert a["shot_context_hash"] == b["shot_context_hash"] def test_shot_context_content_change_invalidates(self): a = compute_source_hashes(**_hash_inputs(shot_context={"scene_text": "방"})) b = compute_source_hashes(**_hash_inputs(shot_context={"scene_text": "마트"})) assert a["shot_context_hash"] != b["shot_context_hash"] def test_version_and_model_provider_carried(self): a = compute_source_hashes(**_hash_inputs(prompt_version="2", schema_version=3)) assert a["prompt_version"] == "2" assert a["schema_version"] == 3 assert a["model"] == "gpt" assert a["provider"] == "openai" class TestCardCacheKey: def test_combines_source_hashes_with_bg_and_shot(self): sh = compute_source_hashes(**_hash_inputs()) k1 = compute_card_cache_key(source_hashes=sh, bg_id="L05B09", shot_id="S5_Shot1") k2 = compute_card_cache_key(source_hashes=sh, bg_id="L05B09", shot_id="S5_Shot1") assert k1 == k2 def test_different_shot_changes_cache_key(self): sh = compute_source_hashes(**_hash_inputs()) k1 = compute_card_cache_key(source_hashes=sh, bg_id="L05B09", shot_id="S5_Shot1") k2 = compute_card_cache_key(source_hashes=sh, bg_id="L05B09", shot_id="S5_Shot7") assert k1 != k2 def test_changed_source_hash_changes_cache_key(self): sh1 = compute_source_hashes(**_hash_inputs(shot_context={"x": 1})) sh2 = compute_source_hashes(**_hash_inputs(shot_context={"x": 2})) k1 = compute_card_cache_key(source_hashes=sh1, bg_id="L05B09", shot_id="S5_Shot1") k2 = compute_card_cache_key(source_hashes=sh2, bg_id="L05B09", shot_id="S5_Shot1") assert k1 != k2 # ── v0 deterministic gate (validator) ──────────────────────────────── _GOOD_HASHES = { "fp_render_hash": "fp_abc", "substrate_hash": "sub_abc", "overlay_hash": "ov_abc", "geometry_hash": "geo_abc", "semantic_hash": "not_available", "camera_rec_hash": "cam_abc", "shot_context_hash": "shot_abc", "prompt_version": "1", "schema_version": 1, "model": "gpt", "provider": "openai", } def _base_card(**over): card = { "card_id": "card-0001", "schema_version": 1, "prompt_version": "1", "model": "gpt", "provider": "openai", "bg_id": "L05B09", "shot_id": "S5_Shot1", "fp_id": "fp_l05_main", "semantic_gate_state": "not_available", "substrate_kind": "both", "substrate_status": "ok", "source_hashes": dict(_GOOD_HASHES), "vlm_output": { "status": "ok", "fp_id": "fp_l05_main", "bg_id": "L05B09", "shot_id": "S5_Shot1", "camera_pose_source": "cam_rec#2", "visible_items": [ {"marker_number": 3, "marker_layer": "base", "expected_label": "싱크대", "visibility": "visible", "horizontal_band": "left", "depth_band": "foreground", "occlusion_note": "", "evidence": "좌측 전경 스테인리스 표면", "source_ref": "fp", "confidence": 0.8}, {"marker_number": 7, "marker_layer": "base", "expected_label": "미닫이문", "visibility": "partial", "horizontal_band": "right", "depth_band": "background", "occlusion_note": "식탁에 일부 가림", "evidence": "우측 배경 문틀", "source_ref": "fp", "confidence": 0.7}, ], "scene_visible_description": "전경 왼쪽에 낡은 스테인리스 싱크대와 낮은 조리대, 중앙에 좁은 식탁, 배경 오른쪽에 미닫이문 두 개가 보인다.", "bg_plate_visible_description": "전경 왼쪽에 낡은 스테인리스 싱크대와 낮은 조리대, 중앙에 좁은 식탁, 배경 오른쪽에 미닫이문 두 개가 자리한 좁은 실내.", "plate_description_excludes_transient": "yes", "not_visible_or_occluded_summary": "천장 조명은 프레임 밖이라 보이지 않는다.", "vlm_reported_state": "ok", "self_consistency": {"prose_matches_structured": "consistent", "notes": ""}, "confidence": 0.78, "missing_inputs": [], "diagnostics": [], }, } for k, v in over.items(): if k == "vlm_output": card["vlm_output"].update(v) else: card[k] = v return card def _gate(card, cur=None): return validate_shot_projection_card( card=card, current_source_hashes=cur if cur is not None else dict(_GOOD_HASHES) ) class TestGate: def test_clean_pass(self): r = _gate(_base_card()) assert r["card_state"] == "pass" assert r["validator_state"] == "ok" assert r["leaks"] == [] def test_missing_envelope_key_malformed_blocked(self): card = _base_card() del card["source_hashes"] r = _gate(card) assert r["validator_state"] == "malformed" assert r["card_state"] == "blocked" def test_marker_number_annotation_leak_blocked(self): r = _gate(_base_card(vlm_output={ "scene_visible_description": "전경 왼쪽에 싱크대(marker 3), 배경 오른쪽에 미닫이문이 보인다.", })) assert r["card_state"] == "blocked" assert r["validator_state"] == "leak_detected" def test_english_enum_literal_leak_blocked(self): r = _gate(_base_card(vlm_output={ "bg_plate_visible_description": "foreground left: 싱크대; background right: 미닫이문.", })) assert r["card_state"] == "blocked" assert r["validator_state"] == "leak_detected" def test_korean_spatial_words_are_not_leaks(self): # 왼쪽/중앙/배경/전경 are natural Korean — never flagged even though # english enum literals (left/center/background/foreground) exist. r = _gate(_base_card()) assert r["leaks"] == [] assert r["card_state"] == "pass" def test_english_word_containing_enum_substring_not_flagged(self): # 'background' literal must match as a whole token, not as a # substring of an unrelated english word. r = _gate(_base_card(vlm_output={ "not_visible_or_occluded_summary": "The grounding is offscreen.", })) # 'grounding' contains 'ground' but NOT the literal 'background'/'foreground' assert r["leaks"] == [] def test_card_id_leak_blocked(self): r = _gate(_base_card(vlm_output={ "scene_visible_description": "이 장면은 card-0001 기준이다.", })) assert r["card_state"] == "blocked" assert r["validator_state"] == "leak_detected" def test_stale_source_hash_blocked(self): stale = dict(_GOOD_HASHES); stale["shot_context_hash"] = "CHANGED" r = _gate(_base_card(), cur=stale) assert r["card_state"] == "blocked" def test_self_consistency_contradictory_blocked(self): r = _gate(_base_card(vlm_output={ "self_consistency": {"prose_matches_structured": "contradictory", "notes": "x"}, })) assert r["card_state"] == "blocked" assert r["validator_state"] == "contradiction" def test_semantic_needs_fix_blocked(self): r = _gate(_base_card(semantic_gate_state="needs_fix")) assert r["card_state"] == "blocked" def test_substrate_missing_blocked(self): r = _gate(_base_card(substrate_status="missing")) assert r["card_state"] == "blocked" def test_plate_transient_self_report_no_blocked(self): r = _gate(_base_card(vlm_output={"plate_description_excludes_transient": "no"})) assert r["card_state"] == "blocked" assert r["validator_state"] == "plate_transient_leak" def test_plate_transient_self_report_uncertain_needs_review(self): r = _gate(_base_card(vlm_output={"plate_description_excludes_transient": "uncertain"})) assert r["card_state"] == "needs_review" def test_missing_inputs_is_needs_review_not_block(self): r = _gate(_base_card(vlm_output={"missing_inputs": ["overlay legend 누락"]})) assert r["card_state"] == "needs_review" def test_low_confidence_needs_review(self): r = _gate(_base_card(vlm_output={"confidence": 0.2})) assert r["card_state"] == "needs_review" def test_transient_marker_in_union_passes(self): card = _base_card( marker_registry={ "base_numbers": list(range(1, 18)), "transient_numbers": [18], "ignored_numbers": [], "union_numbers": list(range(1, 19)), "camera_referenced_numbers": [2, 4, 18], "out_of_union_referenced": [], }, vlm_output={ "visible_items": [ {"marker_number": 18, "marker_layer": "transient", "expected_label": "cartons", "visibility": "partial", "horizontal_band": "left", "depth_band": "foreground", "occlusion_note": "", "evidence": "좌측 전경 상자", "source_ref": "transient", "confidence": 0.7}, ], }, ) r = _gate(card) assert r["card_state"] == "pass" def test_camera_ref_out_of_union_blocked(self): card = _base_card( marker_registry={ "base_numbers": list(range(1, 18)), "transient_numbers": [18], "ignored_numbers": [], "union_numbers": list(range(1, 19)), "camera_referenced_numbers": [2, 99], "out_of_union_referenced": [99], }, ) r = _gate(card) assert r["card_state"] == "blocked" assert r["validator_state"] == "marker_contract_gap" # ── vlm-output shape validation + synthetic + envelope + dispatcher ── _INVENTORY = [ {"number": 3, "expected_label": "싱크대", "marker_layer": "base"}, {"number": 7, "expected_label": "미닫이문", "marker_layer": "base"}, {"number": 18, "expected_label": "cartons", "marker_layer": "transient"}, ] _REGISTRY = { "base_numbers": [3, 7], "transient_numbers": [18], "ignored_numbers": [], "union_numbers": [3, 7, 18], "camera_referenced_numbers": [3, 7, 18], "out_of_union_referenced": [], } def _good_vlm_output(**over): out = { "status": "ok", "fp_id": "fp_l05_main", "bg_id": "L05B09", "shot_id": "S5_Shot1", "camera_pose_source": "cam_rec#2", "visible_items": [ {"marker_number": 3, "marker_layer": "base", "expected_label": "싱크대", "visibility": "visible", "horizontal_band": "left", "depth_band": "foreground", "occlusion_note": "", "evidence": "좌측 전경", "source_ref": "fp", "confidence": 0.8}, ], "scene_visible_description": "전경 왼쪽에 싱크대.", "bg_plate_visible_description": "전경 왼쪽에 싱크대가 있는 좁은 실내.", "plate_description_excludes_transient": "yes", "not_visible_or_occluded_summary": "", "vlm_reported_state": "ok", "self_consistency": {"prose_matches_structured": "consistent", "notes": ""}, "confidence": 0.8, "missing_inputs": [], "diagnostics": [], } out.update(over) return out class TestValidateVlmOutput: def test_good_output_ok(self): res = validate_vlm_output( output=_good_vlm_output(), inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1", ) assert res["ok"] is True assert res["blockers"] == [] def test_non_dict_blocked(self): res = validate_vlm_output( output=["not", "a", "dict"], inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1", ) assert res["ok"] is False def test_id_mismatch_blocked(self): res = validate_vlm_output( output=_good_vlm_output(bg_id="WRONG"), inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1", ) assert res["ok"] is False def test_relabelled_marker_blocked(self): # VLM echoes a label that disagrees with the inventory → blocked bad = _good_vlm_output() bad["visible_items"][0]["expected_label"] = "WRONG LABEL" res = validate_vlm_output( output=bad, inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1", ) assert res["ok"] is False def test_unknown_marker_number_blocked(self): bad = _good_vlm_output() bad["visible_items"][0]["marker_number"] = 999 res = validate_vlm_output( output=bad, inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1", ) assert res["ok"] is False def test_bad_enum_blocked(self): bad = _good_vlm_output() bad["visible_items"][0]["horizontal_band"] = "diagonal" res = validate_vlm_output( output=bad, inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1", ) assert res["ok"] is False class TestSyntheticFixture: def test_synthetic_never_auto_passes(self): out = compute_synthetic_fixture( inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1", ) # synthetic asserts nothing about real image content assert out["status"] == "ok" assert out["vlm_reported_state"] == "insufficient_evidence" # echoes inventory markers verbatim nums = {i["marker_number"] for i in out["visible_items"]} assert nums == {3, 7, 18} def test_synthetic_envelope_does_not_pass_gate(self): out = compute_synthetic_fixture( inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1", ) env = build_card_envelope( card_id="c1", schema_version=1, prompt_version="1", model="gpt", provider="openai", bg_id="L05B09", shot_id="S5_Shot1", fp_id="fp_l05_main", semantic_gate_state="not_available", substrate_kind="both", substrate_status="ok", source_hashes=dict(_GOOD_HASHES), vlm_output=out, marker_registry=_REGISTRY, ) r = validate_shot_projection_card(card=env, current_source_hashes=dict(_GOOD_HASHES)) assert r["card_state"] != "pass" class TestBuildCardEnvelope: def test_has_all_required_keys(self): env = build_card_envelope( card_id="c1", schema_version=1, prompt_version="1", model="gpt", provider="openai", bg_id="L05B09", shot_id="S5_Shot1", fp_id="fp_l05_main", semantic_gate_state="not_available", substrate_kind="both", substrate_status="ok", source_hashes=dict(_GOOD_HASHES), vlm_output=_good_vlm_output(), marker_registry=_REGISTRY, ) for k in ("card_id", "schema_version", "prompt_version", "bg_id", "shot_id", "fp_id", "semantic_gate_state", "substrate_kind", "substrate_status", "source_hashes", "vlm_output", "marker_registry"): assert k in env class TestComputeProjectionCard: def test_default_synthetic_path_no_provider(self): out = compute_projection_card( inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1", vlm_provider=None, ) assert out["vlm_reported_state"] == "insufficient_evidence" def test_provider_output_is_validated(self): def good_provider(**kw): return _good_vlm_output() out = compute_projection_card( inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1", vlm_provider=good_provider, fp_image_path="/x.png", ) assert out["status"] == "ok" def test_bad_provider_output_raises(self): def bad_provider(**kw): return _good_vlm_output(bg_id="WRONG") with pytest.raises(ProjectionCardError): compute_projection_card( inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1", vlm_provider=bad_provider, fp_image_path="/x.png", ) def test_non_dict_provider_output_raises(self): def bad_provider(**kw): return "not a dict" with pytest.raises(ProjectionCardError): compute_projection_card( inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1", vlm_provider=bad_provider, fp_image_path="/x.png", ) # ── per-(bg,shot) target iteration (pure) ───────────────────────────── class TestIterCardTargets: def test_one_target_per_bg_shot_with_fp_and_overlay(self): from app.modules.pipeline.shot_projection_card import iter_card_targets overlay_payload = {"overlays": { "L05B09": {"bg_id": "L05B09", "fp_id": "fp_l05_main", "base_markers_to_reference": [{"number": 3, "label": "sink"}], "use_numbered_elements": [3]}, }} master_plan = {"background_catalog": { "L05B09": {"applies_to_shots": ["S5_Shot1", "S5_Shot7"]}, }} targets = iter_card_targets(overlay_payload=overlay_payload, master_plan=master_plan) pairs = {(t["bg_id"], t["shot_id"]) for t in targets} assert pairs == {("L05B09", "S5_Shot1"), ("L05B09", "S5_Shot7")} assert all(t["fp_id"] == "fp_l05_main" for t in targets) assert all(t["ov_entry"]["bg_id"] == "L05B09" for t in targets) def test_bg_without_applies_to_shots_is_skipped(self): from app.modules.pipeline.shot_projection_card import iter_card_targets overlay_payload = {"overlays": {"L09B04": {"bg_id": "L09B04", "fp_id": "fp_x"}}} master_plan = {"background_catalog": {"L09B04": {"applies_to_shots": []}}} targets = iter_card_targets(overlay_payload=overlay_payload, master_plan=master_plan) assert targets == [] def test_empty_inputs_no_targets(self): from app.modules.pipeline.shot_projection_card import iter_card_targets assert iter_card_targets(overlay_payload={}, master_plan={}) == [] class TestProviderPromptContext: def test_prompt_context_threaded_to_provider(self): seen = {} def capturing_provider(**kw): seen.update(kw) return _good_vlm_output() ctx = {"camera_rec": {"bg_id": "L05B09"}, "registry": _REGISTRY} compute_projection_card( inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1", vlm_provider=capturing_provider, fp_image_path="/x.png", prompt_context=ctx, ) assert seen.get("prompt_context") == ctx assert seen.get("fp_image_path") == "/x.png" # ── shot-context assembly (Required 1) ──────────────────────────────── _SHOT_STAGING = {"shots": [ {"scene_index": 5, "shot_index": 1, "camera_direction": "low angle from door", "key_bg_elements": ["sink", "table"], "framing_scale": "wide", "lighting_mood": "dim", "subject_action": "entering", "shot_type": "establishing", "frame_spatial_contract": "two-room"}, {"scene_index": 5, "shot_index": 7, "camera_direction": "close on hands", "key_bg_elements": ["table"], "framing_scale": "close", "shot_type": "insert"}, ]} _SCENE_SAVE = {"segments": [ {"scene_index": 5, "heading": "주방 내부", "text": "수리영이 좁은 주방으로 들어선다."}, {"scene_index": 9, "heading": "마트", "text": "마트 진열대 사이."}, ]} class TestBuildShotContext: def test_assembles_shot_intent_and_scene_for_shot(self): from app.modules.pipeline.shot_projection_card import build_shot_context ctx = build_shot_context( shot_id="S5_Shot1", shot_staging=_SHOT_STAGING, scene_save=_SCENE_SAVE, ) assert ctx["shot_intent"]["camera_direction"] == "low angle from door" assert ctx["shot_intent"]["framing_scale"] == "wide" assert ctx["scene"]["text"] == "수리영이 좁은 주방으로 들어선다." assert ctx["shot_id"] == "S5_Shot1" def test_different_shot_index_different_intent(self): from app.modules.pipeline.shot_projection_card import build_shot_context a = build_shot_context(shot_id="S5_Shot1", shot_staging=_SHOT_STAGING, scene_save=_SCENE_SAVE) b = build_shot_context(shot_id="S5_Shot7", shot_staging=_SHOT_STAGING, scene_save=_SCENE_SAVE) assert a["shot_intent"] != b["shot_intent"] def test_scene_content_change_changes_context(self): from app.modules.pipeline.shot_projection_card import build_shot_context, compute_source_hashes a = build_shot_context(shot_id="S5_Shot1", shot_staging=_SHOT_STAGING, scene_save=_SCENE_SAVE) changed_scene = {"segments": [ {"scene_index": 5, "heading": "주방 내부", "text": "DIFFERENT TEXT 내용이 바뀜."}, {"scene_index": 9, "heading": "마트", "text": "마트 진열대 사이."}, ]} b = build_shot_context(shot_id="S5_Shot1", shot_staging=_SHOT_STAGING, scene_save=changed_scene) ha = compute_source_hashes(**_hash_inputs(shot_context=a)) hb = compute_source_hashes(**_hash_inputs(shot_context=b)) assert ha["shot_context_hash"] != hb["shot_context_hash"] def test_unknown_shot_id_returns_empty_but_stable(self): from app.modules.pipeline.shot_projection_card import build_shot_context ctx = build_shot_context(shot_id="S99_Shot9", shot_staging=_SHOT_STAGING, scene_save=_SCENE_SAVE) assert ctx["shot_intent"] == {} assert ctx["scene"]["text"] == "" # ── pack/model provenance (Required 3) ──────────────────────────────── class TestPackResolution: def test_resolve_known_selector(self): from app.modules.pipeline.shot_projection_card import resolve_pack_version assert resolve_pack_version("1") == "1.202605302212" def test_resolve_unknown_selector_falls_back_to_raw(self): from app.modules.pipeline.shot_projection_card import resolve_pack_version # unknown selector is recorded verbatim so a mismatch is visible, # never silently mapped to a wrong pack. assert resolve_pack_version("99") == "99" def test_provider_constants_match_real_call(self): from app.modules.pipeline.shot_projection_card import PROVIDER_MODEL, PROVIDER_NAME assert PROVIDER_MODEL == "openai/gpt-6-astra" assert PROVIDER_NAME == "openai" def test_source_hashes_records_pack_version(self): a = compute_source_hashes(**_hash_inputs(pack_version="1.202605302212")) assert a["pack_version"] == "1.202605302212" def test_pack_version_change_changes_card_id(self): h1 = compute_source_hashes(**_hash_inputs(pack_version="1.202605302212")) h2 = compute_source_hashes(**_hash_inputs(pack_version="2.999")) k1 = compute_card_cache_key(source_hashes=h1, bg_id="L05B09", shot_id="S5_Shot1") k2 = compute_card_cache_key(source_hashes=h2, bg_id="L05B09", shot_id="S5_Shot1") assert k1 != k2 # ── Required A: validate_vlm_output full echo + type mirror ──────────── class TestValidateVlmOutputStrict: def test_relabelled_marker_layer_blocked(self): # transient #18 echoed as base must fail (base/transient separation) inv = _INVENTORY # 3,7 base ; 18 transient out = _good_vlm_output(visible_items=[{ "marker_number": 18, "marker_layer": "base", "expected_label": "cartons", "visibility": "visible", "horizontal_band": "left", "depth_band": "foreground", "occlusion_note": "", "evidence": "x", "source_ref": "fp", "confidence": 0.8, }]) res = validate_vlm_output(output=out, inventory=inv, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1") assert res["ok"] is False def test_correct_layer_echo_ok(self): out = _good_vlm_output(visible_items=[{ "marker_number": 18, "marker_layer": "transient", "expected_label": "cartons", "visibility": "visible", "horizontal_band": "left", "depth_band": "foreground", "occlusion_note": "", "evidence": "x", "source_ref": "fp", "confidence": 0.8, }]) res = validate_vlm_output(output=out, inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1") assert res["ok"] is True def test_bad_visible_text_field_type_blocked(self): out = _good_vlm_output() out["visible_items"][0]["occlusion_note"] = 123 # not a string res = validate_vlm_output(output=out, inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1") assert res["ok"] is False def test_bad_top_level_confidence_blocked(self): out = _good_vlm_output(confidence="high") # not number|null res = validate_vlm_output(output=out, inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1") assert res["ok"] is False def test_bad_missing_inputs_type_blocked(self): out = _good_vlm_output(missing_inputs="overlay") # not a list res = validate_vlm_output(output=out, inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1") assert res["ok"] is False def test_bad_diagnostics_type_blocked(self): out = _good_vlm_output(diagnostics=[123]) # list of non-str res = validate_vlm_output(output=out, inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1") assert res["ok"] is False def test_bad_self_consistency_notes_blocked(self): out = _good_vlm_output(self_consistency={"prose_matches_structured": "consistent", "notes": 5}) res = validate_vlm_output(output=out, inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1") assert res["ok"] is False # ── Required B: dossier enforced in source_hash ─────────────────────── class TestDossierInSourceHash: def test_source_hashes_records_dossier_hash(self): a = compute_source_hashes(**_hash_inputs(dossier={"fp_id": "x", "base_marker_inventory": [1]})) assert "dossier_hash" in a assert a["dossier_hash"] != "not_available" def test_dossier_change_changes_card_id(self): h1 = compute_source_hashes(**_hash_inputs(dossier={"v": 1})) h2 = compute_source_hashes(**_hash_inputs(dossier={"v": 2})) k1 = compute_card_cache_key(source_hashes=h1, bg_id="L05B09", shot_id="S5_Shot1") k2 = compute_card_cache_key(source_hashes=h2, bg_id="L05B09", shot_id="S5_Shot1") assert k1 != k2 class TestValidateVlmOutputRequiredPresence: def test_missing_top_level_confidence_blocked(self): out = _good_vlm_output() del out["confidence"] res = validate_vlm_output(output=out, inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1") assert res["ok"] is False def test_explicit_null_confidence_ok(self): out = _good_vlm_output(confidence=None) res = validate_vlm_output(output=out, inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1") assert res["ok"] is True def test_missing_top_level_required_key_blocked(self): out = _good_vlm_output() del out["vlm_reported_state"] res = validate_vlm_output(output=out, inventory=_INVENTORY, fp_id="fp_l05_main", bg_id="L05B09", shot_id="S5_Shot1") assert res["ok"] is False