"""shot_conti_light — 경량 콘티 대상 선별·플레이트 리졸버·프롬프트 조립 테스트.""" from __future__ import annotations import json from pathlib import Path from typing import Any, Dict, List from app.modules.pipeline.shot_conti_light import ( build_conti_prompt, conti_targets, resolve_shot_plate_map, run_shot_conti_light, ) CLASSIFY = { "S1sh1": {"person_visible": False, "prev": None, "usage_en": ""}, "S1sh2": {"person_visible": True, "prev": None, "usage_en": ""}, "S1sh3": {"person_visible": True, "prev": "S1sh2", "usage_en": "TAKE x."}, "S2sh1": {"person_visible": True, "prev": None, "usage_en": ""}, } def test_targets_exclude_bgonly_and_prev(): assert conti_targets(CLASSIFY) == ["S1sh2", "S2sh1"] # ── 플레이트 리졸버 ──────────────────────────────────────────────────── def _write_cp(base: Path, step: str, data: Dict[str, Any]): d = base / "checkpoints" / "episodes" / "ep1" / step d.mkdir(parents=True, exist_ok=True) (d / "manifest.json").write_text( json.dumps({"data": data}), encoding="utf-8" ) def test_resolve_shot_plate_map_phase7(tmp_path): proj = tmp_path / "p1" png = proj / "bg.png" png.parent.mkdir(parents=True) png.write_bytes(b"png") missing = proj / "gone.png" _write_cp( proj, "background_render", { "groups": { "bg1": {"status": "ok", "png_path": str(png), "shot_ids": ["1_1", "1_2"]}, "bg2": {"status": "ok", "png_path": str(missing), "shot_ids": ["2_1"]}, "bg3": {"status": "failed", "png_path": str(png), "shot_ids": ["3_1"]}, } }, ) out = resolve_shot_plate_map(str(tmp_path), "p1", "ep1") assert out == {"1_1": png, "1_2": png} # 파일 부재·failed 는 제외 def test_resolve_shot_plate_map_canonical_shot_id_shape(tmp_path): """정본 shot_ids shape "S{si}_Shot{shi}" 파싱 (E2E 5회차 실측 결함 fix).""" proj = tmp_path / "p1" png = proj / "bg.png" png.parent.mkdir(parents=True) png.write_bytes(b"png") _write_cp( proj, "background_render", {"groups": {"L03B02": {"status": "ok", "png_path": str(png), "shot_ids": ["S4_Shot1", "S10_Shot6"]}}}, ) out = resolve_shot_plate_map(str(tmp_path), "p1", "ep1") assert out == {"4_1": png, "10_6": png} def test_resolve_shot_plate_map_chain_fallback(tmp_path): proj = tmp_path / "p1" png = proj / "chain.png" png.parent.mkdir(parents=True) png.write_bytes(b"png") _write_cp( proj, "background_chain_render", {"groups": {"bgA": {"status": "ok", "png_path": str(png), "shot_ids": ["1_1"]}}}, ) out = resolve_shot_plate_map(str(tmp_path), "p1", "ep1") assert out == {"1_1": png} def test_resolve_shot_plate_map_render_takes_precedence(tmp_path): proj = tmp_path / "p1" png_a = proj / "a.png" png_b = proj / "b.png" png_a.parent.mkdir(parents=True) png_a.write_bytes(b"a") png_b.write_bytes(b"b") _write_cp(proj, "background_render", {"groups": {"x": {"status": "ok", "png_path": str(png_a), "shot_ids": ["1_1"]}}}) _write_cp(proj, "background_chain_render", {"groups": {"y": {"status": "ok", "png_path": str(png_b), "shot_ids": ["1_1"]}}}) out = resolve_shot_plate_map(str(tmp_path), "p1", "ep1") assert out["1_1"] == png_a # ── 프롬프트 조립 ────────────────────────────────────────────────────── def test_prompt_assembly_order(): prompt = build_conti_prompt( shot_desc="남자가 문을 연다", place_text="허름한 방", has_plate=True, pose_clauses=["IMMOBILE POSE CLAUSE X"], carried_en="He grips a photo.", movement_en="", figures_en="", ) idx = { "head": prompt.index("Draw ONE storyboard layout frame"), "light": prompt.index("LAYOUT-ONLY THUMBNAIL SKETCH"), "bgref": prompt.index("LOOSE spatial reference"), "loc": prompt.index("THE LOCATION: 허름한 방"), "shot": prompt.index("SHOT TEXT (authoritative, Korean): 남자가 문을 연다"), "pose": prompt.index("IMMOBILE POSE CLAUSE X"), "carried": prompt.index("CARRIED STATE (persist exactly): He grips a photo."), "notext": prompt.index("ABSOLUTELY NO TEXT anywhere"), } order = ["head", "light", "bgref", "loc", "shot", "pose", "carried", "notext"] assert sorted(idx, key=idx.get) == order def test_prompt_without_plate_omits_bgref(): prompt = build_conti_prompt( shot_desc="d", place_text="p", has_plate=False, pose_clauses=[], carried_en="", movement_en="", figures_en="", ) assert "LOOSE spatial reference" not in prompt # ── 오케스트레이션 ───────────────────────────────────────────────────── class FakeGen: def __init__(self): self.calls: List[Dict[str, Any]] = [] def __call__(self, tag, prompt, plate_path, out_path: Path) -> Path: self.calls.append({"tag": tag, "plate": plate_path, "out": out_path}) out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_bytes(b"conti") return out_path def _shots(): return [ {"scene_index": 1, "shot_index": 1, "description": "빈 방"}, {"scene_index": 1, "shot_index": 2, "description": "남자"}, {"scene_index": 1, "shot_index": 3, "description": "이어짐"}, {"scene_index": 2, "shot_index": 1, "description": "여자"}, ] def test_run_skips_and_generates(tmp_path): gen = FakeGen() plate = tmp_path / "plate.png" plate.write_bytes(b"p") out = run_shot_conti_light( shots=_shots(), classify_shots=CLASSIFY, continuity={"pose_canon": [], "carried": {}, "pose_fix": {}}, location_by_scene={1: "방", 2: "골목"}, plate_map={"1_2": plate}, # S2sh1 은 플레이트 없음 out_dir=tmp_path / "conti", gen_fn=gen, ) contis = out["contis"] assert contis["S1sh1"]["skipped_reason"] == "bgonly" assert contis["S1sh3"]["skipped_reason"] == "prev" assert contis["S2sh1"]["skipped_reason"] == "no_plate" assert contis["S1sh2"]["skipped_reason"] is None assert contis["S1sh2"]["image_path"].endswith("conti_S1sh2.png") assert [c["tag"] for c in gen.calls] == ["S1sh2"] assert out["completed_count"] == 1 def _run_conti(tmp_path, gen, plate, out_dir, **over): kw = dict( shots=_shots(), classify_shots=CLASSIFY, continuity={"pose_canon": [], "carried": {}, "pose_fix": {}}, location_by_scene={1: "방", 2: "골목"}, plate_map={"1_2": plate}, out_dir=out_dir, gen_fn=gen, ) kw.update(over) return run_shot_conti_light(**kw) def test_run_resume_skips_with_matching_fingerprint(tmp_path): """지문 일치 sidecar + PNG 존재 → 재개 skip (2차 리뷰 B3 계약).""" plate = tmp_path / "plate.png" plate.write_bytes(b"p") out_dir = tmp_path / "conti" # 1차 실행으로 PNG+sidecar 생성 _run_conti(tmp_path, FakeGen(), plate, out_dir) gen2 = FakeGen() out = _run_conti(tmp_path, gen2, plate, out_dir) assert gen2.calls == [] # 지문 일치 → skip assert out["contis"]["S1sh2"]["skipped_reason"] is None def test_run_regenerates_on_input_change_and_archives(tmp_path): """플레이트 내용 변경(예: 일반→맵 플레이트 전환) → stale 아카이브 후 재생성 (2차 리뷰 B3: 파일 존재만으로 skip 금지).""" plate = tmp_path / "plate.png" plate.write_bytes(b"normal-plate") out_dir = tmp_path / "conti" _run_conti(tmp_path, FakeGen(), plate, out_dir) plate.write_bytes(b"map-plate-content") # 맵 플레이트로 대체된 상황 gen2 = FakeGen() out = _run_conti(tmp_path, gen2, plate, out_dir) assert [c["tag"] for c in gen2.calls] == ["S1sh2"] # 재생성 assert out["contis"]["S1sh2"]["skipped_reason"] is None assert list(out_dir.glob("conti_S1sh2.stale_*.png")) # 구본 보존 def test_run_extra_fingerprint_change_regenerates(tmp_path): plate = tmp_path / "plate.png" plate.write_bytes(b"p") out_dir = tmp_path / "conti" _run_conti(tmp_path, FakeGen(), plate, out_dir, extra_fingerprint={"pack": "1"}) gen2 = FakeGen() _run_conti(tmp_path, gen2, plate, out_dir, extra_fingerprint={"pack": "2"}) assert [c["tag"] for c in gen2.calls] == ["S1sh2"] def test_run_force_regenerates_even_when_matching(tmp_path): plate = tmp_path / "plate.png" plate.write_bytes(b"p") out_dir = tmp_path / "conti" _run_conti(tmp_path, FakeGen(), plate, out_dir) gen2 = FakeGen() _run_conti(tmp_path, gen2, plate, out_dir, force=True) assert [c["tag"] for c in gen2.calls] == ["S1sh2"] def test_run_existing_png_without_sidecar_regenerates(tmp_path): """무기록 PNG = 출처 불명 → 재생성.""" gen = FakeGen() plate = tmp_path / "plate.png" plate.write_bytes(b"p") out_dir = tmp_path / "conti" out_dir.mkdir() (out_dir / "conti_S1sh2.png").write_bytes(b"old") out = _run_conti(tmp_path, gen, plate, out_dir) assert [c["tag"] for c in gen.calls] == ["S1sh2"] assert out["contis"]["S1sh2"]["skipped_reason"] is None def test_pose_and_carried_injected(tmp_path): gen = FakeGen() plate = tmp_path / "plate.png" plate.write_bytes(b"p") captured = {} def gen_capture(tag, prompt, plate_path, out_path): captured[tag] = prompt out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_bytes(b"c") return out_path run_shot_conti_light( shots=_shots(), classify_shots=CLASSIFY, continuity={ "pose_canon": [ {"character_ko": "남자", "pose_en": "slumped against wall", "shots": ["S1sh2"], "basis_ko": ""} ], # v3 (감사 1-B, 2026-08-27): carried 는 세 갈래이고 # `pose_fix` 의 교정본은 **제자리에** 들어 있다. "carried": {"S1sh2": { "carried_contract": "scoped_v1", "objects_en": "the wall is scuffed", "people": [{"character_ko": "남자", "character_short_id": "C01", "state_en": "grips a photo (fixed)", "basis_ko": ""}], "offscreen_effects_en": "", "basis_ko": ""}}, "pose_fix": {"S1sh2": {"movement_en": "", "figures_en": ""}}, }, location_by_scene={1: "방", 2: "골목"}, plate_map={"1_2": plate}, out_dir=tmp_path / "conti", gen_fn=gen_capture, visible_char_sids_by_tag={"S1sh2": {"C01"}}, ) p = captured["S1sh2"] assert "slumped against wall" in p # pose_fix 교정본이 제자리에 들어와 그대로 나간다 assert "grips a photo (fixed)" in p assert "the wall is scuffed" in p def test_carried_person_not_in_this_shot_is_not_injected(tmp_path): """★감사 1-B — 그 샷에 배정 안 된 인물의 상태는 안 나간다. 같은 프롬프트에 "they must be one of: …" 가 함께 나가므로, 목록 밖 인물을 CARRIED 로 밀어 넣으면 모델이 둘 중 하나를 고른다. """ plate = tmp_path / "plate.png" plate.write_bytes(b"p") captured = {} def gen_capture(tag, prompt, plate_path, out_path): captured[tag] = prompt out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_bytes(b"c") return out_path run_shot_conti_light( shots=_shots(), classify_shots=CLASSIFY, continuity={ "pose_canon": [], "carried": {"S1sh2": { "carried_contract": "scoped_v1", "objects_en": "the toolbox stays open", "people": [ {"character_ko": "남자", "character_short_id": "C01", "state_en": "the man still holds the photo", "basis_ko": ""}, {"character_ko": "여자", "character_short_id": "C02", "state_en": "the woman waits by the door", "basis_ko": ""}], "offscreen_effects_en": "", "basis_ko": ""}}, "pose_fix": {}, }, location_by_scene={1: "방", 2: "골목"}, plate_map={"1_2": plate}, out_dir=tmp_path / "conti", gen_fn=gen_capture, visible_char_sids_by_tag={"S1sh2": {"C01"}}, ) p = captured["S1sh2"] assert "the toolbox stays open" in p assert "the man still holds the photo" in p assert "the woman waits by the door" not in p, ( "그 샷에 없는 인물의 상태가 콘티 프롬프트에 실렸다") def test_legacy_carried_stops_the_conti_before_buying(tmp_path): """옛 한 칸 값으로는 **새 콘티를 사지 않는다.**""" import pytest from app.modules.pipeline.shot_continuity_author import ( LegacyCarriedContract, ) plate = tmp_path / "plate.png" plate.write_bytes(b"p") bought = [] def gen_capture(tag, prompt, plate_path, out_path): bought.append(tag) out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_bytes(b"c") return out_path with pytest.raises(LegacyCarriedContract) as e: run_shot_conti_light( shots=_shots(), classify_shots=CLASSIFY, continuity={ "pose_canon": [], "carried": {"S1sh2": {"carried_en": "grips a photo"}}, "pose_fix": {}, }, location_by_scene={1: "방", 2: "골목"}, plate_map={"1_2": plate}, out_dir=tmp_path / "conti", gen_fn=gen_capture, ) assert "force" in str(e.value) assert not bought, "멈추기 전에 그림을 샀다" def test_step_registered(): from app.core.step_catalog import STEP_CATALOG from app.core.steps import STEP_CLASSES assert "shot_conti_light" in STEP_CLASSES entry = STEP_CATALOG["shot_conti_light"] assert entry.applicability == "if_still_recipe" assert "shot_ref_classify" in entry.depends_on assert "shot_continuity" in entry.depends_on # ── intermediate 등록 (3차 리뷰 B1) ─────────────────────────────────── class _FakeAsset: def __init__(self, aid, rel): self.id = aid self.file_path = rel self.prompt_used = None self.generation_model = None def input_edge(self): import json as _j raw = getattr(self, "input_image_ids", None) return _j.loads(raw) if raw is not None else None def meta(self): import json as _j raw = getattr(self, "pipeline_metadata_json", None) return _j.loads(raw) if raw else {} class _FakeStore: """rel→asset 저장소 — annotate 는 **실제** annotate_generated_asset 을 사용해 None=미갱신/metadata merge semantics 를 그대로 재현 (4차 H1).""" def __init__(self): self.by_rel = {} self.seq = 0 def find(self, rel): return self.by_rel.get(rel) def new(self, *, rel, tag, asset_type, model, prompt): self.seq += 1 a = _FakeAsset(f"asset-{self.seq}", rel) a.asset_type = asset_type a.prompt_used = prompt a.generation_model = model self.by_rel[rel] = a return a def annotate(self, asset, *, role, input_ids, meta): from app.services.image_capture.annotate import ( annotate_generated_asset, ) annotate_generated_asset( asset, pipeline_role=role, stage="conti", input_image_ids=input_ids, pipeline_metadata=meta, ) def _register(store, contis, map_plates): from app.modules.pipeline.shot_conti_light import ( register_intermediate_assets, ) register_intermediate_assets( contis=contis, map_plates=map_plates, rel_fn=lambda p: p, # 테스트: 상대화 identity find_asset_by_rel=store.find, new_asset=store.new, annotate_fn=store.annotate, conti_model="gpt-image-2", plate_model="nb2", ) def _touch(tmp_path, name, content=b"x"): p = tmp_path / name p.write_bytes(content) return p def test_fresh_map_flow_has_both_edges(tmp_path): """(a) fresh run: canon(master/map)→map_plate 와 map_plate→conti 두 edge 모두 성립 — map_plate 선등록으로 conti 입력이 UUID 로 이어진다.""" master = _touch(tmp_path, "master.png") site = _touch(tmp_path, "map.png") plate = _touch(tmp_path, "map_plate_S1sh1.png") conti = _touch(tmp_path, "conti_S1sh1.png") store = _FakeStore() # canon 자산은 이미 등록돼 있음 (outdoor_place_canon) store.by_rel[str(master)] = _FakeAsset("canon-master", str(master)) store.by_rel[str(site)] = _FakeAsset("canon-map", str(site)) map_plates = {"S1sh1": { "plate_path": str(plate), "prompt": "map plate prompt", "source_master_png": str(master), "source_map_png": str(site), }} contis = {"S1sh1": { "image_path": str(conti), "plate_path": str(plate), "prompt": "conti prompt", "skipped_reason": None, }} _register(store, contis, map_plates) plate_asset = store.by_rel[str(plate)] assert plate_asset.input_edge() == ["canon-master", "canon-map"] conti_asset = store.by_rel[str(conti)] assert conti_asset.input_edge() == [plate_asset.id] assert map_plates["S1sh1"]["asset_id"] == plate_asset.id assert contis["S1sh1"]["asset_id"] == conti_asset.id def test_resume_after_rollback_recovers_edges(tmp_path): """(b) PNG+sidecar 만 남은 재시도(등록 rollback 모사)에서도 reuse entry 의 source 경로로 두 edge 복구.""" master = _touch(tmp_path, "master.png") site = _touch(tmp_path, "map.png") plate = _touch(tmp_path, "map_plate_S1sh1.png") conti = _touch(tmp_path, "conti_S1sh1.png") store = _FakeStore() # rollback → 아무것도 등록 안 된 상태 store.by_rel[str(master)] = _FakeAsset("canon-master", str(master)) store.by_rel[str(site)] = _FakeAsset("canon-map", str(site)) # reuse 경로 entry 에도 source 2경로 존재 (outdoor_map_conti reuse 계약) map_plates = {"S1sh1": { "plate_path": str(plate), "prompt": None, "source_master_png": str(master), "source_map_png": str(site), }} contis = {"S1sh1": {"image_path": str(conti), "plate_path": str(plate), "prompt": None, "skipped_reason": None}} _register(store, contis, map_plates) assert store.by_rel[str(plate)].input_edge() == [ "canon-master", "canon-map", ] assert store.by_rel[str(conti)].input_edge() == [ store.by_rel[str(plate)].id, ] def test_existing_row_lineage_updated_on_plate_switch(tmp_path): """(c) 일반→맵 플레이트 전환 후 같은 경로 기존 conti row 의 input_image_ids 가 새 plate 로 교체 (stale edge 잔존 금지).""" old_plate = _touch(tmp_path, "old_plate.png") new_plate = _touch(tmp_path, "map_plate_S1sh1.png") conti = _touch(tmp_path, "conti_S1sh1.png") store = _FakeStore() store.by_rel[str(old_plate)] = _FakeAsset("old-plate", str(old_plate)) # 기존 conti row (예전 plate edge 를 갖고 있었음) existing_conti = _FakeAsset("conti-row", str(conti)) store.annotate( existing_conti, role="conti_light", input_ids=["old-plate"], meta={"shot_tag": "S1sh1", "unresolved_input_paths": []}, ) existing_conti.generation_model = "old-model" store.by_rel[str(conti)] = existing_conti map_plates = {"S1sh1": { "plate_path": str(new_plate), "prompt": "p", "source_master_png": None, "source_map_png": None, }} contis = {"S1sh1": {"image_path": str(conti), "plate_path": str(new_plate), "prompt": "new prompt", "skipped_reason": None}} _register(store, contis, map_plates) new_plate_id = store.by_rel[str(new_plate)].id assert existing_conti.input_edge() == [new_plate_id] assert existing_conti.prompt_used == "new prompt" assert existing_conti.generation_model == "gpt-image-2" # 4차 MINOR def test_existing_row_edges_cleared_and_restored(tmp_path): """4차 H1 전이: 기존 row(old edge+old unresolved) → (a) 현재 입력 미해결이면 input_image_ids=[] 로 old edge 제거+unresolved 기록 → (b) 이후 resolve 되면 새 UUID 교체+unresolved=[] (실제 annotate semantics — None=미갱신/metadata merge — 로 검증).""" plate = _touch(tmp_path, "plate.png") # asset row 없음 (미해결 상황) conti = _touch(tmp_path, "conti_S1sh1.png") store = _FakeStore() existing = _FakeAsset("conti-row", str(conti)) store.annotate( existing, role="conti_light", input_ids=["old-plate"], meta={"shot_tag": "S1sh1", "unresolved_input_paths": ["/old/missing.png"]}, ) store.by_rel[str(conti)] = existing contis = {"S1sh1": {"image_path": str(conti), "plate_path": str(plate), "prompt": None, "skipped_reason": None}} # (a) 현재 입력 미해결 → old edge 제거([]) + 현재 unresolved 로 교체 _register(store, contis, {}) assert existing.input_edge() == [] assert existing.meta()["unresolved_input_paths"] == [str(plate)] # (b) plate asset 이 등록된 뒤 재실행 → 새 UUID + unresolved=[] store.by_rel[str(plate)] = _FakeAsset("plate-row", str(plate)) _register(store, contis, {}) assert existing.input_edge() == ["plate-row"] assert existing.meta()["unresolved_input_paths"] == [] def test_unresolved_existing_input_recorded(tmp_path): """기대 입력 파일이 실재하는데 asset UUID 미해결 → unresolved 진단.""" plate = _touch(tmp_path, "plate.png") # asset row 없음 conti = _touch(tmp_path, "conti_S1sh1.png") store = _FakeStore() contis = {"S1sh1": {"image_path": str(conti), "plate_path": str(plate), "prompt": None, "skipped_reason": None}} _register(store, contis, {}) conti_asset = store.by_rel[str(conti)] assert conti_asset.input_edge() == [] # exact current-state (H1) assert conti_asset.meta()["unresolved_input_paths"] == [str(plate)] def test_recipe_fresh_run_ordering(): """Codex 1차 리뷰 BLOCKING-1 회귀 가드: fresh run 에서 콘티는 플레이트 (background_render) 뒤에 실행되고, scene_image_pipeline 은 레시피 3스텝 생산자 완료를 dependency 로 보장한다.""" from app.core.step_manifest import STEP_MANIFEST conti = STEP_MANIFEST["shot_conti_light"] bg = STEP_MANIFEST["background_render"] assert conti["order"] > bg["order"] assert "background_render" in conti["depends_on"] sip = STEP_MANIFEST["scene_image_pipeline"] for dep in ("shot_ref_classify", "shot_continuity", "shot_conti_light"): assert dep in sip["depends_on"], dep assert sip["order"] > conti["order"] def test_lane_flow_sketch_edge_direct_to_clean_map(tmp_path): """R4 (2026-07-16): PIL control asset 소멸 — 스케치 input=클린 canon 맵 UUID 직결, role/asset_type=중립(lane_storyboard_sketch), lane_marker_control 등록 0.""" from app.modules.pipeline.shot_conti_light import ( register_intermediate_assets, ) base = _touch(tmp_path, "canon_map.png") sketch = _touch(tmp_path, "lane_sketch_S1sh2.png") store = _FakeStore() store.by_rel[str(base)] = _FakeAsset("canon-map", str(base)) lane_contis = {"S1sh2": { "status": "ok", "lane": "map_marker", "image_path": str(sketch), "base_map_path": str(base), "prompt": "lane sketch prompt", }} register_intermediate_assets( contis={}, map_plates={}, lane_contis=lane_contis, rel_fn=lambda p: p, find_asset_by_rel=store.find, new_asset=store.new, annotate_fn=store.annotate, conti_model="gpt-image-2", plate_model="nb2", ) sketch_asset = store.by_rel[str(sketch)] assert sketch_asset.input_edge() == ["canon-map"] assert sketch_asset.asset_type == "lane_storyboard_sketch" assert sketch_asset.pipeline_role == "lane_storyboard_sketch" assert sketch_asset.prompt_used == "lane sketch prompt" assert lane_contis["S1sh2"]["asset_id"] == sketch_asset.id # control asset 미등록 (PIL 산출 소멸) assert not any( getattr(a, "asset_type", None) == "lane_marker_control" for a in store.by_rel.values() ) def test_structure_policy_entry_not_registered(tmp_path): """structure_plate 정책 entry(image_path 없음)는 asset 등록 자연 skip.""" from app.modules.pipeline.shot_conti_light import ( register_intermediate_assets, ) store = _FakeStore() lane_contis = {"S4sh1": { "status": "ab_select_ready", "lane": "structure_plate", "seed_path": str(_touch(tmp_path, "seed.png")), }} register_intermediate_assets( contis={}, map_plates={}, lane_contis=lane_contis, rel_fn=lambda p: p, find_asset_by_rel=store.find, new_asset=store.new, annotate_fn=store.annotate, conti_model="gpt-image-2", plate_model="nb2", ) assert store.by_rel == {} assert "asset_id" not in lane_contis["S4sh1"] def test_step_source_has_no_pil_marker_render(): """지시 ① 구조 잠금: 스텝이 render_marker_map(PIL 합성)을 참조하지 않는다 — 코드가 이미지에 그리는 경로 0.""" import inspect from app.core.steps import shot_conti_light_step as mod assert "render_marker_map" not in inspect.getsource(mod) # ── R2 (2026-07-16): 복잡 구조물 A/B 정책 finalize ─────────────────────── def _pending_entry(tmp_path, seed_ok=True): seed = tmp_path / "seed.png" if seed_ok: seed.write_bytes(b"SAMPLE") return { "lane": "structure_plate", "group_id": "G1", "segment_id": "seg1", "status": "ab_select_pending", "seed_path": str(seed), "seed_asset_id": "aid-seed", "seed_status": "ok" if seed_ok else "failed", # 2026-07-17 seed-bg: typed 배경 권위 — 기존 테스트는 plate 케이스 "bg_source": "plate", "bg_path": "", "bg_reason": "", } def test_finalize_ab_ready_when_seed_plate_conti_all_present(tmp_path): from app.modules.pipeline.shot_conti_light import ( finalize_structure_ab_entries, ) lane = {"S4sh1": _pending_entry(tmp_path)} conti_png = tmp_path / "conti.png" plate_png = tmp_path / "plate.png" conti_png.write_bytes(b"SAMPLE") plate_png.write_bytes(b"SAMPLE") contis = {"S4sh1": { "image_path": str(conti_png), "plate_path": str(plate_png), "error": None, "skipped_reason": None, }} classify = {"S4sh1": {"person_visible": True, "prev": None}} delta = finalize_structure_ab_entries( lane_contis=lane, contis=contis, classify_shots=classify) assert lane["S4sh1"]["status"] == "ab_select_ready" assert lane["S4sh1"]["conti_path"].endswith("conti.png") assert lane["S4sh1"]["plate_path"].endswith("plate.png") # unique shot 계약 — 일반 콘티 집계에 이미 포함, lane 가산 0 assert delta == {"applicable": 0, "completed": 0, "failed": 0} def test_finalize_ab_bypass_bgonly_and_prev(tmp_path): from app.modules.pipeline.shot_conti_light import ( finalize_structure_ab_entries, ) lane = { "S4sh1": _pending_entry(tmp_path), "S4sh2": _pending_entry(tmp_path), } classify = { "S4sh1": {"person_visible": False, "prev": None}, "S4sh2": {"person_visible": True, "prev": "S4sh1"}, } delta = finalize_structure_ab_entries( lane_contis=lane, contis={}, classify_shots=classify) assert lane["S4sh1"]["status"] == "ab_select_bypass" assert lane["S4sh1"]["bypass_reason"] == "bg_only" assert lane["S4sh2"]["status"] == "ab_select_bypass" assert lane["S4sh2"]["bypass_reason"] == "prev" assert delta == {"applicable": 0, "completed": 0, "failed": 0} def test_finalize_ab_missing_conti_fail_closed_counts_once(tmp_path): from app.modules.pipeline.shot_conti_light import ( finalize_structure_ab_entries, ) lane = {"S4sh1": _pending_entry(tmp_path)} lane["S4sh1"].update( bg_source="failed", bg_reason="배정 플레이트 결손(assigned=L04B01) — seed 하강 금지", ) contis = {"S4sh1": { "image_path": None, "plate_path": None, "error": None, "skipped_reason": "no_plate", }} classify = {"S4sh1": {"person_visible": True, "prev": None}} delta = finalize_structure_ab_entries( lane_contis=lane, contis=contis, classify_shots=classify) # eligible 결손 = fail-closed (B 단독 조용한 진행 금지) assert lane["S4sh1"]["status"] == "failed" assert "seed 하강 금지" in lane["S4sh1"]["error"] # 일반 콘티가 대상으로 치지 않은 결손 — lane 가산 1 assert delta == {"applicable": 1, "completed": 0, "failed": 1} def test_finalize_ab_conti_gen_error_no_double_count(tmp_path): from app.modules.pipeline.shot_conti_light import ( finalize_structure_ab_entries, ) lane = {"S4sh1": _pending_entry(tmp_path)} contis = {"S4sh1": { "image_path": None, "plate_path": str(tmp_path / "p.png"), "error": "SAMPLE gen error", "skipped_reason": None, }} classify = {"S4sh1": {"person_visible": True, "prev": None}} delta = finalize_structure_ab_entries( lane_contis=lane, contis=contis, classify_shots=classify) assert lane["S4sh1"]["status"] == "failed" # 생성 실패는 일반 콘티 failed_count 에 이미 포함 — 이중 가산 금지 assert delta == {"applicable": 0, "completed": 0, "failed": 0} def test_finalize_ab_seed_missing_fail_closed(tmp_path): from app.modules.pipeline.shot_conti_light import ( finalize_structure_ab_entries, ) lane = {"S4sh1": _pending_entry(tmp_path, seed_ok=False)} conti_png = tmp_path / "conti.png" plate_png = tmp_path / "plate.png" conti_png.write_bytes(b"SAMPLE") plate_png.write_bytes(b"SAMPLE") contis = {"S4sh1": { "image_path": str(conti_png), "plate_path": str(plate_png), "error": None, "skipped_reason": None, }} classify = {"S4sh1": {"person_visible": True, "prev": None}} delta = finalize_structure_ab_entries( lane_contis=lane, contis=contis, classify_shots=classify) assert lane["S4sh1"]["status"] == "failed" assert "seed" in lane["S4sh1"]["error"] # 배치 리뷰 BLOCKING-1: 일반 콘티 completed 로 이미 가산된 샷의 정책 # 실패 = completed -1 / failed +1 (applicable 이중 가산 0 — # 샷 최종값 applicable=1/completed=0/failed=1) assert delta == {"applicable": 0, "completed": -1, "failed": 1} def test_finalize_ab_ignores_map_marker_entries(tmp_path): from app.modules.pipeline.shot_conti_light import ( finalize_structure_ab_entries, ) lane = {"S5sh1": {"lane": "map_marker", "status": "ok", "image_path": "x.png"}} delta = finalize_structure_ab_entries( lane_contis=lane, contis={}, classify_shots={}) assert lane["S5sh1"]["status"] == "ok" assert delta == {"applicable": 0, "completed": 0, "failed": 0} # ── seed-bg 승격 (2026-07-17 Codex 합의) — typed 배경 권위 해석 ────────── def test_bg_resolver_priority_plate_over_seed(tmp_path): """(d) plate+seed 있음 → plate (기존 계약 불변).""" from app.modules.pipeline.shot_conti_light import ( resolve_structure_bg_source, ) plate = _touch(tmp_path, "plate.png") seed = _touch(tmp_path, "seed.png") src, path, reason = resolve_structure_bg_source( shot_key="7_2", plate_map={"7_2": plate}, assign_by_key={"7_2": "L05B01"}, seed_path=str(seed), seed_status="ok", ) assert (src, path) == ("plate", str(plate)) and not reason def test_bg_resolver_structural_absence_promotes_seed(tmp_path): """(a) 배정 기록 자체가 없고 seed 유효 → seed-bg 승격.""" from app.modules.pipeline.shot_conti_light import ( resolve_structure_bg_source, ) seed = _touch(tmp_path, "seed.png") src, path, _ = resolve_structure_bg_source( shot_key="7_2", plate_map={}, assign_by_key={}, seed_path=str(seed), seed_status="ok", ) assert (src, path) == ("seed", str(seed)) def test_bg_resolver_assigned_but_missing_never_falls_to_seed(tmp_path): """(c) 배정 기록이 있는데 파일/그룹 결손 → seed 하강 금지, failed (손상 감지 fail-closed — Codex 조건 2).""" from app.modules.pipeline.shot_conti_light import ( resolve_structure_bg_source, ) seed = _touch(tmp_path, "seed.png") src, path, reason = resolve_structure_bg_source( shot_key="7_2", plate_map={}, assign_by_key={"7_2": "L05B01"}, seed_path=str(seed), seed_status="ok", ) assert src == "failed" and "seed 하강 금지" in reason def test_bg_resolver_no_plate_no_seed_failed(tmp_path): """(b) plate 없음+seed 없음 → failed.""" from app.modules.pipeline.shot_conti_light import ( resolve_structure_bg_source, ) src, _, reason = resolve_structure_bg_source( shot_key="7_2", plate_map={}, assign_by_key={}, seed_path=None, seed_status="missing", ) assert src == "failed" and "소스 없음" in reason def _seed_bg_entry(tmp_path, **over): seed = tmp_path / "seed.png" if not seed.exists(): seed.write_bytes(b"SAMPLE") e = { "lane": "structure_plate", "group_id": "G1", "segment_id": "s1", "status": "ab_select_pending", "seed_path": str(seed), "seed_asset_id": "aid-seed", "seed_status": "ok", "bg_source": "seed", "bg_path": str(seed), "bg_reason": "", } e.update(over) return e def test_finalize_seed_bg_ready_requires_conti_from_seed(tmp_path): """(a) seed-bg eligible: 콘티가 정확히 seed 를 참조했을 때만 ready.""" from app.modules.pipeline.shot_conti_light import ( finalize_structure_ab_entries, ) lane = {"S7sh2": _seed_bg_entry(tmp_path)} conti_png = _touch(tmp_path, "conti.png") contis = {"S7sh2": { "image_path": str(conti_png), "plate_path": lane["S7sh2"]["bg_path"], "error": None, "skipped_reason": None, }} classify = {"S7sh2": {"person_visible": True, "prev": None}} delta = finalize_structure_ab_entries( lane_contis=lane, contis=contis, classify_shots=classify) assert lane["S7sh2"]["status"] == "ab_select_ready" assert delta == {"applicable": 0, "completed": 0, "failed": 0} # 콘티가 다른 이미지를 참조하면 fail-closed lane2 = {"S7sh2": _seed_bg_entry(tmp_path)} contis2 = {"S7sh2": { "image_path": str(conti_png), "plate_path": str(_touch(tmp_path, "other.png")), "error": None, "skipped_reason": None, }} finalize_structure_ab_entries( lane_contis=lane2, contis=contis2, classify_shots=classify) assert lane2["S7sh2"]["status"] == "failed" assert "불일치" in lane2["S7sh2"]["error"] def test_finalize_seed_bg_bypass_semantics(tmp_path): """(e) prev→prev-only(bg 무관 bypass), bg_only→seed-bg-only, bg_source=failed 인 bg_only 는 fail-closed.""" from app.modules.pipeline.shot_conti_light import ( finalize_structure_ab_entries, ) lane = { "S7sh1": _seed_bg_entry(tmp_path), "S7sh3": _seed_bg_entry(tmp_path), "S7sh4": _seed_bg_entry( tmp_path, bg_source="failed", bg_path="", bg_reason="SAMPLE 소스 없음"), } classify = { "S7sh1": {"person_visible": True, "prev": "S7sh0"}, "S7sh3": {"person_visible": False, "prev": None}, "S7sh4": {"person_visible": False, "prev": None}, } delta = finalize_structure_ab_entries( lane_contis=lane, contis={}, classify_shots=classify) assert lane["S7sh1"]["status"] == "ab_select_bypass" assert lane["S7sh1"]["bypass_reason"] == "prev" assert lane["S7sh3"]["status"] == "ab_select_bypass" assert lane["S7sh3"]["bypass_reason"] == "bg_only" assert lane["S7sh4"]["status"] == "failed" assert delta == {"applicable": 1, "completed": 0, "failed": 1} # ── seed-bg 재리뷰 (B2·H3·N4) — fail-open 차단·typed 감사 참값 ────────── def test_bg_resolver_stale_plate_file_never_falls_to_seed(tmp_path): """(재리뷰 2) plate_map 에 key 존재+파일 결손 = failed — assign 유무 무관 seed 하강 금지 (생산 기록=손상 감지 대상).""" from app.modules.pipeline.shot_conti_light import ( resolve_structure_bg_source, ) seed = _touch(tmp_path, "seed.png") src, _, reason = resolve_structure_bg_source( shot_key="7_2", plate_map={"7_2": tmp_path / "gone.png"}, # 기록 있음, 파일 없음 assign_by_key={}, seed_path=str(seed), seed_status="ok", ) assert src == "failed" and "seed 하강 금지" in reason def test_any_status_assignment_loader_keeps_failed_groups(tmp_path): """(재리뷰 1) producer 실패(failed) 그룹의 배정 기록 보존 — ok-only 로더가 기록을 지워 '구조적 무생산' 오인(seed fail-open)되던 결함.""" import json from app.modules.pipeline.plate_select import ( load_bg_assignment_any_status, ) cp_dir = tmp_path / "P1" / "checkpoints" / "episodes" / "E1" / "background_render" cp_dir.mkdir(parents=True) (cp_dir / "manifest.json").write_text(json.dumps({ "data": {"groups": { "L05B01": {"status": "failed", "png_path": "", "shot_ids": ["S7_Shot2"]}, }}, }), encoding="utf-8") assign = load_bg_assignment_any_status(str(tmp_path), "P1", "E1") assert assign == {"7_2": "L05B01"} def test_finalize_plate_bg_path_synced_to_effective_plate(tmp_path): """(재리뷰 3) plate authority 교체 후에도 entry.bg_path 는 콘티가 실제 참조한 최종 effective plate 와 일치 (typed 감사 참값).""" from app.modules.pipeline.shot_conti_light import ( finalize_structure_ab_entries, ) lane = {"S4sh1": _pending_entry(tmp_path)} lane["S4sh1"]["bg_path"] = str(tmp_path / "old_plate.png") # 선택 전 기록 conti_png = _touch(tmp_path, "conti.png") chosen = _touch(tmp_path, "chosen_plate.png") # authority 선택본 contis = {"S4sh1": { "image_path": str(conti_png), "plate_path": str(chosen), "error": None, "skipped_reason": None, }} classify = {"S4sh1": {"person_visible": True, "prev": None}} finalize_structure_ab_entries( lane_contis=lane, contis=contis, classify_shots=classify) assert lane["S4sh1"]["status"] == "ab_select_ready" assert lane["S4sh1"]["bg_path"] == str(chosen) assert lane["S4sh1"]["bg_path"] == lane["S4sh1"]["plate_path"] def test_finalize_unknown_bg_source_rejected(tmp_path): """(재리뷰 4) enum default-deny — 허용 밖 bg_source 는 ready 로 통과 금지.""" from app.modules.pipeline.shot_conti_light import ( finalize_structure_ab_entries, ) lane = {"S4sh1": _pending_entry(tmp_path)} lane["S4sh1"]["bg_source"] = "typo" conti_png = _touch(tmp_path, "conti.png") plate_png = _touch(tmp_path, "plate.png") contis = {"S4sh1": { "image_path": str(conti_png), "plate_path": str(plate_png), "error": None, "skipped_reason": None, }} classify = {"S4sh1": {"person_visible": True, "prev": None}} delta = finalize_structure_ab_entries( lane_contis=lane, contis=contis, classify_shots=classify) assert lane["S4sh1"]["status"] == "failed" assert "허용 밖" in lane["S4sh1"]["error"] assert delta["failed"] == 1 def test_lane_marker_map_asset_registration_provenance(tmp_path): """v5 (Codex HIGH-2): 마커 맵 asset 의 prompt_used=annotate effective prompt(스케치 prompt 아님), lineage=canon 맵→마커 맵→스케치 UUID.""" from app.modules.pipeline.shot_conti_light import ( register_intermediate_assets, ) base = _touch(tmp_path, "canon_map.png") marker = _touch(tmp_path, "lane_marker_map_S3sh1.png") sketch = _touch(tmp_path, "lane_sketch_S3sh1.png") store = _FakeStore() store.by_rel[str(base)] = _FakeAsset("canon-map", str(base)) lane_contis = {"S3sh1": { "status": "ok", "image_path": str(sketch), "base_map_path": str(base), "marker_map_path": str(marker), "marker_prompt": "MARKER annotate effective prompt", "prompt": "SKETCH prompt", }} register_intermediate_assets( contis={}, map_plates={}, rel_fn=lambda p: p, find_asset_by_rel=store.find, new_asset=store.new, annotate_fn=store.annotate, conti_model="gpt-image-2", plate_model="nb2", lane_contis=lane_contis, ) marker_asset = store.by_rel[str(marker)] sketch_asset = store.by_rel[str(sketch)] assert marker_asset.prompt_used == "MARKER annotate effective prompt" assert sketch_asset.prompt_used == "SKETCH prompt" assert marker_asset.input_edge() == ["canon-map"] assert sketch_asset.input_edge() == [marker_asset.id] # 구 entry(marker_map_path 부재)=기존 lineage byte-identical lane_old = {"S9sh9": { "status": "ok", "image_path": str(sketch), "base_map_path": str(base), "prompt": "OLD sketch prompt", }} register_intermediate_assets( contis={}, map_plates={}, rel_fn=lambda p: p, find_asset_by_rel=store.find, new_asset=store.new, annotate_fn=store.annotate, conti_model="gpt-image-2", plate_model="nb2", lane_contis=lane_old, ) assert store.by_rel[str(sketch)].input_edge() == ["canon-map"]