"""entity_canon UPSERT — short_id swap 회귀 가드 (force 모드 sync blocker fix). Phase 9.2 발견: force 재실행 시 cp의 short_id 매핑이 기존 entity_canon row와 swap 형태로 다를 때 (예: row_A=L18, row_B=L20 → cp가 row_A=L20, row_B=L18 명시), 단순 UPDATE는 partial unique index `uq_entity_canon_short_id` 위반 → dispatcher abort. fix: cp의 short_id와 같은 기존 row의 short_id를 NULL로 임시 set (partial index가 NULL은 검사 제외) + flush → 새 short_id 할당. 테스트는 conftest.py가 격리한 theroad_test DB를 SessionLocal로 사용. 각 테스트는 자체 project_id로 격리(uuid)하고 마지막에 rollback. """ from __future__ import annotations import json import uuid from datetime import datetime, timezone from pathlib import Path import pytest @pytest.fixture def db_session_with_project(tmp_path, monkeypatch): """conftest.py가 격리한 theroad_test DB의 SessionLocal + 매 테스트마다 fresh user/project row seed (FK 의존성 충족) + projects_dir tmp 격리 + rollback. yield: (session, project_id, episode_id, tmp_path) Claude opus B1 가드: monkeypatch를 fixture에 통합 — settings.projects_dir이 sync 호출 전에 확실히 tmp로 가리키도록 (이전 inline 적용은 실증상 OK이나 명확성 위해 fixture로 보강). """ monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) # FK target ORM 매퍼 미리 등록 (NoReferencedTableError 회피). from app.models.catalog import UserAccount, ProjectRegistry # noqa: F401 from app.models.project import EntityCanon, Episode # noqa: F401 from app.core.database import SessionLocal, init_db init_db() # schema/indexes 보장 (idempotent) session = SessionLocal() try: now = datetime.now(timezone.utc).isoformat() uid = f"test-user-{uuid.uuid4()}" pid = f"test-swap-{uuid.uuid4()}" eid = f"test-swap-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="swap-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() yield session, pid, eid, tmp_path finally: session.rollback() session.close() def _write_cp(tmp_path: Path, pid: str, eid: str, step: str, payload: dict) -> None: cp_dir = tmp_path / pid / "checkpoints" / "episodes" / eid / step cp_dir.mkdir(parents=True, exist_ok=True) (cp_dir / "manifest.json").write_text(json.dumps(payload), encoding="utf-8") def test_entity_canon_short_id_swap_no_unique_violation(db_session_with_project): """force 재실행 시 두 row의 short_id가 swap 매핑돼도 UNIQUE VIOLATION 없이 처리. DB: row_A(name=boatyard, L18), row_B(name=market, L20) cp: boatyard→L20, market→L18 (swap) before fix: row_A.short_id=L20 시도 → UNIQUE VIOLATION (row_B의 L20 충돌) after fix: pre-pass에서 둘 다 NULL → swap 안전 """ db_session, pid, eid, tmp_path = db_session_with_project from app.models.project import EntityCanon now = datetime.now(timezone.utc).isoformat() row_a_id = str(uuid.uuid4()) row_b_id = str(uuid.uuid4()) db_session.add(EntityCanon( id=row_a_id, project_id=pid, short_id="L18", name="boatyard", entity_type="location", description="old desc A", t2i_prompt="old prompt A", stable_traits="[]", created_at=now, updated_at=now, )) db_session.add(EntityCanon( id=row_b_id, project_id=pid, short_id="L20", name="market", entity_type="location", description="old desc B", t2i_prompt="old prompt B", stable_traits="[]", created_at=now, updated_at=now, )) db_session.flush() # Area B (Task 4 / C4, 2026-05-13): metadata_json normalized shape 의무. # 본 test 는 short_id swap 메커니즘 검증 — fixture 에 normalized shape 주입. _loc_md = { "location": {"space_profile": { "kind": "single_space", "allowed_space_keys": ["main"], "default_space_key": None, }}, "visual_identity": None, } cp_payload = { "schema_version": 3, "status": "completed", "data": { "characters": [], "locations": [ {"name": "boatyard", "short_id": "L20", "description": "new A", "t2i_prompt": "new prompt A", "visual_traits": [], "metadata_json": _loc_md}, {"name": "market", "short_id": "L18", "description": "new B", "t2i_prompt": "new prompt B", "visual_traits": [], "metadata_json": _loc_md}, ], "props": [], }, } _write_cp(tmp_path, pid, eid, "entity_t2i", cp_payload) from app.services.checkpoint_sync import EntitySyncService result = EntitySyncService(db_session, pid, eid).sync_from_checkpoint() assert result["synced"] == 2 assert result["skipped"] == 0 db_session.expire_all() row_a = db_session.query(EntityCanon).filter(EntityCanon.id == row_a_id).one() row_b = db_session.query(EntityCanon).filter(EntityCanon.id == row_b_id).one() # ★뒤집음 (Codex 계약 2026-09-03 06:55 · 신원은 producer 의 short_id): 행은 제 short_id 를 지키고 **이름·내용이 따라온다**. # UNIQUE 위반 없음은 그대로(pre-pass NULL 뒤 원래 지도로 제 행을 찾아 되돌린다). assert row_a.short_id == "L18" and row_a.name == "market" and row_a.description == "new B" assert row_b.short_id == "L20" and row_b.name == "boatyard" and row_b.description == "new A" assert row_a.t2i_prompt == "new prompt B" assert row_b.t2i_prompt == "new prompt A" # M3: 모든 갱신 필드 검증 (description뿐 아니라 t2i_prompt도) def test_entity_canon_new_short_id_steals_from_orphan(db_session_with_project, caplog): """cp가 기존 row의 short_id를 다른 row에 할당 시도해도 충돌 없이 처리. DB: row_A(name=boatyard, L18), row_B(name=market, L20) cp: boatyard→L20 만 (market은 cp에서 사라짐 → orphan) fix: short_id 매칭 기준으로 NULL set → row_B의 L20도 NULL → row_A가 L20 차지. row_B는 short_id=NULL인 채로 잔존 (orphan, link cleanup 별도 단계). 또한 entity_sync_service가 NULL 잔존 row에 대해 logger.warning을 emit하는지 검증 (silent corruption 방지 가드). """ import logging db_session, pid, eid, tmp_path = db_session_with_project from app.models.project import EntityCanon now = datetime.now(timezone.utc).isoformat() row_a_id = str(uuid.uuid4()) row_b_id = str(uuid.uuid4()) db_session.add(EntityCanon( id=row_a_id, project_id=pid, short_id="L18", name="boatyard", entity_type="location", description="", t2i_prompt="", stable_traits="[]", created_at=now, updated_at=now, )) db_session.add(EntityCanon( id=row_b_id, project_id=pid, short_id="L20", name="market", entity_type="location", description="", t2i_prompt="", stable_traits="[]", created_at=now, updated_at=now, )) db_session.flush() # Area B (Task 4 / C4): metadata_json normalized shape 의무 (fixture hygiene). _loc_md = { "location": {"space_profile": { "kind": "single_space", "allowed_space_keys": ["main"], "default_space_key": None, }}, "visual_identity": None, } cp_payload = { "schema_version": 3, "status": "completed", "data": { "characters": [], "locations": [ {"name": "boatyard", "short_id": "L20", "description": "", "t2i_prompt": "", "visual_traits": [], "metadata_json": _loc_md}, ], "props": [], }, } _write_cp(tmp_path, pid, eid, "entity_t2i", cp_payload) from app.services.checkpoint_sync import EntitySyncService with caplog.at_level(logging.WARNING, logger="app.services.checkpoint_sync.entity_sync_service"): result = EntitySyncService(db_session, pid, eid).sync_from_checkpoint() assert result["synced"] == 1 db_session.expire_all() row_a = db_session.query(EntityCanon).filter(EntityCanon.id == row_a_id).one() row_b = db_session.query(EntityCanon).filter(EntityCanon.id == row_b_id).one() # ★뒤집음 (Codex 계약 2026-09-03 06:55): L20 의 정본은 row_B — 그 행이 boatyard 로 갱신되고 row_A 는 CP 에 없으니 손대지 않는다(L18). NULL 고아 없음 → 고아 경고도 없다. assert row_b.short_id == "L20" and row_b.name == "boatyard", "L20 의 정본 행이 갱신된다" assert row_a.short_id == "L18", "CP 에 없는 행은 손대지 않는다" # 사용자 가드: orphan NULL 잔존 시 logger.warning emit (silent corruption 방지) warn_msgs = [r.message for r in caplog.records if r.levelname == "WARNING"] assert not any("orphan after sync" in m for m in warn_msgs), warn_msgs def test_outlook_stale_delete_with_character_outlook_reference(db_session_with_project): """force 재실행 시 cp에서 사라진 outlook이 character_outlook 참조 중일 때 처리. DB: row_A(O01, suit) outlook + character_C + character_outlook(C, O01) cp: 다른 outlook(O02)만 있음 → O01은 stale before fix: DELETE FROM entity_canon → ForeignKeyViolation ("character_outlook_outlook_id_fkey") after fix: DELETE FROM character_outlook 먼저 → DELETE FROM entity_canon 통과 M2 Fix 2 (2026-05-01) 변경 — 이전에는 cp의 outlooks=[]로도 cleanup 분기 진입했으나 이제는 completed+empty data를 cascade로 간주하여 skip. cleanup을 검증하려면 cp에 최소 1건의 outlook이 있어야 (= sync 진입) stale row(`O01`)를 처리한다. """ db_session, pid, eid, tmp_path = db_session_with_project from app.models.project import EntityCanon, CharacterOutlook now = datetime.now(timezone.utc).isoformat() char_id = str(uuid.uuid4()) outlook_id = str(uuid.uuid4()) co_id = str(uuid.uuid4()) db_session.add(EntityCanon( id=char_id, project_id=pid, short_id="C01", name="hero", entity_type="character", description="test", created_at=now, updated_at=now, )) db_session.add(EntityCanon( id=outlook_id, project_id=pid, short_id="O01", name="suit", entity_type="outlook", description="business", created_at=now, updated_at=now, )) db_session.flush() db_session.add(CharacterOutlook( id=co_id, character_id=char_id, outlook_id=outlook_id, project_id=pid, created_at=now, )) db_session.flush() # cp에는 다른 outlook(O02)만 있고 O01은 stale. # (M2 Fix 2: completed+empty data는 cascade로 간주되어 sync skip 되므로 # cleanup 분기를 trigger하려면 cp에 최소 1건의 outlook이 있어야 함.) cp_payload = { "status": "completed", "data": { "outlooks": [ {"name": "casual", "outlook_id": "O02", "description": "casual wear"}, ], "scene_assignments": [], }, } _write_cp(tmp_path, pid, eid, "outlook_phase3", cp_payload) from app.services.checkpoint_sync import OutlookSyncService # 오류 없이 sync 통과해야 한다 (원래 이 시험이 막던 것: FK violation). result = OutlookSyncService(db_session, pid, eid).sync_from_checkpoint() assert result["outlooks"] == 1 # cp 의 O02 만 이번 화 것이다 db_session.expire_all() # ★★★2026-09-04 뒤집음. 원래 단정은 「이번 화 CP 에 없는 O01 은 canon 째 # **삭제**된다」였다. 그 삭제가 **프로젝트 범위**라서 다른 화의 아웃룩까지 # 지웠다 — 실측 da049582 에서 1화의 O03 회색외투가 그렇게 사라졌고, # O01·O02 는 1화가 만든 그림을 붙든 채 이름만 3화 것으로 바뀌었다. # 이제는 **이 화의 링크만** 풀고 canon 은 남긴다. kept_outlook = db_session.query(EntityCanon).filter( EntityCanon.id == outlook_id ).first() assert kept_outlook is not None, ( "이번 화 CP 에 없다는 이유로 아웃룩 canon 을 지웠다 — 다른 화의 자산이다") # 옷 배정도 그대로 — 다른 화가 쓰고 있을 수 있다. kept_co = db_session.query(CharacterOutlook).filter( CharacterOutlook.id == co_id ).first() assert kept_co is not None, "다른 화의 옷 배정을 지웠다" # 이 화의 링크에는 O01 이 없어야 한다 — 「이번 화 것이 아니다」는 그렇게 적는다. from app.models.project import EntityEpisodeLink linked = db_session.query(EntityEpisodeLink).filter( EntityEpisodeLink.episode_id == eid, EntityEpisodeLink.canon_id == outlook_id, ).first() assert linked is None, "이번 화 CP 에 없는 아웃룩이 이 화 링크에 남았다" # character row 는 보존 char_row = db_session.query(EntityCanon).filter( EntityCanon.id == char_id ).first() assert char_row is not None, "character row preserved" def test_outlook_canon_short_id_pre_pass_no_unique_violation(db_session_with_project): """OutlookSyncService hotfix — pre-pass NULL set이 정상 케이스에서도 깨지지 않음. Outlook sync는 short_id 우선 매칭(line 77)이라 swap intent를 코드 의미상 처리하지 못 한다 (name swap만 됨). 따라서 swap 자체 검증은 entity_canon sync에 위임하고, outlook은 hotfix가 정상 케이스에서 회귀를 유발하지 않는지만 검증. DB: row_A(O01, suit). cp: 같은 row 갱신. fix: pre-pass에서 row_A.short_id=NULL → 다시 O01 → 통과 (UNIQUE OK). """ db_session, pid, eid, tmp_path = db_session_with_project from app.models.project import EntityCanon now = datetime.now(timezone.utc).isoformat() row_a_id = str(uuid.uuid4()) db_session.add(EntityCanon( id=row_a_id, project_id=pid, short_id="O01", name="suit", entity_type="outlook", description="business", created_at=now, updated_at=now, )) db_session.flush() cp_payload = { "status": "completed", "data": { "outlooks": [ {"name": "suit", "outlook_id": "O01", "description": "business updated"}, ], "scene_assignments": [], }, } _write_cp(tmp_path, pid, eid, "outlook_phase3", cp_payload) from app.services.checkpoint_sync import OutlookSyncService result = OutlookSyncService(db_session, pid, eid).sync_from_checkpoint() # sync 통과 — UNIQUE VIOLATION 없음 assert result["outlooks"] == 1 # ★★2026-09-04 뒤집음. 이 단정은 원래 「관계 없는 아웃룩은 **지운다**」였다. # 그 삭제가 프로젝트 범위라서 **다른 화의 아웃룩까지 지웠다** (실측 # da049582: 1화 O03 회색외투 실종). 지금은 표시만 한다. # ★원래 이 시험이 막던 것 — 「swap 이 UNIQUE VIOLATION 없이 지나가는가」 — # 은 위 `outlooks == 1` 이 그대로 지킨다. from app.core.entity_identity import CANON_STATUS_ORPHANED # ★★2026-09-04 뒤집음. 원래는 「옷 배정이 없으니 고아 → **삭제**」로 # `orphans_removed == 1` 이었다. 그런데 이 아웃룩은 **이번 화 CP 가 방금 # 선언한 것**이고 sync 가 이 화 링크까지 만들어 준다. 링크가 있는 것을 # 고아로 세던 것이 결함이다 — 고아 판정은 옷 배정만이 아니라 링크·관계· # 이미지를 **다 보고** 정한다. assert result["orphans_marked"] == 0, ( "이번 화가 쓰는 아웃룩을 고아로 셌다") db_session.expire_all() kept = db_session.query(EntityCanon).filter(EntityCanon.id == row_a_id).first() assert kept is not None, "쓰이고 있는 아웃룩을 지웠다" assert kept.status != CANON_STATUS_ORPHANED, ( f"쓰이는 아웃룩에 고아 표시가 붙었다: {kept.status}")