"""SceneContextLoader._load_dependencies — Area D-next-min + C7 L3 fail-fast. scene_checkpoint_loaders.py 가 L2 검증을 수행하지만, scene_context_loader 는 raw shot_dependency_t2i cp 를 그대로 detail_steps 의 forward_zoom_targets + keep_summary 로 흘려보내고 있었다. legacy v5 cp (schema_version=1 또는 missing, keep_elements List[str]) / legacy v6 cp (schema_version=2, keep_elements 에 kind=immobilized_character 포함) / legacy v7 cp (schema_version=3, subject_kind 부재) 가 이 경로로 진입 시 silent fallback 또는 downstream TypeError 가 났다. 본 테스트는 fail-fast contract 를 강제: schema_version != 4 또는 keep_elements shape 위반 → friendly AppError (force re-run 안내). legacy shot_dependency / scene_dependency fallback path 는 그대로 (옛 step 명 — 본 fix 의 scope 외). C7 (2026-05-20) — schema_version 4, keep_elements entry 는 {label, kind∈ {environment, static_prop}, subject_kind∈{non_human_visual_element, human_or_character_reference}}. v5/v6/v7 cp (schema_version absent/1/2/3) 는 legacy 로 분류 + AppError. """ from __future__ import annotations from typing import Any, Dict, Optional import pytest from app.core.errors import AppError from app.core.steps.scene_context_loader import SceneContextLoader class _FakeRunner: project_id = "p" episode_id = "e" db = None project_config = None def __init__( self, cps: Dict[str, Any], step_runs: Optional[Dict[str, Dict[str, Any]]] = None, ) -> None: self._cps = cps self._runs = step_runs or {} def _load_prev_checkpoint(self, step_id: str): return self._cps.get(step_id) def _get_step_run(self, step_id: str): return self._runs.get(step_id) def build_opik_metadata(self): return {} def _make_loader( cps: Dict[str, Any], step_runs: Optional[Dict[str, Dict[str, Any]]] = None, ) -> SceneContextLoader: return SceneContextLoader(_FakeRunner(cps, step_runs)) # ────────────────────────────────────────────────────────────────────── # Happy path — v8 valid cp (schema_version=4, kind 2종 + subject_kind). # ────────────────────────────────────────────────────────────────────── def test_v8_valid_cp_returns_dependencies() -> None: loader = _make_loader({ "shot_dependency_t2i": { "schema_version": 4, "data": {"dependencies": [{ "scene_index": 1, "shot_index": 2, "location_refs": [{ "ref_usage": "zoom_in_detail", "ignore_elements": "", "keep_elements": [ {"label": "wooden bench against the wall", "kind": "environment", "subject_kind": "non_human_visual_element"}, {"label": "broken vase scattered on the floor", "kind": "static_prop", "subject_kind": "non_human_visual_element"}, ], }], }]}, }, }) deps = loader._load_dependencies() assert len(deps) == 1 assert deps[0]["scene_index"] == 1 def test_v8_valid_empty_keep_elements_passes() -> None: loader = _make_loader({ "shot_dependency_t2i": { "schema_version": 4, "data": {"dependencies": [{ "scene_index": 3, "shot_index": 1, "location_refs": [{ "ref_usage": "atmosphere_reference", "ignore_elements": "Do not copy background layout.", "keep_elements": [], }], }]}, }, }) deps = loader._load_dependencies() assert deps[0]["scene_index"] == 3 # ────────────────────────────────────────────────────────────────────── # Legacy v5 / v6 / v7 cp — schema_version missing / 1 / 2 / 3. # ────────────────────────────────────────────────────────────────────── def test_v5_cp_missing_schema_version_raises() -> None: loader = _make_loader({ "shot_dependency_t2i": { # schema_version absent (legacy v5 — written before Area D-next bump) "data": {"dependencies": [{ "scene_index": 1, "shot_index": 1, "location_refs": [{"keep_elements": []}], }]}, }, }) with pytest.raises(AppError) as exc_info: loader._load_dependencies() assert exc_info.value.code == "step.scene_context_loader.legacy_keep_elements_cp" def test_v5_cp_schema_version_1_raises() -> None: loader = _make_loader({ "shot_dependency_t2i": { "schema_version": 1, "data": {"dependencies": [{ "scene_index": 1, "shot_index": 1, "location_refs": [{"keep_elements": []}], }]}, }, }) with pytest.raises(AppError) as exc_info: loader._load_dependencies() assert exc_info.value.code == "step.scene_context_loader.legacy_keep_elements_cp" def test_v6_cp_schema_version_2_raises_as_legacy() -> None: """Area D-next-min (신규) — v6 cp (schema_version=2) 도 legacy. v7 bump 후 operator force re-run 의무. 본 fail-fast 가 immobilized_character entry 가 downstream forward_zoom_targets 에 silently 유입 차단.""" loader = _make_loader({ "shot_dependency_t2i": { "schema_version": 2, # Area D-next v6 (legacy after Area D-next-min) "data": {"dependencies": [{ "scene_index": 1, "shot_index": 1, "location_refs": [{ "keep_elements": [ {"label": "wooden bench", "kind": "environment"}, ], }], }]}, }, }) with pytest.raises(AppError) as exc_info: loader._load_dependencies() assert exc_info.value.code == "step.scene_context_loader.legacy_keep_elements_cp" def test_v7_cp_schema_version_3_raises_as_legacy() -> None: """C7 (신규) — v7 cp (schema_version=3, subject_kind 부재) 도 legacy. v8 bump 후 accepted = 4 로 전환되었으므로 v7 cp 는 force re-run 의무. 본 fail-fast 가 subject_kind 없는 keep_elements entry 가 downstream forward_zoom_targets 에 silently 유입되는 것을 차단 (accepted 3→4 전환 증명).""" loader = _make_loader({ "shot_dependency_t2i": { "schema_version": 3, # C7 v8 bump 후 legacy "data": {"dependencies": [{ "scene_index": 1, "shot_index": 1, "location_refs": [{ "keep_elements": [ {"label": "wooden bench", "kind": "environment"}, ], }], }]}, }, }) with pytest.raises(AppError) as exc_info: loader._load_dependencies() assert exc_info.value.code == "step.scene_context_loader.legacy_keep_elements_cp" # ────────────────────────────────────────────────────────────────────── # Shape validation — keep_elements legacy str / missing key / invalid kind. # ────────────────────────────────────────────────────────────────────── def test_v8_cp_legacy_string_keep_elements_raises() -> None: loader = _make_loader({ "shot_dependency_t2i": { "schema_version": 4, "data": {"dependencies": [{ "scene_index": 1, "shot_index": 1, "location_refs": [{ "ref_usage": "exact_background", "keep_elements": ["wooden bench", "broken vase"], }], }]}, }, }) with pytest.raises(AppError) as exc_info: loader._load_dependencies() assert exc_info.value.code == ( "step.scene_context_loader.keep_elements_legacy_str" ) def test_v8_cp_missing_keep_elements_key_raises() -> None: loader = _make_loader({ "shot_dependency_t2i": { "schema_version": 4, "data": {"dependencies": [{ "scene_index": 1, "shot_index": 1, "location_refs": [{"ref_usage": "exact_background"}], }]}, }, }) with pytest.raises(AppError) as exc_info: loader._load_dependencies() assert exc_info.value.code == ( "step.scene_context_loader.keep_elements_missing_key" ) def test_v8_cp_invalid_kind_enum_raises() -> None: loader = _make_loader({ "shot_dependency_t2i": { "schema_version": 4, "data": {"dependencies": [{ "scene_index": 1, "shot_index": 1, "location_refs": [{ "keep_elements": [ {"label": "x", "kind": "unknown_kind", "subject_kind": "non_human_visual_element"}, ], }], }]}, }, }) with pytest.raises(AppError) as exc_info: loader._load_dependencies() assert exc_info.value.code == ( "step.scene_context_loader.keep_elements_entry_invalid" ) def test_v8_cp_legacy_immobilized_character_kind_raises() -> None: """Area D-next-min (신규) — schema_version=4 envelope 안이라도 kind= immobilized_character entry 가 들어오면 enum 2종 외라 fail-fast. (v8 producer 가 정상 emit 했다면 발생 X, 그러나 manual edit / drift 검출).""" loader = _make_loader({ "shot_dependency_t2i": { "schema_version": 4, "data": {"dependencies": [{ "scene_index": 1, "shot_index": 1, "location_refs": [{ "keep_elements": [ {"label": "the dead detective", "kind": "immobilized_character", "subject_kind": "non_human_visual_element"}, ], }], }]}, }, }) with pytest.raises(AppError) as exc_info: loader._load_dependencies() assert exc_info.value.code == ( "step.scene_context_loader.keep_elements_entry_invalid" ) assert "immobilized_character" in str(exc_info.value.message) def test_v8_cp_missing_label_field_raises() -> None: loader = _make_loader({ "shot_dependency_t2i": { "schema_version": 4, "data": {"dependencies": [{ "scene_index": 1, "shot_index": 1, "location_refs": [{ "keep_elements": [ {"kind": "environment", "subject_kind": "non_human_visual_element"}, ], }], }]}, }, }) with pytest.raises(AppError) as exc_info: loader._load_dependencies() assert exc_info.value.code == ( "step.scene_context_loader.keep_elements_entry_invalid" ) # ────────────────────────────────────────────────────────────────────── # Legacy fallback path — shot_dependency_t2i 미존재. # ────────────────────────────────────────────────────────────────────── def test_t2i_cp_absent_falls_back_to_shot_dependency() -> None: """shot_dependency_t2i 미존재 → 옛 shot_dependency 사용. 본 fix scope 외.""" loader = _make_loader({ "shot_dependency": { "data": {"dependencies": [{ "scene_index": 1, "shot_index": 1, "location_refs": [{"keep_elements": []}], }]}, }, }) deps = loader._load_dependencies() assert len(deps) == 1 def test_t2i_cp_empty_dependencies_falls_back() -> None: """t2i cp 존재 + dependencies 비어있음 → legacy fallback (현재 동작 유지).""" loader = _make_loader({ "shot_dependency_t2i": { "schema_version": 4, "data": {"dependencies": []}, }, "shot_dependency": { "data": {"dependencies": [{ "scene_index": 2, "shot_index": 1, "location_refs": [], }]}, }, }) deps = loader._load_dependencies() assert len(deps) == 1 assert deps[0]["scene_index"] == 2 def test_no_cp_at_all_returns_empty() -> None: loader = _make_loader({}) assert loader._load_dependencies() == []