"""Phase A verify_completion 단위 테스트용 fixture. 각 step의 verify_completion()은 DB(EntityCanon/EntityEpisodeLink/ImageAsset) + 파일 시스템(low_freq_skip JSON, ref/composite PNG 등) 양쪽을 검증한다. fixture는: - ProjectRegistry/Episode FK target row seed (UserAccount 1개도 함께) - EntityCanon character N개 + EntityEpisodeLink 연결 - low_freq_skip JSON write helper (`save_low_freq_skip_ids`) - RefImageGenStep instance 반환 + `_add_ref_asset` helper attach 설계: tests/services/test_entity_canon_short_id_swap.py의 db_session_with_project 패턴 그대로 — production DB 보호, 매 테스트 fresh project_id, 종료 시 rollback. """ from __future__ import annotations import json import uuid from datetime import datetime, timezone from pathlib import Path from typing import Iterable import pytest @pytest.fixture def seed_episode_with_chars(tmp_path, monkeypatch): """chars(short_id list) + skip(short_id list) → RefImageGenStep instance 반환. Returns: factory function (chars: List[str], skip: List[str]) → runner runner는 _add_ref_asset(short_id, file_path) helper 노출. """ monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) # FK target ORM 매퍼 사전 등록. from app.models.catalog import UserAccount, ProjectRegistry # noqa: F401 from app.models.project import ( # noqa: F401 EntityCanon, EntityEpisodeLink, Episode, ImageAsset, ) from app.core.database import SessionLocal, init_db from app.core.steps.image_steps import RefImageGenStep from app.core.low_freq_skip import save_low_freq_skip_ids init_db() session = SessionLocal() now = datetime.now(timezone.utc).isoformat() uid = f"test-user-{uuid.uuid4()}" pid = f"test-verify-{uuid.uuid4()}" eid = f"test-verify-ep-{uuid.uuid4()}" # 시드: UserAccount → ProjectRegistry → Episode (FK 순서). session.add(UserAccount( id=uid, username=f"u_{uid}", display_name="t", password_hash="x", role="creator", is_active=1, created_at=now, updated_at=now, )) session.flush() session.add(ProjectRegistry( id=pid, name="verify-test", description="", created_by=uid, created_at=now, updated_at=now, )) session.flush() session.add(Episode( id=eid, project_id=pid, episode_number=1, title="t", source_filename="f.txt", source_path="/tmp/f.txt", language="en", status="uploaded", created_at=now, updated_at=now, )) session.flush() # short_id → canon_id 매핑 (이후 fixture 안에서 재사용). short_to_canon: dict[str, str] = {} def _make(chars: Iterable[str], skip: Iterable[str] = ()) -> "RefImageGenStep": chars = list(chars) skip = list(skip) # EntityCanon (character) + EntityEpisodeLink seed. for sid in chars: cid = str(uuid.uuid4()) short_to_canon[sid] = cid session.add(EntityCanon( id=cid, project_id=pid, short_id=sid, name=f"char_{sid}", entity_type="character", description="", t2i_prompt="", stable_traits="[]", created_at=now, updated_at=now, )) session.add(EntityEpisodeLink( id=str(uuid.uuid4()), project_id=pid, canon_id=cid, episode_id=eid, source="extracted", t2i_appearance_count=0, )) session.flush() # low_freq_skip 적용 (canon_id 단위). if skip: skip_ids = [short_to_canon[sid] for sid in skip if sid in short_to_canon] save_low_freq_skip_ids(pid, eid, skip_ids) # RefImageGenStep instance — manifest 등록된 step_id이므로 정상 init 가능. runner = RefImageGenStep( step_id="ref_image_gen", project_id=pid, episode_id=eid, db=session, project_config={}, ) # ImageAsset row INSERT helper attach. def _add_ref_asset(short_id: str, file_path: str) -> str: asset_id = str(uuid.uuid4()) session.add(ImageAsset( id=asset_id, project_id=pid, asset_type="reference", entity_id=short_to_canon[short_id], episode_id=eid, file_path=file_path, is_primary=1, status="generated", created_at=now, )) session.flush() return asset_id runner._add_ref_asset = _add_ref_asset # type: ignore[attr-defined] return runner try: yield _make finally: session.rollback() session.close() @pytest.fixture def seed_episode_with_mixed_entities(tmp_path, monkeypatch): """character + prop + (선택) location/outlook 혼합 시드 → RefImageGenStep instance. A6 IMPORTANT 회귀 가드: production reference_phase1_service.py:81은 `entity_type not in ("location", "outlook")`인 모든 엔티티에 대해 ref 이미지를 생성한다 (character + prop + 기타). verify는 이 scope을 그대로 반영해야 prop ref 누락이 false-clean으로 묻히지 않는다. Returns: factory function ( chars: List[str] = (), props: List[str] = (), locations: List[str] = (), outlooks: List[str] = (), skip: List[str] = (), ) → runner runner._add_ref_asset(short_id, file_path) helper 노출 (entity_type 자동 lookup). """ monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) from app.models.catalog import UserAccount, ProjectRegistry # noqa: F401 from app.models.project import ( # noqa: F401 EntityCanon, EntityEpisodeLink, Episode, ImageAsset, ) from app.core.database import SessionLocal, init_db from app.core.steps.image_steps import RefImageGenStep from app.core.low_freq_skip import save_low_freq_skip_ids init_db() session = SessionLocal() now = datetime.now(timezone.utc).isoformat() uid = f"test-user-{uuid.uuid4()}" pid = f"test-verify-mixed-{uuid.uuid4()}" eid = f"test-verify-mixed-ep-{uuid.uuid4()}" session.add(UserAccount( id=uid, username=f"u_{uid}", display_name="t", password_hash="x", role="creator", is_active=1, created_at=now, updated_at=now, )) session.flush() session.add(ProjectRegistry( id=pid, name="verify-mixed-test", description="", created_by=uid, created_at=now, updated_at=now, )) session.flush() session.add(Episode( id=eid, project_id=pid, episode_number=1, title="t", source_filename="f.txt", source_path="/tmp/f.txt", language="en", status="uploaded", created_at=now, updated_at=now, )) session.flush() short_to_canon: dict[str, str] = {} def _make( chars: Iterable[str] = (), props: Iterable[str] = (), locations: Iterable[str] = (), outlooks: Iterable[str] = (), skip: Iterable[str] = (), ) -> "RefImageGenStep": groups = [ ("character", list(chars)), ("prop", list(props)), ("location", list(locations)), ("outlook", list(outlooks)), ] skip = list(skip) for entity_type, sids in groups: for sid in sids: cid = str(uuid.uuid4()) short_to_canon[sid] = cid session.add(EntityCanon( id=cid, project_id=pid, short_id=sid, name=f"{entity_type}_{sid}", entity_type=entity_type, description="", t2i_prompt="", stable_traits="[]", created_at=now, updated_at=now, )) session.add(EntityEpisodeLink( id=str(uuid.uuid4()), project_id=pid, canon_id=cid, episode_id=eid, source="extracted", t2i_appearance_count=0, )) session.flush() if skip: skip_ids = [short_to_canon[sid] for sid in skip if sid in short_to_canon] save_low_freq_skip_ids(pid, eid, skip_ids) runner = RefImageGenStep( step_id="ref_image_gen", project_id=pid, episode_id=eid, db=session, project_config={}, ) def _add_ref_asset(short_id: str, file_path: str) -> str: asset_id = str(uuid.uuid4()) session.add(ImageAsset( id=asset_id, project_id=pid, asset_type="reference", entity_id=short_to_canon[short_id], episode_id=eid, file_path=file_path, is_primary=1, status="generated", created_at=now, )) session.flush() return asset_id runner._add_ref_asset = _add_ref_asset # type: ignore[attr-defined] return runner try: yield _make finally: session.rollback() session.close() @pytest.fixture def seed_episode_with_combos(tmp_path, monkeypatch): """combos: List[(char_short_id, outlook_short_id)] → CompositeImageGenStep instance. char_short_id의 EntityCanon(character) + outlook_short_id의 EntityCanon(outlook) + EntityEpisodeLink(character) + CharacterOutlook(character_id, outlook_id) row seed. O00은 short_id="O00"로 표시. Returns: factory function (combos: List[Tuple[str, str]], skip: List[str]) → runner runner는 _add_composite_asset(char_sid, outlook_sid, file_path) helper 노출. """ monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) # FK target ORM 매퍼 사전 등록. from app.models.catalog import UserAccount, ProjectRegistry # noqa: F401 from app.models.project import ( # noqa: F401 EntityCanon, EntityEpisodeLink, Episode, ImageAsset, CharacterOutlook, ) from app.core.database import SessionLocal, init_db from app.core.steps.image_steps import CompositeImageGenStep from app.core.low_freq_skip import save_low_freq_skip_ids init_db() session = SessionLocal() now = datetime.now(timezone.utc).isoformat() uid = f"test-user-{uuid.uuid4()}" pid = f"test-verify-comp-{uuid.uuid4()}" eid = f"test-verify-comp-ep-{uuid.uuid4()}" # 시드: UserAccount → ProjectRegistry → Episode (FK 순서). session.add(UserAccount( id=uid, username=f"u_{uid}", display_name="t", password_hash="x", role="creator", is_active=1, created_at=now, updated_at=now, )) session.flush() session.add(ProjectRegistry( id=pid, name="verify-comp-test", description="", created_by=uid, created_at=now, updated_at=now, )) session.flush() session.add(Episode( id=eid, project_id=pid, episode_number=1, title="t", source_filename="f.txt", source_path="/tmp/f.txt", language="en", status="uploaded", created_at=now, updated_at=now, )) session.flush() # short_id → canon_id 매핑. short_to_canon: dict[str, str] = {} def _ensure_canon(sid: str, entity_type: str) -> str: """short_id → canon_id 보장 (idempotent).""" if sid in short_to_canon: return short_to_canon[sid] cid = str(uuid.uuid4()) short_to_canon[sid] = cid session.add(EntityCanon( id=cid, project_id=pid, short_id=sid, name=f"{entity_type}_{sid}", entity_type=entity_type, description="", t2i_prompt="", stable_traits="[]", created_at=now, updated_at=now, )) return cid def _make(combos: Iterable[tuple[str, str]], skip: Iterable[str] = ()) -> "CompositeImageGenStep": combos = list(combos) skip = list(skip) chars: list[str] = [] # 1단계: EntityCanon (character + outlook) 모두 등록 + flush. for char_sid, outlook_sid in combos: _ensure_canon(char_sid, "character") _ensure_canon(outlook_sid, "outlook") session.flush() # 2단계: EntityEpisodeLink (character만) + CharacterOutlook 등록. for char_sid, outlook_sid in combos: cid = short_to_canon[char_sid] oid = short_to_canon[outlook_sid] if char_sid not in chars: chars.append(char_sid) session.add(EntityEpisodeLink( id=str(uuid.uuid4()), project_id=pid, canon_id=cid, episode_id=eid, source="extracted", t2i_appearance_count=0, )) # ★배정에 **화를 적는다** (alembic 014, 2026-09-04). 안 적으면 # 이 행은 legacy(화 모름)로 취급되고, 그 갈래는 「인물·아웃룩이 # **둘 다** 이 화에 걸려 있을 때만」 인정한다 — 이 fixture 는 # 아웃룩 링크를 안 심으므로 배정이 통째로 안 보인다. # 실환경에서는 sync 가 늘 화를 적으므로 여기가 하한이다. session.add(CharacterOutlook( id=str(uuid.uuid4()), project_id=pid, character_id=cid, outlook_id=oid, episode_id=eid, created_at=now, )) session.flush() # low_freq_skip 적용 (canon_id 단위, character만 의미 있음). if skip: skip_ids = [short_to_canon[sid] for sid in skip if sid in short_to_canon] save_low_freq_skip_ids(pid, eid, skip_ids) runner = CompositeImageGenStep( step_id="composite_image_gen", project_id=pid, episode_id=eid, db=session, project_config={}, ) # ImageAsset(reference, composite:{char_uuid}:{outlook_uuid}) row INSERT helper. def _add_composite_asset(char_sid: str, outlook_sid: str, file_path: str) -> str: asset_id = str(uuid.uuid4()) char_uuid = short_to_canon[char_sid] outlook_uuid = short_to_canon[outlook_sid] session.add(ImageAsset( id=asset_id, project_id=pid, asset_type="reference", entity_id=char_uuid, episode_id=eid, file_path=file_path, prompt_used=f"[composite:{char_uuid}:{outlook_uuid}] test", is_primary=1, status="generated", created_at=now, )) session.flush() return asset_id runner._add_composite_asset = _add_composite_asset # type: ignore[attr-defined] return runner try: yield _make finally: session.rollback() session.close() @pytest.fixture def seed_episode_with_staging(tmp_path, monkeypatch): """states: List[(char_short_id, state)] → CharacterStateVariantStep instance. char_short_id의 EntityCanon(character) + EntityEpisodeLink seed + shot_staging 체크포인트 manifest.json (character_angles[].subject_state = state). state values: 'dead', 'severely_injured', 'unconscious' (subject_state 헬퍼 is_immobilized_state 게이트 통과 enum, Area #2 W2 shot_staging v13). Returns: factory function (states: List[Tuple[str, str]]) → runner runner는 _add_state_variant_asset(char_sid, state, file_path) helper 노출. """ monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) # FK target ORM 매퍼 사전 등록. from app.models.catalog import UserAccount, ProjectRegistry # noqa: F401 from app.models.project import ( # noqa: F401 EntityCanon, EntityEpisodeLink, Episode, ImageAsset, ) from app.core.database import SessionLocal, init_db from app.core.steps.image_steps import CharacterStateVariantStep from app.core.low_freq_skip import save_low_freq_skip_ids init_db() session = SessionLocal() now = datetime.now(timezone.utc).isoformat() uid = f"test-user-{uuid.uuid4()}" pid = f"test-verify-sv-{uuid.uuid4()}" eid = f"test-verify-sv-ep-{uuid.uuid4()}" # 시드: UserAccount → ProjectRegistry → Episode (FK 순서). session.add(UserAccount( id=uid, username=f"u_{uid}", display_name="t", password_hash="x", role="creator", is_active=1, created_at=now, updated_at=now, )) session.flush() session.add(ProjectRegistry( id=pid, name="verify-sv-test", description="", created_by=uid, created_at=now, updated_at=now, )) session.flush() session.add(Episode( id=eid, project_id=pid, episode_number=1, title="t", source_filename="f.txt", source_path="/tmp/f.txt", language="en", status="uploaded", created_at=now, updated_at=now, )) session.flush() # short_id → canon_id 매핑. short_to_canon: dict[str, str] = {} short_to_name: dict[str, str] = {} def _make( states: Iterable[tuple[str, str]], skip_chars: Iterable[str] = (), ) -> "CharacterStateVariantStep": states = list(states) skip_chars = list(skip_chars) # 1단계: EntityCanon (character) + EntityEpisodeLink seed. unique_chars = {char_sid for char_sid, _ in states} for char_sid in unique_chars: if char_sid in short_to_canon: continue cid = str(uuid.uuid4()) short_to_canon[char_sid] = cid char_name = f"char_{char_sid}" short_to_name[char_sid] = char_name session.add(EntityCanon( id=cid, project_id=pid, short_id=char_sid, name=char_name, entity_type="character", description="", t2i_prompt="", stable_traits="[]", created_at=now, updated_at=now, )) session.add(EntityEpisodeLink( id=str(uuid.uuid4()), project_id=pid, canon_id=cid, episode_id=eid, source="extracted", t2i_appearance_count=0, )) session.flush() # low_freq_skip 적용 (canon_id 단위) — ref_image_gen 이 base ref 를 # skip 한 registered character 시뮬레이션. character_state_variant # 의 _execute/verify expected 가 이 set 을 제외하는지 검증용. if skip_chars: skip_ids = [short_to_canon[s] for s in skip_chars if s in short_to_canon] save_low_freq_skip_ids(pid, eid, skip_ids) # 2단계: shot_staging 체크포인트 manifest.json 작성. # Area #2 W2 v13 producer 형식: {"data": {"shots": [{"character_angles": # [{"character": name, "subject_state": state, ...}]}]}} cp_dir = Path(tmp_path) / pid / "checkpoints" / "episodes" / eid / "shot_staging" cp_dir.mkdir(parents=True, exist_ok=True) shots = [] for char_sid, state in states: shots.append({ "character_angles": [ {"character": short_to_name[char_sid], "gaze_direction_kind": "closed_eyes", "subject_state": state} ] }) manifest = {"data": {"shots": shots}} (cp_dir / "manifest.json").write_text( json.dumps(manifest, ensure_ascii=False), encoding="utf-8", ) runner = CharacterStateVariantStep( step_id="character_state_variant", project_id=pid, episode_id=eid, db=session, project_config={}, ) # ImageAsset(reference, [state_variant:{char_uuid}:{state}]) row INSERT helper. def _add_state_variant_asset(char_sid: str, state: str, file_path: str) -> str: asset_id = str(uuid.uuid4()) char_uuid = short_to_canon[char_sid] session.add(ImageAsset( id=asset_id, project_id=pid, asset_type="reference", entity_id=char_uuid, episode_id=eid, file_path=file_path, prompt_used=f"[state_variant:{char_uuid}:{state}] {short_to_name[char_sid]}", is_primary=0, status="generated", created_at=now, )) session.flush() return asset_id runner._add_state_variant_asset = _add_state_variant_asset # type: ignore[attr-defined] return runner try: yield _make finally: session.rollback() session.close() @pytest.fixture def seed_episode_with_outlook_wearers(tmp_path, monkeypatch): """outlook wearer reverse-lookup verify 시나리오 fixture. A7 critical defect 회귀 가드: `outlook_expected_ids = [oid for oid in outlook_ids if oid not in skipped]` 같은 type-mismatch 비교 (outlook UUID vs character/prop UUID set)는 항상 no-op. production은 outlook의 **wearer character_id**가 low-freq면 그 outlook generation을 skip한다 (reference_phase2_service.py:122). verify는 모든 wearer가 low-freq일 때만 outlook을 expected에서 제외해야 한다. Returns: factory function ( outlooks: List[str], # 시드할 outlook short_id wearers: List[Tuple[str, str]], # (char_short, outlook_short) skip_chars: List[str] = (), # low_freq_skip 대상 character short_id characters: List[str] = (), # 추가 character short_id (link 만 시드) ) → CompositeImageGenStep runner._add_outlook_asset(short_id, file_path) helper 노출. """ monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) from app.models.catalog import UserAccount, ProjectRegistry # noqa: F401 from app.models.project import ( # noqa: F401 EntityCanon, EntityEpisodeLink, Episode, ImageAsset, CharacterOutlook, ) from app.core.database import SessionLocal, init_db from app.core.steps.image_steps import CompositeImageGenStep from app.core.low_freq_skip import save_low_freq_skip_ids init_db() session = SessionLocal() now = datetime.now(timezone.utc).isoformat() uid = f"test-user-{uuid.uuid4()}" pid = f"test-verify-ow-{uuid.uuid4()}" eid = f"test-verify-ow-ep-{uuid.uuid4()}" session.add(UserAccount( id=uid, username=f"u_{uid}", display_name="t", password_hash="x", role="creator", is_active=1, created_at=now, updated_at=now, )) session.flush() session.add(ProjectRegistry( id=pid, name="verify-ow-test", description="", created_by=uid, created_at=now, updated_at=now, )) session.flush() session.add(Episode( id=eid, project_id=pid, episode_number=1, title="t", source_filename="f.txt", source_path="/tmp/f.txt", language="en", status="uploaded", created_at=now, updated_at=now, )) session.flush() short_to_canon: dict[str, str] = {} def _ensure_canon(sid: str, entity_type: str) -> str: if sid in short_to_canon: return short_to_canon[sid] cid = str(uuid.uuid4()) short_to_canon[sid] = cid session.add(EntityCanon( id=cid, project_id=pid, short_id=sid, name=f"{entity_type}_{sid}", entity_type=entity_type, description="", t2i_prompt="", stable_traits="[]", created_at=now, updated_at=now, )) return cid def _make( outlooks: Iterable[str], wearers: Iterable[tuple[str, str]] = (), skip_chars: Iterable[str] = (), characters: Iterable[str] = (), ) -> "CompositeImageGenStep": outlooks = list(outlooks) wearers = list(wearers) skip_chars = list(skip_chars) characters = list(characters) # 1단계: EntityCanon (outlook + character) 시드. for sid in outlooks: _ensure_canon(sid, "outlook") all_chars = set(characters) | {c for c, _ in wearers} for sid in all_chars: _ensure_canon(sid, "character") session.flush() # 2단계: EntityEpisodeLink (outlook + character) — 모두 link. for sid in outlooks + sorted(all_chars): session.add(EntityEpisodeLink( id=str(uuid.uuid4()), project_id=pid, canon_id=short_to_canon[sid], episode_id=eid, source="extracted", t2i_appearance_count=0, )) session.flush() # 3단계: CharacterOutlook (wearer mapping). for char_sid, outlook_sid in wearers: session.add(CharacterOutlook( id=str(uuid.uuid4()), project_id=pid, character_id=short_to_canon[char_sid], outlook_id=short_to_canon[outlook_sid], created_at=now, )) session.flush() # low_freq_skip은 character canon_id 기준 (production 동일). if skip_chars: skip_ids = [short_to_canon[s] for s in skip_chars if s in short_to_canon] save_low_freq_skip_ids(pid, eid, skip_ids) runner = CompositeImageGenStep( step_id="composite_image_gen", project_id=pid, episode_id=eid, db=session, project_config={}, ) def _add_outlook_asset(short_id: str, file_path: str) -> str: asset_id = str(uuid.uuid4()) session.add(ImageAsset( id=asset_id, project_id=pid, asset_type="reference", entity_id=short_to_canon[short_id], episode_id=eid, file_path=file_path, is_primary=1, status="generated", created_at=now, )) session.flush() return asset_id runner._add_outlook_asset = _add_outlook_asset # type: ignore[attr-defined] return runner try: yield _make finally: session.rollback() session.close() @pytest.fixture def seed_episode_with_floor_plans(tmp_path, monkeypatch): """Phase 7 floor_plan_render verify 시나리오 fixture. Plan-vs-production 정찰 결과 mismatch 4건 반영: - asset_type='floor_plan' (NOT 'reference') - variant_type=fp_id (traceability + UPSERT 매치 키) - expected source = master_plan.data.plans[gid].plan.floor_plans[].fp_id (NOT master_plan.data.groups[].needs_floor_plan; 후자는 production에 없음) - renderable filter = floor_plan_prompt.data.floor_plans[fid].status=='ok' (실패 prompt fp는 expected에서 제외) background_mode gate는 production line 93과 동일 (off/chain_only는 step skip). test에서는 settings.background_mode 를 monkeypatch로 강제 주입. register/verify parity 시나리오 지원: - master_plan_fps tuple 의 loc_short="" → 그 fp 는 ref background 미생성 → primary_loc_id 빈값 + canon 미생성 (미사용 location floor plan 케이스). - skip_loc_canons → 그 loc_short 의 EntityCanon 만 미생성 (ref background 는 생성 → primary_loc_id 는 있으나 canon resolve 불가 케이스). Returns: factory function ( master_plan_fps: List[Tuple[loc_short_id, fp_id]] = (), prompt_status: Dict[fp_id, "ok"|"failed"] = None, # default: 모두 "ok" background_mode: str = "on", skip_loc_canons: Iterable[str] = (), # canon 미생성 loc_short ) → FloorPlanRenderStep runner._add_floor_plan_asset(loc_short, fp_id, file_path) helper 노출. runner._set_master_plan_missing() helper — master_plan cp 자체 없음 시나리오. """ monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) from app.models.catalog import UserAccount, ProjectRegistry # noqa: F401 from app.models.project import ( # noqa: F401 EntityCanon, Episode, ImageAsset, ) from app.core.database import SessionLocal, init_db from app.core.steps.floor_plan_render_step import FloorPlanRenderStep init_db() session = SessionLocal() now = datetime.now(timezone.utc).isoformat() uid = f"test-user-{uuid.uuid4()}" pid = f"test-verify-fp-{uuid.uuid4()}" eid = f"test-verify-fp-ep-{uuid.uuid4()}" session.add(UserAccount( id=uid, username=f"u_{uid}", display_name="t", password_hash="x", role="creator", is_active=1, created_at=now, updated_at=now, )) session.flush() session.add(ProjectRegistry( id=pid, name="verify-fp-test", description="", created_by=uid, created_at=now, updated_at=now, )) session.flush() session.add(Episode( id=eid, project_id=pid, episode_number=1, title="t", source_filename="f.txt", source_path="/tmp/f.txt", language="en", status="uploaded", created_at=now, updated_at=now, )) session.flush() short_to_canon: dict[str, str] = {} def _ensure_loc_canon(sid: str) -> str: if sid in short_to_canon: return short_to_canon[sid] cid = str(uuid.uuid4()) short_to_canon[sid] = cid session.add(EntityCanon( id=cid, project_id=pid, short_id=sid, name=f"loc_{sid}", entity_type="location", description="", t2i_prompt="", stable_traits="[]", created_at=now, updated_at=now, )) return cid def _make( master_plan_fps: Iterable[tuple[str, str]] = (), prompt_status: dict[str, str] | None = None, background_mode: str = "on", skip_loc_canons: Iterable[str] = (), ) -> "FloorPlanRenderStep": master_plan_fps = list(master_plan_fps) skip_loc_canons = set(skip_loc_canons) if prompt_status is None: prompt_status = {fid: "ok" for _, fid in master_plan_fps} # 1) location EntityCanon 시드 — loc_short 비어있지 않고 skip 대상 아닌 것만. # 빈 loc_short → ref bg 미생성(primary_loc 빈값) + canon 미생성. # skip_loc_canons → primary_loc 은 있으나 canon resolve 불가 케이스. for loc_short, _fid in master_plan_fps: if loc_short and loc_short not in skip_loc_canons: _ensure_loc_canon(loc_short) session.flush() # 2) background_master_plan 체크포인트 작성 # production shape: data.plans[group_id].{status, plan: {floor_plans[], backgrounds[]}} # master_plan_fps의 (loc_short, fp_id) 쌍을 단일 group "g1"에 묶음. cp_root = Path(tmp_path) / pid / "checkpoints" / "episodes" / eid master_dir = cp_root / "background_master_plan" master_dir.mkdir(parents=True, exist_ok=True) floor_plan_specs = [] background_specs = [] for loc_short, fid in master_plan_fps: floor_plan_specs.append({ "fp_id": fid, "sub_location": (loc_short or "").lower(), "depends_on_fp": [], }) # 같은 sub_location을 ref하는 가상 bg → primary_loc_id 결정 신호. # 빈 loc_short → ref bg 미생성 → 그 fp 의 primary_loc_id 는 빈값. if loc_short: background_specs.append({ "bg_id": f"bg_{fid}", "loc_id": loc_short, "depends_on_fp": [fid], }) master_manifest = { "data": { "plans": { "g1": { "status": "ok", "plan": { "floor_plans": floor_plan_specs, "backgrounds": background_specs, }, } } } } (master_dir / "manifest.json").write_text( json.dumps(master_manifest, ensure_ascii=False), encoding="utf-8", ) # 3) floor_plan_prompt 체크포인트 작성 (renderable filter용) prompt_dir = cp_root / "floor_plan_prompt" prompt_dir.mkdir(parents=True, exist_ok=True) floor_plans_map = {} for _loc, fid in master_plan_fps: st = prompt_status.get(fid, "ok") if st == "ok": floor_plans_map[fid] = { "status": "ok", "t2i_prompt": f"prompt for {fid}", } else: floor_plans_map[fid] = { "status": "failed", "error": "stub failure", } prompt_manifest = {"data": {"floor_plans": floor_plans_map}} (prompt_dir / "manifest.json").write_text( json.dumps(prompt_manifest, ensure_ascii=False), encoding="utf-8", ) # 4) settings.background_mode monkeypatch monkeypatch.setattr( "app.core.config.settings.background_mode", background_mode, ) runner = FloorPlanRenderStep( step_id="floor_plan_render", project_id=pid, episode_id=eid, db=session, project_config={}, ) # 5) ImageAsset(asset_type='floor_plan') row INSERT helper. # entity_id = location canon_id, variant_type = fp_id (UPSERT 매치 키 동치). def _add_floor_plan_asset(loc_short: str, fp_id: str, file_path: str) -> str: asset_id = str(uuid.uuid4()) session.add(ImageAsset( id=asset_id, project_id=pid, asset_type="floor_plan", entity_id=short_to_canon[loc_short], episode_id=eid, file_path=file_path, variant_index=0, variant_label="v00", variant_type=fp_id, is_primary=1, status="generated", generation_model="gpt-image-2", created_at=now, )) session.flush() return asset_id def _set_master_plan_missing() -> None: """master_plan 체크포인트 자체 제거 (no-master-plan 시나리오).""" mp_path = master_dir / "manifest.json" if mp_path.exists(): mp_path.unlink() runner._add_floor_plan_asset = _add_floor_plan_asset # type: ignore[attr-defined] runner._set_master_plan_missing = _set_master_plan_missing # type: ignore[attr-defined] return runner try: yield _make finally: session.rollback() session.close() @pytest.fixture def seed_episode_with_outlooks(tmp_path, monkeypatch): """outlooks(short_id list) + skip(short_id list) → CompositeImageGenStep instance. outlook_short_id의 EntityCanon(outlook) + EntityEpisodeLink(outlook) seed. O00은 short_id="O00"로 표시 (verify에서 expected에서 제외). skip 항목은 low_freq_skip JSON에 기록 (canon_id 단위). Composite step의 verify_completion이 outlook standalone(Phase 2)도 함께 검증하므로, 본 fixture는 character는 시드하지 않고 outlook만 준비한다 → composite_expected=0 / outlook_expected=N. Returns: factory function (outlooks: List[str], skip: List[str]) → runner runner는 _add_outlook_asset(short_id, file_path) helper 노출. """ monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) # FK target ORM 매퍼 사전 등록. from app.models.catalog import UserAccount, ProjectRegistry # noqa: F401 from app.models.project import ( # noqa: F401 EntityCanon, EntityEpisodeLink, Episode, ImageAsset, ) from app.core.database import SessionLocal, init_db from app.core.steps.image_steps import CompositeImageGenStep from app.core.low_freq_skip import save_low_freq_skip_ids init_db() session = SessionLocal() now = datetime.now(timezone.utc).isoformat() uid = f"test-user-{uuid.uuid4()}" pid = f"test-verify-os-{uuid.uuid4()}" eid = f"test-verify-os-ep-{uuid.uuid4()}" # 시드: UserAccount → ProjectRegistry → Episode (FK 순서). session.add(UserAccount( id=uid, username=f"u_{uid}", display_name="t", password_hash="x", role="creator", is_active=1, created_at=now, updated_at=now, )) session.flush() session.add(ProjectRegistry( id=pid, name="verify-os-test", description="", created_by=uid, created_at=now, updated_at=now, )) session.flush() session.add(Episode( id=eid, project_id=pid, episode_number=1, title="t", source_filename="f.txt", source_path="/tmp/f.txt", language="en", status="uploaded", created_at=now, updated_at=now, )) session.flush() # short_id → canon_id 매핑. short_to_canon: dict[str, str] = {} def _make(outlooks: Iterable[str], skip: Iterable[str] = ()) -> "CompositeImageGenStep": outlooks = list(outlooks) skip = list(skip) # EntityCanon (outlook) + EntityEpisodeLink seed. for sid in outlooks: cid = str(uuid.uuid4()) short_to_canon[sid] = cid session.add(EntityCanon( id=cid, project_id=pid, short_id=sid, name=f"outlook_{sid}", entity_type="outlook", description="", t2i_prompt="", stable_traits="[]", created_at=now, updated_at=now, )) session.add(EntityEpisodeLink( id=str(uuid.uuid4()), project_id=pid, canon_id=cid, episode_id=eid, source="extracted", t2i_appearance_count=0, )) session.flush() # low_freq_skip 적용 (canon_id 단위). if skip: skip_ids = [short_to_canon[sid] for sid in skip if sid in short_to_canon] save_low_freq_skip_ids(pid, eid, skip_ids) runner = CompositeImageGenStep( step_id="composite_image_gen", project_id=pid, episode_id=eid, db=session, project_config={}, ) # ImageAsset(reference, entity_id=outlook_uuid) row INSERT helper. def _add_outlook_asset(short_id: str, file_path: str) -> str: asset_id = str(uuid.uuid4()) session.add(ImageAsset( id=asset_id, project_id=pid, asset_type="reference", entity_id=short_to_canon[short_id], episode_id=eid, file_path=file_path, is_primary=1, status="generated", created_at=now, )) session.flush() return asset_id runner._add_outlook_asset = _add_outlook_asset # type: ignore[attr-defined] return runner try: yield _make finally: session.rollback() session.close() @pytest.fixture def seed_episode_with_chain_bg_planner(tmp_path, monkeypatch): """Phase 5+ planner-driven chain_bg verify 시나리오 fixture. background_planner 체크포인트(`data.chain_bg_order`)에 group_id 리스트를 기록하고 location EntityCanon + EntityEpisodeLink + 선택적 ImageAsset을 시드한다. plan-vs-production 정찰 결과 mismatch 4건 반영 (plan §4.4 Task 23): - asset_type='chain_bg' (NOT 'reference') - traceability = variant_type=group_id (NOT prompt_used regex) - expected source = background_planner.data.chain_bg_order (NOT background_chain_planning.data.groups; 후자는 chain plan용) - 추가 위치는 background_chain_render_step.py에 직접 (NOT image_steps.py) Returns: factory function ( chain_order: List[str], # group_id 순서 location_short_id: str = "L01", # 모든 group이 매핑될 location ) → BackgroundChainRenderStep runner._add_chain_bg_asset(group_id, file_path, location_short_id="L01") helper 노출. """ monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) from app.models.catalog import UserAccount, ProjectRegistry # noqa: F401 from app.models.project import ( # noqa: F401 EntityCanon, EntityEpisodeLink, Episode, ImageAsset, ) from app.core.database import SessionLocal, init_db from app.core.steps.background_chain_render_step import BackgroundChainRenderStep init_db() session = SessionLocal() now = datetime.now(timezone.utc).isoformat() uid = f"test-user-{uuid.uuid4()}" pid = f"test-verify-cbgp-{uuid.uuid4()}" eid = f"test-verify-cbgp-ep-{uuid.uuid4()}" session.add(UserAccount( id=uid, username=f"u_{uid}", display_name="t", password_hash="x", role="creator", is_active=1, created_at=now, updated_at=now, )) session.flush() session.add(ProjectRegistry( id=pid, name="verify-cbgp-test", description="", created_by=uid, created_at=now, updated_at=now, )) session.flush() session.add(Episode( id=eid, project_id=pid, episode_number=1, title="t", source_filename="f.txt", source_path="/tmp/f.txt", language="en", status="uploaded", created_at=now, updated_at=now, )) session.flush() short_to_canon: dict[str, str] = {} def _ensure_location(short_id: str) -> str: if short_id in short_to_canon: return short_to_canon[short_id] cid = str(uuid.uuid4()) short_to_canon[short_id] = cid session.add(EntityCanon( id=cid, project_id=pid, short_id=short_id, name=f"location_{short_id}", entity_type="location", description="", t2i_prompt="", stable_traits="[]", created_at=now, updated_at=now, )) session.add(EntityEpisodeLink( id=str(uuid.uuid4()), project_id=pid, canon_id=cid, episode_id=eid, source="extracted", t2i_appearance_count=0, )) return cid def _make( chain_order: Iterable[str], location_short_id: str = "L01", ) -> "BackgroundChainRenderStep": chain_order = list(chain_order) _ensure_location(location_short_id) session.flush() # background_planner 체크포인트 작성 (production 진실: data.chain_bg_order). cp_dir = ( Path(tmp_path) / pid / "checkpoints" / "episodes" / eid / "background_planner" ) cp_dir.mkdir(parents=True, exist_ok=True) manifest = {"data": {"chain_bg_order": chain_order, "floor_plans": []}} (cp_dir / "manifest.json").write_text( json.dumps(manifest, ensure_ascii=False), encoding="utf-8", ) runner = BackgroundChainRenderStep( step_id="background_chain_render", project_id=pid, episode_id=eid, db=session, project_config={}, ) def _add_chain_bg_asset( group_id: str, file_path: str, location_short_id: str = "L01", variant_index: int = 1, ) -> str: """planner UPSERT 패턴 mirror — asset_type='chain_bg' + variant_type=group_id.""" asset_id = str(uuid.uuid4()) session.add(ImageAsset( id=asset_id, project_id=pid, asset_type="chain_bg", entity_id=short_to_canon[location_short_id], episode_id=eid, file_path=file_path, variant_index=variant_index, variant_label=f"v{variant_index:02d}", variant_type=group_id, is_primary=0, status="generated", created_at=now, )) session.flush() return asset_id runner._add_chain_bg_asset = _add_chain_bg_asset # type: ignore[attr-defined] return runner try: yield _make finally: session.rollback() session.close() @pytest.fixture def seed_episode_with_chain_bg_legacy(tmp_path, monkeypatch): """Phase 4 legacy chain_bg verify 시나리오 fixture (planner cp 없음 + node 단위). background_chain_planning 체크포인트(`data.locations[loc_id].nodes[]`)에 노드 ID 리스트를 기록하고 location EntityCanon + 선택적 ImageAsset(asset_type= 'background_chain_node') 시드한다. plan-vs-production 정찰 결과 (plan §4.4 Task 23): plan은 `chain_plan.nodes`로 가정했으나 production background_chain_planning.py:663은 `nodes`를 location dict 직속으로 둔다. Returns: factory function ( nodes_by_location: Dict[str, List[str]], # loc_short_id → [node_id, ...] write_planner_cp: bool = False, # 빈 chain_bg_order로 planner cp 작성 ) → BackgroundChainRenderStep runner._add_chain_node_asset(node_id, file_path, location_short_id) helper 노출. """ monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) from app.models.catalog import UserAccount, ProjectRegistry # noqa: F401 from app.models.project import ( # noqa: F401 EntityCanon, EntityEpisodeLink, Episode, ImageAsset, ) from app.core.database import SessionLocal, init_db from app.core.steps.background_chain_render_step import BackgroundChainRenderStep init_db() session = SessionLocal() now = datetime.now(timezone.utc).isoformat() uid = f"test-user-{uuid.uuid4()}" pid = f"test-verify-cbgl-{uuid.uuid4()}" eid = f"test-verify-cbgl-ep-{uuid.uuid4()}" session.add(UserAccount( id=uid, username=f"u_{uid}", display_name="t", password_hash="x", role="creator", is_active=1, created_at=now, updated_at=now, )) session.flush() session.add(ProjectRegistry( id=pid, name="verify-cbgl-test", description="", created_by=uid, created_at=now, updated_at=now, )) session.flush() session.add(Episode( id=eid, project_id=pid, episode_number=1, title="t", source_filename="f.txt", source_path="/tmp/f.txt", language="en", status="uploaded", created_at=now, updated_at=now, )) session.flush() short_to_canon: dict[str, str] = {} def _ensure_location(short_id: str) -> str: if short_id in short_to_canon: return short_to_canon[short_id] cid = str(uuid.uuid4()) short_to_canon[short_id] = cid session.add(EntityCanon( id=cid, project_id=pid, short_id=short_id, name=f"location_{short_id}", entity_type="location", description="", t2i_prompt="", stable_traits="[]", created_at=now, updated_at=now, )) session.add(EntityEpisodeLink( id=str(uuid.uuid4()), project_id=pid, canon_id=cid, episode_id=eid, source="extracted", t2i_appearance_count=0, )) return cid def _make( nodes_by_location: dict | None = None, write_planner_cp: bool = False, ) -> "BackgroundChainRenderStep": """nodes_by_location=None → planning cp 자체를 만들지 않음 (no_data 케이스). write_planner_cp=True인 경우 빈 chain_bg_order로 planner cp를 만들어, production _load_planner_chain_order가 None을 반환해 legacy fallback 진입하는 분기를 방어적으로 검증한다. """ nodes_by_location = nodes_by_location or {} # location seed (모든 등장한 short_id). for loc_sid in nodes_by_location.keys(): _ensure_location(loc_sid) session.flush() # background_chain_planning 체크포인트 작성 (production 진실: # data.locations[loc_id].nodes[]). if nodes_by_location: cp_dir = ( Path(tmp_path) / pid / "checkpoints" / "episodes" / eid / "background_chain_planning" ) cp_dir.mkdir(parents=True, exist_ok=True) locations_payload = {} for loc_sid, node_ids in nodes_by_location.items(): locations_payload[loc_sid] = { "location_id": loc_sid, "nodes": [{"id": nid} for nid in node_ids], "status": "ok", } manifest = {"data": {"locations": locations_payload}} (cp_dir / "manifest.json").write_text( json.dumps(manifest, ensure_ascii=False), encoding="utf-8", ) if write_planner_cp: # 빈 chain_bg_order → planner path가 None을 반환해 legacy로 fallback. cp_dir2 = ( Path(tmp_path) / pid / "checkpoints" / "episodes" / eid / "background_planner" ) cp_dir2.mkdir(parents=True, exist_ok=True) manifest2 = {"data": {"chain_bg_order": []}} (cp_dir2 / "manifest.json").write_text( json.dumps(manifest2, ensure_ascii=False), encoding="utf-8", ) runner = BackgroundChainRenderStep( step_id="background_chain_render", project_id=pid, episode_id=eid, db=session, project_config={}, ) def _add_chain_node_asset( node_id: str, file_path: str, location_short_id: str, ) -> str: """legacy UPSERT 패턴 mirror — asset_type='background_chain_node' + variant_type=node_id.""" asset_id = str(uuid.uuid4()) session.add(ImageAsset( id=asset_id, project_id=pid, asset_type="background_chain_node", entity_id=short_to_canon[location_short_id], episode_id=eid, file_path=file_path, variant_type=node_id, is_primary=0, status="generated", created_at=now, )) session.flush() return asset_id runner._add_chain_node_asset = _add_chain_node_asset # type: ignore[attr-defined] return runner try: yield _make finally: session.rollback() session.close()