"""★DB 경계까지 신원은 short_id (Codex BLOCK 2026-09-03 06:55 · 실측 f7cc45c576c0 3판): sync 의 pre-pass 가 겹치는 행의 short_id 를 NULL 로 만든 뒤 지도를 다시 만들어 원래 SID→행 결속을 잃었고, (type,name) dict 가 동명·동타입 둘을 한 행으로 접었다 → LP05 가 LP07 에 합쳐짐. 필수 끝점 다섯을 실제 PostgreSQL 로 잠근다. fixture 낱말은 일반어.""" from __future__ import annotations import json import uuid from datetime import datetime, timezone from pathlib import Path import pytest from app.modules.pipeline.grounding_entity_sync_ext import CHUNK_SCHEMA_MARKER, CHUNK_SCHEMA_VERSION NEUTRAL = {"location": None, "visual_identity": None} LOC_META = {"location": {"space_profile": {"kind": "single_space", "allowed_space_keys": ["main"], "default_space_key": None}}, "visual_identity": None} def _cp(tmp: Path, pid: str, eid: str, step: str, payload: dict) -> None: d = tmp / pid / "checkpoints" / "episodes" / eid / step d.mkdir(parents=True, exist_ok=True) (d / "manifest.json").write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") def _marked(data): return {"status": "completed", CHUNK_SCHEMA_MARKER: CHUNK_SCHEMA_VERSION, "data": data} def _ent(name, sid, meta): return {"name": name, "short_id": sid, "description": "설명", "visual_traits": ["가"], "t2i_prompt": "지문", "metadata_json": meta} def _t2i(parts=(("바닥", "LP05"), ("바닥", "LP07"))): return _marked({"locations": [_ent("가게", "L03", LOC_META), _ent("길", "L02", LOC_META)], "location_parts": [_ent(n, s, NEUTRAL) for n, s in parts], "characters": [], "props": []}) def _rel(): return _marked({"relations": [ {"relation_type": "part_of", "participants": [{"short_id": "LP05", "role": "part", "order": 1}, {"short_id": "L03", "role": "whole", "order": 2}]}, {"relation_type": "part_of", "participants": [{"short_id": "LP07", "role": "part", "order": 1}, {"short_id": "L02", "role": "whole", "order": 2}]}]}) @pytest.fixture def seeded(tmp_path, monkeypatch): monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) from app.core.database import SessionLocal, init_db from app.models.catalog import ProjectRegistry, UserAccount from app.models.project import Episode init_db() session = SessionLocal() try: now = datetime.now(timezone.utc).isoformat() uid = f"test-user-{uuid.uuid4()}"; pid = f"test-sid-{uuid.uuid4()}"; eid = f"test-sid-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="sid", 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="ko", status="uploaded", created_at=now, updated_at=now)); session.flush() yield session, pid, eid, tmp_path finally: session.rollback(); session.close() def _sync(db, pid, eid): from app.services.checkpoint_sync.entity_sync_service import EntitySyncService from app.services.checkpoint_sync.relation_sync_service import RelationSyncService EntitySyncService(db, pid, eid).sync_from_checkpoint(); db.flush() RelationSyncService(db, pid, eid).sync_from_checkpoint(); db.flush() def _rows(db, pid): from app.models.project import EntityCanon, RelationFact canons = db.query(EntityCanon).filter(EntityCanon.project_id == pid, EntityCanon.entity_type == "location_part").all() facts = db.query(RelationFact).filter(RelationFact.project_id == pid, RelationFact.relation_type == "part_of").all() return {c.short_id: c.id for c in canons}, len(facts) class TestTheFiveEndpoints: def test_first_sync_on_an_empty_db_creates_both_and_both_relations(self, seeded): db, pid, eid, tmp = seeded _cp(tmp, pid, eid, "entity_t2i", _t2i()); _cp(tmp, pid, eid, "entity_relation", _rel()) _sync(db, pid, eid) canons, nfacts = _rows(db, pid) assert set(canons) == {"LP05", "LP07"} and nfacts == 2 def test_a_second_sync_of_the_same_cp_is_stable(self, seeded): db, pid, eid, tmp = seeded _cp(tmp, pid, eid, "entity_t2i", _t2i()); _cp(tmp, pid, eid, "entity_relation", _rel()) _sync(db, pid, eid); before, _ = _rows(db, pid) _sync(db, pid, eid); after, nfacts = _rows(db, pid) assert before == after and len(after) == 2 and nfacts == 2 # DB id · short_id 안정 · 추가/삭제 0 def test_recovery_shape_keeps_the_existing_row_and_adds_only_the_missing_one(self, seeded): """실제 실패 복구 모양: DB 에 LP07 만 있고 CP 에 LP05+LP07.""" db, pid, eid, tmp = seeded _cp(tmp, pid, eid, "entity_t2i", _t2i(parts=(("바닥", "LP07"),))); _cp(tmp, pid, eid, "entity_relation", _marked({"relations": [ {"relation_type": "part_of", "participants": [{"short_id": "LP07", "role": "part", "order": 1}, {"short_id": "L02", "role": "whole", "order": 2}]}]})) _sync(db, pid, eid); before, _ = _rows(db, pid) assert set(before) == {"LP07"} _cp(tmp, pid, eid, "entity_t2i", _t2i()); _cp(tmp, pid, eid, "entity_relation", _rel()) _sync(db, pid, eid); after, nfacts = _rows(db, pid) assert set(after) == {"LP05", "LP07"} and after["LP07"] == before["LP07"] and nfacts == 2 def test_a_rename_under_the_same_short_id_updates_the_existing_canon(self, seeded): db, pid, eid, tmp = seeded _cp(tmp, pid, eid, "entity_t2i", _t2i()); _cp(tmp, pid, eid, "entity_relation", _rel()) _sync(db, pid, eid); before, _ = _rows(db, pid) _cp(tmp, pid, eid, "entity_t2i", _t2i(parts=(("바닥", "LP05"), ("흙길", "LP07")))) _sync(db, pid, eid); after, _ = _rows(db, pid) from app.models.project import EntityCanon assert after["LP07"] == before["LP07"] assert db.query(EntityCanon).filter(EntityCanon.project_id == pid, EntityCanon.short_id == "LP07").one().name == "흙길" def test_a_legacy_row_without_an_id_is_reused_once_and_two_of_them_fail_closed(self, seeded): from app.core.errors import AppError from app.models.project import EntityCanon db, pid, eid, tmp = seeded now = datetime.now(timezone.utc).isoformat() db.add(EntityCanon(id=str(uuid.uuid4()), project_id=pid, short_id=None, name="바닥", entity_type="location_part", description="", t2i_prompt="", stable_traits="[]", metadata_json=json.dumps(NEUTRAL), created_at=now, updated_at=now)); db.flush() _cp(tmp, pid, eid, "entity_t2i", _t2i(parts=(("바닥", "LP05"),))); _cp(tmp, pid, eid, "entity_relation", _marked({"relations": []})) _sync(db, pid, eid); canons, _ = _rows(db, pid) assert set(canons) == {"LP05"} and len(canons) == 1 # NULL-sid 단일 후보를 이어받았다 db.add(EntityCanon(id=str(uuid.uuid4()), project_id=pid, short_id=None, name="벽", entity_type="location_part", description="", t2i_prompt="", stable_traits="[]", metadata_json=json.dumps(NEUTRAL), created_at=now, updated_at=now)) db.add(EntityCanon(id=str(uuid.uuid4()), project_id=pid, short_id=None, name="벽", entity_type="location_part", description="", t2i_prompt="", stable_traits="[]", metadata_json=json.dumps(NEUTRAL), created_at=now, updated_at=now)); db.flush() _cp(tmp, pid, eid, "entity_t2i", _t2i(parts=(("바닥", "LP05"), ("벽", "LP06")))) with pytest.raises(AppError) as e: _sync(db, pid, eid) assert e.value.code == "entity_sync.identity_ambiguous" class TestTheResolverRules: def test_a_differing_name_match_is_not_a_conflict_and_a_foreign_id_row_is_not_reused(self): from app.modules.pipeline.grounding_entity_sync_ext import resolve_canon class R: def __init__(self, sid): self.short_id = sid a, b = R("LP05"), R("LP07") assert resolve_canon(short_id="LP07", entity_type="location_part", name="바닥", by_short_id={"LP05": a, "LP07": b}, by_type_name={("location_part", "바닥"): [a, b]}) is b assert resolve_canon(short_id="LP09", entity_type="location_part", name="바닥", by_short_id={"LP05": a}, by_type_name={("location_part", "바닥"): [a]}) is None free = R(None) assert resolve_canon(short_id="LP09", entity_type="location_part", name="바닥", by_short_id={}, by_type_name={("location_part", "바닥"): [free]}) is free with pytest.raises(ValueError): resolve_canon(short_id="", entity_type="location_part", name="바닥", by_short_id={}, by_type_name={("location_part", "바닥"): [a, b]})