"""G4.1 Phase 8 Task 20 — RenderPromptCard consumer wiring integration tests. 본 테스트는 G4.1 Phase 1~7 의 producer→inject→verify→reuse 흐름이 한 시나리오 안에서 정확히 wiring 되어 있는지 검증한다. 단위 테스트 (test_render_prompt_card*) 가 각 helper 를 검증하는 것과 달리, integration 테스트는 다음 9 시나리오를 end-to-end (mocked LLM/DB) 로 cover: 1. TestRenderPromptCardInjection — `_analyze_one()` user_prompt 안에 정확히 1 개 `[RenderPromptCard v1]` block + canonical JSON. 2. TestCardCpFieldPreservation — result top-level `render_prompt_card` + `render_prompt_card_hash` deep copy + cp round-trip 보존. 3. TestCardWinsOverLegacyConflict (R2-I3) — card 와 legacy block 이 충돌하는 fixture → user_prompt 자체에 "card wins" / "primary contract" 명시 검증. 4. TestT2iReviewPostEditSentinelStale (R2-I4 / R3-I1) — verify_completion 직접 호출 → severity="partial" + missing_msgs 안 "sentinel_drifted" literal + metadata.sentinel_drifted 안 (si, shi, "t2i_prompt_hash") tuple. 5. TestForwardZoomFullListIntegration (R1-I7 / R2-B4) — forward_zoom_targets count > 6 + 각 keep_elements count > 5 → card payload + hash 모든 entry 보존. 6. TestPreviousShotRefsVariants (R1-I8) — `zoom_in_detail` + `continuation` ref_usage 분기 모두 카드 안 적용. 7. TestAssetRequirementsBranches (R1-I8 + R3-B2) — 3 분기 명확 분리: (a) required-only / (b) forbidden-only-pure (no outlook) / (c) mixed. R3-B2 sentinel pattern 으로 `outlook_pairs=[]` 명시 보존 진단. 8. TestUserEditedCardDrift (R2-I6 / R3-I2) — `_user_edited_card_contract_ violated()` helper unit + `SceneDetailStep._execute()` 직접 호출 + fresh path 진입 검증. 9. TestV15ToV16MigrationSmoke (R1-B1 / R2-I5 / R3-B1) — completed v15 cp fixture (schema_version=6, render_prompt_card 부재) → `SceneDetailStep. run("resume")` 직접 호출 → step_run mode="force" transition + cp 가 v16 로 갱신되어 card 둘 다 포함. LLM 호출 (`call_structured`) 및 SceneContextLoader DB 호출은 모두 mock — 실제 API/DB 미사용. """ from __future__ import annotations import copy import json from pathlib import Path from typing import Any, Dict, List, Optional from unittest.mock import MagicMock, patch import pytest from app.core.errors import AppError from app.core.integrity_report import CompletionReport from app.core.steps.detail_steps import ( SCENE_DETAIL_SCHEMA_VERSION, SceneDetailStep, _user_edited_card_contract_violated, ) from app.core.steps.render_prompt_card import ( CARD_SCHEMA_VERSION, assert_card_shape, build_render_prompt_card, compute_card_hash, ) # ────────────────────────────────────────────────────────────────────────── # Fixture helpers # ────────────────────────────────────────────────────────────────────────── # R3-B2: sentinel 기반 None vs [] 명시 구분. # `outlook_pairs or default` / `fixed_elements or []` 패턴 제거 — caller 가 # 명시적으로 [] 를 넘기면 빈 list 그대로 보존 (forbidden-only-pure fixture # 진정성 보장). None 만 default 로 치환 (= sentinel 미전달 case). _DEFAULT_SENTINEL = object() def _make_ctx_for_shot( *, close_framing: bool = False, bg_on: bool = True, bg_id: str = "cb_main_room", outlook_pairs=_DEFAULT_SENTINEL, fixed_elements=_DEFAULT_SENTINEL, ) -> Dict[str, Any]: """build_render_prompt_card 의 input ctx — test fixture. R3-B2: explicit None vs sentinel. - `outlook_pairs=[]` (caller 가 빈 list 명시) → 빈 list 그대로 보존. - `outlook_pairs=None` → 명시적 None (builder 가 raise — fail-fast). - `outlook_pairs=_DEFAULT_SENTINEL` (= 인자 미전달) → default 채움. `default if x is None else x` 같은 silent 변환 패턴 금지. """ if outlook_pairs is _DEFAULT_SENTINEL: outlook_pairs = [{"character_id": "C01", "outlook_id": "O02"}] if fixed_elements is _DEFAULT_SENTINEL: fixed_elements = [] return { "scene_index": 12, "shot_index": 4, "seg": {"index": 12, "text": "scene text full content here"}, "shot_info": { "shot_index": 4, "camera_direction": ( "extreme close-up of hand" if close_framing else "medium shot of doorway" ), "primary_subject": "the observer at the doorway", }, "visible_entities": ["C01"], "outlook_pairs": outlook_pairs, # R3-B2: explicit, no `or default` "perception_mode": None, "staging": { "camera_direction": ( "extreme close-up of hand" if close_framing else "medium shot of doorway" ), # framing_scale enum SOT v1 (2026-05-15): helper read fail-fast 정합. "framing_scale": "close" if close_framing else "medium", "lighting_mood": "warm dim", # Area C (2026-05-12) — required by build_id_policy. 빈 list = # 비-재현면 shot (applies=False). "key_bg_elements": [], # Area #1 W5 (2026-05-16) — shot_staging v12 top-level required # field. helper SOT graceful empty (default policy # id_and_outlook_required 가 visible_entities 에 적용). "subject_reference_policy": [], }, "bg_id": bg_id if bg_on else None, "bg_owned": ["door", "window"] if bg_on else [], "bg_camera_meta": ( {"camera_position": "southeast doorway"} if bg_on else None ), "bg_guide": "doorway view" if bg_on else None, "is_close_framing": close_framing, "background_mode_on": bg_on, "fixed_elements": fixed_elements, # R3-B2: explicit, no `or []` "previous_shot_refs": [], "forward_zoom_targets": [], } def _make_scene_detail_step_skeleton( tmp_path, monkeypatch, *, bg_mode: str = "off", ): """SceneDetailStep — DB / cp 의존성 우회한 minimal skeleton. `verify_completion` test 처럼 mixin/loader 호출이 필요한 케이스에서만 사용. SceneContextLoader 가 self.db 를 만지므로 db 도 MagicMock 으로 stub. """ monkeypatch.setattr( "app.core.config.settings.projects_dir", str(tmp_path), raising=False, ) monkeypatch.setattr( "app.core.config.settings.background_mode", bg_mode, raising=False, ) step = SceneDetailStep.__new__(SceneDetailStep) step.project_id = "P1" step.episode_id = "E1" step.project_config = {} step.step_id = "scene_detail" step.run_id = "run-test" step.opik_context = {} step.db = MagicMock() step.db.execute.return_value.fetchall.return_value = [] step.db.execute.return_value.fetchone.return_value = None # StepRunner.__init__ 가 만들어주는 attribute 들 — skeleton 에선 직접 set. from app.core.step_manifest import get_manifest_dict step.manifest = get_manifest_dict("scene_detail") step._cp_dir = ( Path(tmp_path) / "P1" / "checkpoints" / "episodes" / "E1" / "scene_detail" ) return step def _build_card_block(card: Dict[str, Any]) -> str: """`_analyze_one()` 의 prefix block 과 동일 canonical JSON inject.""" return ( "[RenderPromptCard v1]\n" + json.dumps( card, sort_keys=True, ensure_ascii=False, separators=(",", ":") ) + "\n\n" ) # ────────────────────────────────────────────────────────────────────────── # Class 1: TestRenderPromptCardInjection # ────────────────────────────────────────────────────────────────────────── class TestRenderPromptCardInjection: """`_analyze_one()` user_prompt prefix = `[RenderPromptCard v1]` block. Plan body Task 20 step body §1+§2 — 정확히 1개 block + first prefix + canonical JSON parseable. """ def test_user_prompt_has_exactly_one_card_block(self) -> None: # spec §7.2: scene_detail user_prompt 는 RenderPromptCard 블록 1 회만. ctx = _make_ctx_for_shot() card = build_render_prompt_card(**ctx) block = _build_card_block(card) legacy = "[Minimal narrative input]\nscene summary: foo\n" user_prompt = block + legacy assert user_prompt.count("[RenderPromptCard v1]") == 1 # 첫 블록 위치 — prefix. assert user_prompt.startswith("[RenderPromptCard v1]") def test_card_json_parseable_and_valid(self) -> None: ctx = _make_ctx_for_shot() card = build_render_prompt_card(**ctx) block_json = json.dumps( card, sort_keys=True, ensure_ascii=False, separators=(",", ":"), ) parsed = json.loads(block_json) # Round-trip 후에도 shape 통과. assert_card_shape(parsed) # G4.2: card now carries _card_metadata envelope-sibling # (lift_status + rule_source). Hash-excluded + free-form (R2-I5). # Inject path strips it (detail_steps.py:1942-1945) but the card # itself round-trips with all keys. # Area B (2026-05-13): render_contracts top-level field 추가 # (visible prop ∩ visual_identity.reference_required=true). assert set(parsed.keys()) == { "schema_version", "shot_key", "render_strategy", "id_policy", "background_binding", "continuity_elements_used", "asset_requirements", "render_contracts", "_card_metadata", } # ────────────────────────────────────────────────────────────────────────── # Class 2: TestCardCpFieldPreservation (R1-I6) # ────────────────────────────────────────────────────────────────────────── class TestCardCpFieldPreservation: """R1-I6: card 는 G4.1 에서 CP-only debug 필드. scene_still_normalizer preservation 요구는 본 G4.1 scope 밖 (DB persistence 는 G4.6+ / G5). in-memory deep copy + checkpoint round-trip JSON 보존만 검증. """ def test_in_memory_deep_copy_preserves_card(self) -> None: ctx = _make_ctx_for_shot() card = build_render_prompt_card(**ctx) cp_result = { "schema_version": SCENE_DETAIL_SCHEMA_VERSION, "prompt_version": "16.202605041200", "render_prompt_card": card, "render_prompt_card_hash": compute_card_hash(card), "t2i_variations": [{ "t2i_prompt": "A figure stands by the door.", "owned_validation": { # G3.2 sentinel coexist "schema_version": 2, "validator": "scene_detail_owned_objects.v1", "owned_hash": "x" * 16, "camera_direction_hash": "y" * 16, "t2i_prompt_hash": "z" * 16, "owned_usage_hash": "0123456789abcdef", # C2 v1 sentinel v2 "violations": [], }, }], } copied = copy.deepcopy(cp_result) # card / hash 보존. assert copied["render_prompt_card"] == card assert copied["render_prompt_card_hash"] == cp_result["render_prompt_card_hash"] # G3.2 sentinel 보존 — card 가 sentinel 대체하지 않음 (분리된 scope). assert ( copied["t2i_variations"][0]["owned_validation"]["validator"] == "scene_detail_owned_objects.v1" ) def test_checkpoint_json_roundtrip_preserves_card(self, tmp_path) -> None: # cp 가 disk 에 저장되었다 다시 로드된 경우에도 card 보존 (in-memory # 외 추가 일관성 가드). ctx = _make_ctx_for_shot() card = build_render_prompt_card(**ctx) cp_payload = { "schema_version": SCENE_DETAIL_SCHEMA_VERSION, "render_prompt_card": card, "render_prompt_card_hash": compute_card_hash(card), } cp_path = tmp_path / "manifest.json" cp_path.write_text(json.dumps(cp_payload), encoding="utf-8") loaded = json.loads(cp_path.read_text(encoding="utf-8")) assert loaded["render_prompt_card"] == card assert loaded["render_prompt_card_hash"] == cp_payload[ "render_prompt_card_hash" ] # JSON round-trip 후 shape 도 무사. assert_card_shape(loaded["render_prompt_card"]) # ────────────────────────────────────────────────────────────────────────── # Class 3: TestCardWinsOverLegacyConflict (R2-I3) # ────────────────────────────────────────────────────────────────────────── class TestCardWinsOverLegacyConflict: """R2-I3 (R1-I4 cross-check): card field 와 legacy prompt block 이 충돌 → card 가 winner. 구현 단위 (LLM 호출 전) 에서는 prompt 자체에 "card wins" 명시 grep + user_prompt 조립 순서 검증. """ def test_v16_system_md_declares_card_wins(self) -> None: # R1-I4 + R2-I3: v16 system.md 의 priority section 이 명시적으로 # "card 가 winner" 선언. legacy compat blocks 보다 우선. sys_md = Path( "/Users/manta/Documents/Projects/TheRoad-I1/prompts/_base/" "scene_detail/16.202605041200/system.md" ) text = sys_md.read_text(encoding="utf-8") joined_lower = text.lower() # 어느 한 표현 이상은 들어 있어야 contract 강화로 인정. needles = ( "card wins", "primary contract", "card winner", "card always wins", "card is winner", "card 가 winner", "card 가 항상 winner", ) assert any(n in joined_lower for n in needles), ( f"v16 system.md 에 card-wins precedence 명시 필요 — " f"found none of {needles}" ) def test_card_id_policy_overrides_legacy_block_in_user_prompt(self) -> None: # card.id_policy 가 C##O## 강제하는데 legacy block 이 보통명사 권장 # 한다고 가정. user_prompt 조립 시 card 가 prefix → LLM 이 card 우선. # 본 test 는 prompt 조립 순서 + card 의 explicit composite ID 강제만 # 검증 (LLM output 검증은 canary scope). ctx = _make_ctx_for_shot() card = build_render_prompt_card(**ctx) card_block = _build_card_block(card) legacy_conflict_block = ( "[Legacy compat]\n" "Use common nouns for visible characters.\n" ) user_prompt = card_block + legacy_conflict_block # card 가 prefix. assert user_prompt.index("[RenderPromptCard v1]") < user_prompt.index( "[Legacy compat]" ) # card 의 must_use_composite_character_ids = True 가 explicit 으로 # 들어 있음 → LLM 이 card winner rule 따라 composite 사용. assert '"must_use_composite_character_ids":true' in user_prompt # ────────────────────────────────────────────────────────────────────────── # Class 4: TestT2iReviewPostEditSentinelStale (R1-I5 / R2-I4 / R3-I1) # ────────────────────────────────────────────────────────────────────────── class TestT2iReviewPostEditSentinelStale: """R1-I5 + R2-I4 + R3-I1: t2i_review post-edit 후 owned_validation 의 t2i_prompt_hash 가 stale → `verify_completion()` 직접 호출 → severity= "partial" + missing_msgs 안 "sentinel_drifted" literal + metadata. sentinel_drifted 안 (si, shi, "t2i_prompt_hash") tuple. R3-I1 강화: substring → exact literal + metadata tuple + severity literal. card hash 자체는 shot-level 이라 t2i_prompt 수정과 무관 — 재계산 X. judge 재호출 비추천 (G3.2 결정). """ def test_card_hash_unchanged_when_only_t2i_prompt_edited(self) -> None: # G3.2 sentinel 의 t2i_prompt_hash 는 t2i_prompt 와 binding — # t2i_review in-place 수정 시 stale 된다. # card hash 는 shot-level (variation-level prompt 와 무관) — 변하지 않음. from app.core.steps._owned_helpers import compute_t2i_prompt_hash original_prompt = "A figure stands by the door." edited_prompt = "A figure stands by the doorway, holding a key." original_hash = compute_t2i_prompt_hash(original_prompt) edited_hash = compute_t2i_prompt_hash(edited_prompt) assert original_hash != edited_hash ctx = _make_ctx_for_shot() card = build_render_prompt_card(**ctx) h_before = compute_card_hash(card) # t2i_review 가 t2i_prompt 만 수정 — card 자체는 그대로. h_after = compute_card_hash(card) assert h_before == h_after # card hash 무관성 def test_verify_completion_marks_partial_on_sentinel_drift( self, tmp_path, monkeypatch, ) -> None: # R2-I4 + R3-I1: hash 비교만 하던 R1-I5 test 강화 → # SceneDetailStep.verify_completion() 직접 호출 + literal severity # ("partial") + literal missing message ("sentinel_drifted") + # metadata tuple ((si, shi, "t2i_prompt_hash")) assert. from app.core.steps._owned_helpers import compute_t2i_prompt_hash step = _make_scene_detail_step_skeleton( tmp_path, monkeypatch, bg_mode="off", ) # verify_completion 의 card recompute path 가 outlook_pairs=[] 로 # 고정 호출 — fixture card 도 동일하게 빌드해야 sentinel-only 분리. # bg_on=False → bg-mode-off 분기 = card recompute 와 일치. ctx = _make_ctx_for_shot(bg_on=False, outlook_pairs=[]) card = build_render_prompt_card(**ctx) edited_prompt = "A figure stands by the doorway, holding a key." # sentinel 의 hash 는 *원본* prompt 의 hash 로 frozen → stale. stale_sentinel_hash = compute_t2i_prompt_hash( "A figure stands by the door." ) # close framing = False, owned=[] (bg-off path 와 일치) → close-skip # validator 가 아닌 FULL validator 로 계산. from app.core.steps._owned_helpers import ( OWNED_VALIDATOR_FULL, compute_camera_direction_hash, compute_owned_hash, ) # camera_direction 은 verify_completion 의 staging_map 에서 가져옴. # SceneContextLoader._load_staging_map 는 빈 cp 면 {} → cam_dir="". cp_result = { "schema_version": SCENE_DETAIL_SCHEMA_VERSION, "prompt_version": "16.202605041200", "data": { "scenes": [{ "scene_index": 12, "_shot_index": 4, "visible_entities": [], # match recompute input "render_prompt_card": card, "render_prompt_card_hash": compute_card_hash(card), "t2i_variations": [{ "t2i_prompt": edited_prompt, "owned_validation": { "schema_version": 2, "validator": OWNED_VALIDATOR_FULL, "owned_hash": compute_owned_hash([]), "camera_direction_hash": ( compute_camera_direction_hash("") ), "t2i_prompt_hash": stale_sentinel_hash, # stale "owned_usage_hash": "0123456789abcdef", # C2 v1 sentinel v2 "violations": [], }, }], }], }, } step._last_execute_result = cp_result report = step.verify_completion() # R3-I1: literal severity + literal missing message + metadata tuple. assert report.is_complete is False, ( f"R3-I1: report.is_complete must be False (got " f"{report.is_complete!r})" ) assert report.severity == "partial", ( f"R3-I1: report.severity must be literal 'partial' " f"(got {report.severity!r})" ) # R3-I1 + I4: missing_msgs 안 production literal "sentinel_drifted" 포함. # (실제 production 메세지: "N owned_validation sentinel_drifted: ...") assert any("sentinel_drifted" in m for m in report.missing), ( f"R3-I1 / I4: report.missing must include 'sentinel_drifted' literal " f"(missing={list(report.missing)!r})" ) # R3-I1: metadata 의 sentinel_drifted list literal tuple 포함. sentinel_drifted = (report.metadata or {}).get( "sentinel_drifted" ) or [] assert (12, 4, "t2i_prompt_hash") in [ tuple(s) for s in sentinel_drifted ], ( f"R3-I1: report.metadata.sentinel_drifted must include " f"(12, 4, 't2i_prompt_hash') tuple (got {sentinel_drifted!r})" ) # ────────────────────────────────────────────────────────────────────────── # Class 5: TestForwardZoomFullListIntegration (R1-I7 / R2-B4) # ────────────────────────────────────────────────────────────────────────── class TestForwardZoomFullListIntegration: """R1-I7: forward_zoom_targets count > 6 + 각 keep_elements count > 5 → card payload + card hash 가 모든 entry 보존. legacy [:6]/[:5] cap 의 영향 받지 않음 (G4.4 lift 시 처리 — non-goal in G4.1). """ def test_forward_zoom_full_list_preserved_in_card(self) -> None: ctx = _make_ctx_for_shot() ctx["forward_zoom_targets"] = [ { "scene_index": 12, "shot_index": 5 + i, "description": f"forward zoom #{i}", "keep_elements": [{"label": f"k_{j}", "kind": "environment"} for j in range(7)], # > 5 } for i in range(8) # > 6 ] c = build_render_prompt_card(**ctx) assert len(c["continuity_elements_used"]["forward_zoom_targets"]) == 8 for entry in c["continuity_elements_used"]["forward_zoom_targets"]: assert len(entry["keep_elements"]) == 7 def test_forward_zoom_count_change_drifts_hash(self) -> None: # full list 보존 → entry 수 변화 시 hash drift 검출. ctx1 = _make_ctx_for_shot() ctx1["forward_zoom_targets"] = [ {"scene_index": 12, "shot_index": 5 + i, "description": f"f#{i}", "keep_elements": []} for i in range(7) ] ctx2 = _make_ctx_for_shot() ctx2["forward_zoom_targets"] = [ {"scene_index": 12, "shot_index": 5 + i, "description": f"f#{i}", "keep_elements": []} for i in range(8) ] c1 = build_render_prompt_card(**ctx1) c2 = build_render_prompt_card(**ctx2) assert compute_card_hash(c1) != compute_card_hash(c2) # ────────────────────────────────────────────────────────────────────────── # Class 6: TestPreviousShotRefsVariants (R1-I8) # ────────────────────────────────────────────────────────────────────────── class TestPreviousShotRefsVariants: """R1-I8: previous_shot_refs 다양 case (zoom_in_detail / continuation). 각각 card 에 보존 + ref_usage 변경 시 hash drift. """ def test_zoom_in_detail_preserved(self) -> None: ctx = _make_ctx_for_shot() ctx["previous_shot_refs"] = [{ "scene_index": 12, "shot_index": 3, "ref_usage": "zoom_in_detail", "keep_elements": [{"label": "lamp", "kind": "environment"}], }] c = build_render_prompt_card(**ctx) psr = c["continuity_elements_used"]["previous_shot_refs"] assert psr[0]["ref_usage"] == "zoom_in_detail" def test_continuation_preserved(self) -> None: ctx = _make_ctx_for_shot() ctx["previous_shot_refs"] = [{ "scene_index": 12, "shot_index": 3, "ref_usage": "continuation", "keep_elements": [{"label": "chair", "kind": "static_prop"}], }] c = build_render_prompt_card(**ctx) psr = c["continuity_elements_used"]["previous_shot_refs"] assert psr[0]["ref_usage"] == "continuation" def test_ref_usage_change_drifts_hash(self) -> None: ctx1 = _make_ctx_for_shot() ctx2 = _make_ctx_for_shot() ctx1["previous_shot_refs"] = [{ "scene_index": 12, "shot_index": 3, "ref_usage": "zoom_in_detail", "keep_elements": [], }] ctx2["previous_shot_refs"] = [{ "scene_index": 12, "shot_index": 3, "ref_usage": "continuation", "keep_elements": [], }] c1 = build_render_prompt_card(**ctx1) c2 = build_render_prompt_card(**ctx2) assert compute_card_hash(c1) != compute_card_hash(c2) # ────────────────────────────────────────────────────────────────────────── # Class 7: TestAssetRequirementsBranches (R2-I7 + R3-B2) # ────────────────────────────────────────────────────────────────────────── class TestAssetRequirementsBranches: """R2-I7 + R3-B2: asset_requirements 3 분기 분리. (a) required-only — outlook+bg 둘 다 required. forbidden 0. (b) forbidden-only-pure — close framing skip + outlook 없음 → required 0. (c) mixed — close framing + outlook 존재 → required+forbidden 공존. R3-B2: forbidden-only fixture 의 `outlook_pairs=[]` 가 sentinel pattern 으로 빈 list 보존되는지 진단. """ def test_required_only_branch(self) -> None: # (a) required-only: non-close + bg_on + outlook_pairs 존재. ctx = _make_ctx_for_shot(close_framing=False, bg_on=True) c = build_render_prompt_card(**ctx) assert c["asset_requirements"]["readiness_policy"] == "block_if_missing" assert c["asset_requirements"]["forbidden_refs"] == [] kinds = [r["kind"] for r in c["asset_requirements"]["required_refs"]] assert "character_outlook" in kinds assert "background" in kinds # bg required def test_forbidden_only_branch_pure(self) -> None: # (b) forbidden-only: close framing + bg_on + **outlook_pairs 없음**. # 진정한 forbidden-only — required_refs 비어 있어야 함. # R3-B2: `outlook_pairs=[]` 가 sentinel default 로 치환되지 않고 # 빈 list 그대로 보존되는지 확인 (sentinel 패턴 검증). ctx = _make_ctx_for_shot( close_framing=True, bg_on=True, outlook_pairs=[], ) # R3-B2 진단: ctx 가 빈 list 보존했는지. assert ctx["outlook_pairs"] == [], ( "R3-B2 regression — _make_ctx_for_shot(outlook_pairs=[]) 가 " "default 로 치환됨 (sentinel 패턴 깨짐)" ) c = build_render_prompt_card(**ctx) assert c["asset_requirements"]["required_refs"] == [] assert any( f["kind"] == "background" for f in c["asset_requirements"]["forbidden_refs"] ) assert ( c["asset_requirements"]["readiness_policy"] == "skipped_by_policy" ) def test_mixed_branch_required_and_forbidden(self) -> None: # (c) mixed: close framing + bg_on + outlook_pairs 존재. # bg forbidden + outlook required 공존. ctx = _make_ctx_for_shot(close_framing=True, bg_on=True) c = build_render_prompt_card(**ctx) assert any( f["kind"] == "background" for f in c["asset_requirements"]["forbidden_refs"] ) assert any( r["kind"] == "character_outlook" for r in c["asset_requirements"]["required_refs"] ) # bg 는 required 가 아님 (forbidden 으로 갔으므로). assert not any( r["kind"] == "background" for r in c["asset_requirements"]["required_refs"] ) assert ( c["asset_requirements"]["readiness_policy"] == "skipped_by_policy" ) # ────────────────────────────────────────────────────────────────────────── # Class 8: TestUserEditedCardDrift (R2-I6 + R3-I2) # ────────────────────────────────────────────────────────────────────────── class TestUserEditedCardDrift: """R1-I8 + R2-I6 + R3-I2: `_user_edited` reuse path 의 card hash drift. Task 19 helper `_user_edited_card_contract_violated()` unit test + R3-I2 강화: SceneDetailStep._execute() 직접 호출 + fresh path 진입 검증 (`_analyze_one()` 호출 transition + result `_user_edited` False transition). G3.1/G3.2 의 `_user_edited_owned_contract_violated` test shape mirror. """ def test_helper_returns_none_when_card_matches(self) -> None: # R2-I6: helper 가 None 반환 = reuse OK. ctx = _make_ctx_for_shot() card = build_render_prompt_card(**ctx) h = compute_card_hash(card) # helper 는 ctx 에서 builder input 만 splat — `ctx` key 제외. # _make_ctx_for_shot 은 builder signature 와 동일한 dict 라 그대로 사용. violation = _user_edited_card_contract_violated( stored_card=card, stored_hash=h, card_inputs=ctx, ) assert violation is None # reuse OK def test_helper_returns_reason_on_owned_drift(self) -> None: # R2-I6: helper 가 str 반환 = reuse 거부 → fresh. ctx_old = _make_ctx_for_shot() ctx_new = _make_ctx_for_shot() ctx_new["bg_owned"] = ["door", "window", "TV"] # owned 변경 card_old = build_render_prompt_card(**ctx_old) h_old = compute_card_hash(card_old) violation = _user_edited_card_contract_violated( stored_card=card_old, stored_hash=h_old, card_inputs=ctx_new, ) assert violation is not None assert "hash_drift" in violation def test_helper_returns_reason_when_card_missing(self) -> None: # R2-I6: stored_card None → "card_or_hash_missing". ctx = _make_ctx_for_shot() violation = _user_edited_card_contract_violated( stored_card=None, stored_hash=None, card_inputs=ctx, ) assert violation == "card_or_hash_missing" def test_user_edited_branch_invokes_fresh_path_via_execute( self, tmp_path, monkeypatch, ) -> None: # R3-I2 (강화): helper simulation 만 하던 R2-I6 test 변경. 실제 # SceneDetailStep._execute() 직접 호출 + fresh path 진입 검증. # G3.1/G3.2 의 `_user_edited_owned_contract_violated` test shape mirror. # # fixture: _user_edited=True + variation confidence='high' (G3.1 # evidence 통과) + card hash mismatch → reuse 거부 → fresh # `_analyze_one()` 호출. from app.core.steps.scene_context_loader import SceneContextLoader ctx_old = _make_ctx_for_shot() ctx_drifted = _make_ctx_for_shot(outlook_pairs=[ {"character_id": "C01", "outlook_id": "O99"}, # outlook 변경 ]) card_old = build_render_prompt_card(**ctx_old) h_old = compute_card_hash(card_old) # cp fixture: _user_edited=True + G3.1 evidence pass + card stale. v15_user_edited_cp = { "schema_version": SCENE_DETAIL_SCHEMA_VERSION, "prompt_version": "16.202605041200", "data": { "scenes": [{ "scene_index": 12, "_shot_index": 4, "_user_edited": True, # user 가 cp 를 수동 편집 "render_prompt_card": card_old, "render_prompt_card_hash": h_old, "visible_entities": ["C01"], "t2i_variations": [{ "t2i_prompt": "A figure stands by the door.", "confidence": "high", # G3.1 evidence 통과 "source_facts": ["fact1"], "visual_inferences": ["inf1"], "creative_decisions": ["dec1"], "owned_validation": { "schema_version": 2, "validator": "scene_detail_owned_objects.v1", "owned_hash": "x" * 16, "camera_direction_hash": "y" * 16, "t2i_prompt_hash": "z" * 16, "owned_usage_hash": "0123456789abcdef", # C2 v1 sentinel v2 "violations": [], }, }], }], }, } # disk 에 이전 cp 보존 — _load_prev_checkpoint 가 읽을 path. cp_dir = ( Path(tmp_path) / "P1" / "checkpoints" / "episodes" / "E1" / "scene_detail" ) cp_dir.mkdir(parents=True, exist_ok=True) cp_path = cp_dir / "manifest.json" cp_path.write_text(json.dumps(v15_user_edited_cp), encoding="utf-8") step = _make_scene_detail_step_skeleton( tmp_path, monkeypatch, bg_mode="off", ) # ctx (drifted): visible 1, shot, segments 1 entry 모두 minimal. # `SceneContextLoader.load_all` 를 mock 해서 무거운 DB 호출 우회. from app.core.dto.scene_analysis import SceneAnalysisContext mock_ctx = SceneAnalysisContext(project_id="P1", episode_id="E1") mock_ctx.segments = [{"scene_index": 12, "text": "scene 12 text"}] mock_ctx.selected_map = {12: {4}} # set per dataclass annotation mock_ctx.shot_scenes_map = { 12: [{"shot_index": 4, "camera_direction": "medium shot of doorway"}], } mock_ctx.scene_visible = {12: ["C01"]} mock_ctx.summaries = {12: "summary"} mock_ctx.world_rules = {} mock_ctx.planning_context = None mock_ctx.staging_map = { "12_4": {"camera_direction": "medium shot of doorway", "framing_scale": "medium"}, } mock_ctx.chain_bg_owned_by_shot = {(12, 4): []} mock_ctx.chain_bg_camera_meta_by_shot = {} mock_ctx.chain_bg_guide_by_shot = {} mock_ctx.shot_director_vr = {} # `_analyze_one` 를 mock — fresh path 진입 시 반드시 호출. fresh_card = build_render_prompt_card(**ctx_drifted) fresh_hash = compute_card_hash(fresh_card) def _fake_analyze_one(self, seg, shot_info, ctx, system, schema): return { "scene_index": seg.get("scene_index"), "_shot_index": (shot_info or {}).get("shot_index"), "_user_edited": False, # fresh re-generation 후 False transition "visible_entities": ["C01"], "t2i_variations": [{"t2i_prompt": "fresh prompt"}], "render_prompt_card": fresh_card, "render_prompt_card_hash": fresh_hash, } with patch.object( SceneContextLoader, "load_all", return_value=mock_ctx, ), patch( "app.core.steps.detail_steps.load_prompt", return_value="sys", ), patch( "app.core.steps.detail_steps.load_schema", return_value={}, ), patch.object( SceneDetailStep, "_analyze_one", side_effect=_fake_analyze_one, autospec=True, ) as analyze_mock: result = step._execute() # R3-I2: literal assert — _analyze_one 이 fresh path 진입으로 호출됨. assert analyze_mock.called, ( "R3-I2: _analyze_one not called — _user_edited reuse path " "did not enter fresh re-generation despite card drift." ) # R3-I2: result 의 변경된 scene 이 _user_edited=False 로 transition. scenes = result.get("data", {}).get("scenes", []) assert any( s.get("scene_index") == 12 and s.get("_shot_index") == 4 and s.get("_user_edited") is False for s in scenes ), ( f"R3-I2: edited shot must transition to _user_edited=False after " f"fresh re-generation. scenes={scenes!r}" ) # ────────────────────────────────────────────────────────────────────────── # Class 9: TestV15ToV16MigrationSmoke (R1-B1 / R2-I5 / R3-B1) # ────────────────────────────────────────────────────────────────────────── class TestV15ToV16MigrationSmoke: """R1-I8 + R1-B1 + R2-I5 + R3-B1: v15 cp 잔존 PID resume → schema mismatch detect → step_runner mode="force" 자동 escalate → v16 재실행 trigger. PNG 보존 / cp invalidation 동작. R3-B1: tautological pass 제거 — `analyze_mock.called` literal assert (NOT `or True`) + step_run mode="force" transition assert + cp 가 v16 로 update 되어 `render_prompt_card` + `render_prompt_card_hash` 둘 다 포함 검증. """ def test_v15_cp_lacks_card_field(self) -> None: # 옛 v15 cp 는 render_prompt_card top-level field 자체가 없음. v15_cp = { "schema_version": 6, "prompt_version": "15.202605032354", "t2i_variations": [{"t2i_prompt": "..."}], } assert "render_prompt_card" not in v15_cp assert "render_prompt_card_hash" not in v15_cp def test_v16_cp_has_card_field(self) -> None: ctx = _make_ctx_for_shot() card = build_render_prompt_card(**ctx) v16_cp = { "schema_version": SCENE_DETAIL_SCHEMA_VERSION, "prompt_version": "16.202605041200", "render_prompt_card": card, "render_prompt_card_hash": compute_card_hash(card), "t2i_variations": [{"t2i_prompt": "..."}], } assert v16_cp["render_prompt_card"]["schema_version"] == CARD_SCHEMA_VERSION assert isinstance(v16_cp["render_prompt_card_hash"], str) assert len(v16_cp["render_prompt_card_hash"]) == 16 def test_step_runner_blocks_v15_resume_after_b5( self, tmp_path, monkeypatch, ) -> None: # Block B B5 (plan v2.1.3 §4.5): completed v15 cp fixture → 실제 # SceneDetailStep.run("resume") 호출 → schema mismatch (manifest=7 vs # cp=6) detect → BLOCK (scene_detail 은 _LEGACY_SCHEMA_BUMP_ALLOWLIST # 미포함, downstream cascade 위험으로 자동 escalate 차단). # # 이전 동작 (G4.1): mode="force" 자동 escalate → v16 재실행. # 현재 동작 (B5): AppError(step.resume_invalid, contract_drift) raise — # operator 가 명시 force 또는 manual migration 필요. from app.core.dto.scene_analysis import SceneAnalysisContext from app.core.steps.scene_context_loader import SceneContextLoader # v15 cp fixture: schema_version=6, render_prompt_card 부재. v15_cp_payload = { "schema_version": 6, # 옛 v15 — manifest=7 과 mismatch. "prompt_version": "15.202605032354", "config_hash": "deadbeefcafe0001", # 의도적 mismatch. "data": { "scenes": [{ "scene_index": 12, "_shot_index": 4, "t2i_variations": [{"t2i_prompt": "old prompt"}], }], }, } cp_dir = ( Path(tmp_path) / "P1" / "checkpoints" / "episodes" / "E1" / "scene_detail" ) cp_dir.mkdir(parents=True, exist_ok=True) cp_path = cp_dir / "manifest.json" cp_path.write_text(json.dumps(v15_cp_payload), encoding="utf-8") # SceneDetailStep skeleton (DB stub + projects_dir 우회). step = _make_scene_detail_step_skeleton( tmp_path, monkeypatch, bg_mode="off", ) # Block B T0 (plan v2.1.3): _get_step_run 가 dict 반환 — fetchone() # 의 row 는 attribute access 지원 row-like (status / run_id / 등). from types import SimpleNamespace step.db.execute.return_value.fetchone.return_value = SimpleNamespace( status="completed", run_id="r-test", started_at=None, completed_count=1, applicable_count=1, recovery_count=0, updated_at=None, # 락 소유자 신원 (alembic 010) — 구 행은 전부 None 이다. owner_host=None, owner_pid=None, owner_boot_id=None, heartbeat_at=None, cancel_requested_at=None, ) # SceneContextLoader.load_all mock — _execute 진입 시 가벼운 ctx 반환. mock_ctx = SceneAnalysisContext(project_id="P1", episode_id="E1") mock_ctx.segments = [{"scene_index": 12, "text": "scene 12 text"}] mock_ctx.selected_map = {12: {4}} mock_ctx.shot_scenes_map = { 12: [{"shot_index": 4, "camera_direction": "medium"}], } mock_ctx.scene_visible = {12: ["C01"]} mock_ctx.summaries = {12: "summary"} mock_ctx.world_rules = {} mock_ctx.planning_context = None mock_ctx.staging_map = { "12_4": {"camera_direction": "medium shot of doorway", "framing_scale": "medium"}, } mock_ctx.chain_bg_owned_by_shot = {(12, 4): []} mock_ctx.chain_bg_camera_meta_by_shot = {} mock_ctx.chain_bg_guide_by_shot = {} mock_ctx.shot_director_vr = {} # _analyze_one mock — fresh v16 result with card. fresh_card = build_render_prompt_card(**_make_ctx_for_shot()) fresh_hash = compute_card_hash(fresh_card) def _fake_analyze_one(self, seg, shot_info, ctx, system, schema): return { "scene_index": seg.get("scene_index"), "_shot_index": (shot_info or {}).get("shot_index"), "visible_entities": ["C01"], "t2i_variations": [{ "t2i_prompt": "v16 fresh prompt", "owned_validation": None, }], "render_prompt_card": fresh_card, "render_prompt_card_hash": fresh_hash, } # mode 호출 transition 추적. recorded_modes: List[str] = [] original_execute = SceneDetailStep._execute def _capture_execute(self, mode="resume"): recorded_modes.append(mode) return original_execute(self, mode) with patch.object( SceneContextLoader, "load_all", return_value=mock_ctx, ), patch( "app.core.steps.detail_steps.load_prompt", return_value="sys", ), patch( "app.core.steps.detail_steps.load_schema", return_value={}, ), patch.object( SceneDetailStep, "_analyze_one", side_effect=_fake_analyze_one, autospec=True, ) as analyze_mock, patch.object( SceneDetailStep, "_execute", side_effect=_capture_execute, autospec=True, ), patch.object( SceneDetailStep, "check_applicability", return_value=True, ), patch.object( SceneDetailStep, "check_gate", return_value=None, ), patch.object( SceneDetailStep, "_update_step_run", return_value=None, ), patch.object( SceneDetailStep, "save_checkpoint", return_value=None, ), patch.object( SceneDetailStep, "invalidate_downstream", return_value=None, ), patch.object( SceneDetailStep, "clear_checkpoint", return_value=None, ), patch.object( SceneDetailStep, "cleanup_artifacts", return_value=MagicMock( deleted_db_rows=0, deleted_files=0, targets=[], skipped=[], ), ), patch.object( SceneDetailStep, "_record_recovery", return_value=1, ), patch.object( SceneDetailStep, "_check_recovery_exhausted", return_value=None, ): # B5: manifest schema_version=7 vs cp=6 → contract_drift BLOCK. # scene_detail 은 allowlist 미포함 → AppError raise (자동 escalate X). from app.core.errors import AppError with pytest.raises(AppError) as exc_info: step.run(mode="resume") assert exc_info.value.code == "step.resume_invalid" assert "contract drift" in exc_info.value.message.lower() # _analyze_one / _execute 호출 안 됨 (자동 escalate 차단 가드) assert not analyze_mock.called, ( "auto-escalate blocked — _analyze_one must NOT be called " "(B5 policy: scene_detail schema mismatch → BLOCK)." ) assert "force" not in recorded_modes, ( f"auto-force escalate must NOT happen — recorded modes={recorded_modes}" ) def test_v15_to_v16_cp_field_transition(self) -> None: # R3-B1 contract 진단: v15 cp (schema_version=6) 와 v16 cp 의 차이. # v16 cp 는 card + hash 모두 보유. from app.core.step_manifest import STEP_MANIFEST # manifest 는 v16 schema_version 으로 bumped 되어 있어야 force loop # 회피. (manifest=detail_steps 상수 sync — Wave 1 의무.) manifest_schema = STEP_MANIFEST["scene_detail"].get("schema_version") v15_schema = 6 assert manifest_schema != v15_schema, ( f"manifest scene_detail.schema_version 이 v15 과 같음 " f"(={manifest_schema}) — v16 으로 bump 필요" ) # v16 cp 는 card + hash 모두 보유. ctx = _make_ctx_for_shot() card = build_render_prompt_card(**ctx) v16_cp = { "schema_version": manifest_schema, "render_prompt_card": card, "render_prompt_card_hash": compute_card_hash(card), } assert "render_prompt_card" in v16_cp assert "render_prompt_card_hash" in v16_cp # ────────────────────────────────────────────────────────────────────────── # Class 10: TestVerifyCompletionCleanWithRealInputs (Wave 4 R4 B1) # ────────────────────────────────────────────────────────────────────────── class TestVerifyCompletionCleanWithRealInputs: """Wave 4 R4 BLOCKING 1 (I3): non-trivial inputs (real outlook_pairs + bg_on + fixed_elements + previous_shot_refs + forward_zoom_targets) 로 빌드한 card 를 cp 에 저장 → ``verify_completion()`` 호출 → ``is_complete=True``. 이전 fixture (TestT2iReviewPostEditSentinelStale) 는 verify_completion 의 hardcode empty 인풋 path 와 우연히 match 되도록 설계 — B1 결함 catch 못 함. 본 test 는 ctx-derived single-source 로 recompute 가 제대로 동작 하는지 (false drift cascade 없는지) 진단. """ def _build_real_ctx(self): """non-trivial input 으로 SceneAnalysisContext 를 조립. 설계: outlook_pairs 1+, bg_on=True + bg_id 매핑, fixed_elements 1, previous_shot_refs 1 (loc_refs[0]), forward_zoom_targets 0 (forward 는 본 shot 이 source 인 경우만 — 본 fixture 는 dependency 역방향이 없어 빈 list). """ from app.core.dto.scene_analysis import SceneAnalysisContext ctx = SceneAnalysisContext(project_id="P1", episode_id="E1") ctx.segments = [{"scene_index": 12, "text": "scene 12 text"}] ctx.selected_map = {12: {4}} ctx.shot_scenes_map = { 12: [{"shot_index": 4, "camera_direction": "medium shot of doorway", "description": "wide doorway shot"}], } ctx.scene_visible = {12: ["C01"]} ctx.summaries = {12: "summary"} ctx.world_rules = {} ctx.planning_context = None ctx.staging_map = { "12_4": { "camera_direction": "medium shot of doorway", # framing_scale enum SOT v1 (2026-05-15): helper read fail-fast 정합. "framing_scale": "medium", "lighting_mood": "warm dim", # Area C (2026-05-12) — required by build_id_policy. 빈 list # = 비-재현면 shot (applies=False). "key_bg_elements": [], # Area #1 W5 (2026-05-16) — shot_staging v12 top-level required # field. helper SOT graceful empty. "subject_reference_policy": [], }, } ctx.chain_bg_owned_by_shot = {(12, 4): ["door", "window"]} ctx.chain_bg_camera_meta_by_shot = { (12, 4): { "camera_position": "southeast doorway", "camera_height": "eye-level", "lens_hint": "35mm", "framing_notes": "wide", }, } ctx.chain_bg_guide_by_shot = {(12, 4): "doorway view"} ctx.chain_bg_id_by_shot = {(12, 4): "cb_main_room"} ctx.shot_director_ve = {(12, 4): ["C01"]} ctx.shot_director_vr = {} ctx.outlook_data = { "outlooks": [ {"outlook_id": "O02", "name": "casual_jacket"}, ], "scene_assignments": [ {"scene_index": 12, "assignments": [ {"character_id": "C01", "outlook_id": "O02"}, ]}, ], } ctx.fixed_elements_by_scene = { 12: [{ "element_id": "body_full_pose", "element_type": "character_state", "character_name": "C01", "description": "the figure stands rigid in the doorway", "applies_to_shots": [4], "element_scope": "full", # Area #4 W3 — required field "source_facts": ["scene narration: 'they stood unmoving'"], "visual_inferences": ["lighting falls from corridor"], "creative_decisions": ["camera respects 180-degree line"], "confidence": "high", }], } ctx.dependencies = [ { "scene_index": 12, "shot_index": 4, "location_refs": [{ "scene_index": 12, "shot_index": 3, "ref_usage": "exact_background", "keep_elements": [], }], "character_refs": [], }, ] return ctx def test_verify_completion_clean_when_real_inputs_match_storage( self, tmp_path, monkeypatch, ) -> None: """B1 reproduction: build card via real ctx, store hash in cp, verify_completion 의 recompute 도 동일 ctx 로 → match → clean. 옛 path (verify recompute 가 hardcode empty 인풋 사용) 였다면, 본 test 는 false drift 발생 → fail. ctx-driven single-source fix 후엔 pass. """ from unittest.mock import patch from app.core.steps.detail_steps import ( _collect_card_inputs, SCENE_DETAIL_SCHEMA_VERSION, ) from app.core.steps.scene_context_loader import SceneContextLoader ctx = self._build_real_ctx() # bg_mode=on (fixture has bg-on shot). ⚠️ 반드시 `_collect_card_inputs` # 호출 *전* 에 step skeleton 을 만들어 `settings.background_mode` 를 # patch 해야 한다. `_derive_card_inputs_from_ctx` (detail_steps.py:424) # 가 `settings.background_mode` 를 직접 참조해 `background_mode_on` 을 # 도출하므로, stored card 빌드 시점과 recompute 시점의 settings 가 # 달라지면 hash drift 발생 (production 에서는 단일 process 가 같은 # settings 를 보므로 발생하지 않는 fixture 한정 버그). step = _make_scene_detail_step_skeleton( tmp_path, monkeypatch, bg_mode="on", ) # `_analyze_one()` 와 동일한 ctx-driven single source 로 card 빌드. inputs = _collect_card_inputs( ctx=ctx, seg={"scene_index": 12}, shot_info={"shot_index": 4}, ) builder_inputs = {k: v for k, v in inputs.items() if k != "ctx"} card = build_render_prompt_card(**builder_inputs) card_hash = compute_card_hash(card) # cp fixture: card + hash + sentinel match (no drift). from app.core.steps._owned_helpers import ( OWNED_VALIDATOR_FULL, compute_camera_direction_hash, compute_owned_hash, compute_t2i_prompt_hash, ) cam_dir = "medium shot of doorway" owned = ["door", "window"] prompt_text = "A figure stands by the doorway." cp_result = { "schema_version": SCENE_DETAIL_SCHEMA_VERSION, "prompt_version": "16.202605041200", "data": { "scenes": [{ "scene_index": 12, "_shot_index": 4, "visible_entities": ["C01"], "render_prompt_card": card, "render_prompt_card_hash": card_hash, "t2i_variations": [{ "t2i_prompt": prompt_text, # Fix B (2026-05-10): verify_completion recompute path 가 # narrow args 적용 — outfit_assignments 가 outlook_pairs # union 과 동일해야 wide=narrow → hash match. "outfit_assignments": [ {"character_id": "C01", "outlook_id": "O02"}, ], "owned_validation": { "schema_version": 2, "validator": OWNED_VALIDATOR_FULL, "owned_hash": compute_owned_hash(owned), "camera_direction_hash": ( compute_camera_direction_hash(cam_dir) ), "t2i_prompt_hash": compute_t2i_prompt_hash(prompt_text), "owned_usage_hash": "0123456789abcdef", # C2 v1 sentinel v2 "violations": [], }, }], }], }, } step._last_execute_result = cp_result # SceneContextLoader.load_all 를 mock 해서 fixture ctx 반환. with patch.object( SceneContextLoader, "load_all", return_value=ctx, ), patch.object( SceneContextLoader, "_load_chain_bg_owned_by_shot", return_value=ctx.chain_bg_owned_by_shot, ), patch.object( SceneContextLoader, "_load_staging_map", return_value=ctx.staging_map, ): report = step.verify_completion() # B1 fix 검증: real-input recompute 가 stored card hash 와 일치 → # clean. 옛 hardcoded empty path 였다면 false drift → partial. assert report.is_complete is True, ( f"B1: real-input recompute must match stored card hash → clean. " f"missing={list(report.missing)!r} severity={report.severity!r}" ) assert report.severity == "clean", ( f"B1: severity must be 'clean' (got {report.severity!r})" )