"""canonical ref(AI 참조 이미지)의 대상은 **집행 계약의 정책 갈래**(character·prop)뿐이다. ★Codex BLOCK 1 (2026-09-02): 계약은 location_part → background_sidecar 하나라고 선언하는데, readiness·게이트·생산자는 `entity_type not in ("location","outlook")` 로 적어 LP 가 base ref 대상에 들었고, 최종 ref map 은 location 만 빼서 LP 참조 자산이 있으면 실사 sidecar 와 함께 실렸다. 집행자는 한 갈래에 **하나**다. 세 소비자(게이트·readiness·최종 map)를 **끝점**에서 잰다 — 실제 PostgreSQL 위에 다섯 갈래를 다 심고, 옛 LP 참조 자산이 있어도 새지 않는 음성 대조까지. 생산자(`ReferencePhase1Service`)는 `test_the_verified_photo_reaches_canonical_ref_generation` 이 production 입구로 잰다. """ from __future__ import annotations import uuid from pathlib import Path import pytest from sqlalchemy import text as sql_text from app.modules.pipeline import grounding_entity_contract as gc pytestmark = pytest.mark.pg # ── 계약 ────────────────────────────────────────────────────────────── class TestTheContractNamesOneExecutorPerOwner: def test_canonical_owners_are_the_policy_owners(self): assert gc.canonical_ref_owner_types() == gc.owners_enforced_by(gc.ENFORCE_BY_POLICY) assert set(gc.canonical_ref_owner_types()) == {"character", "prop"} def test_sidecar_owners_are_disjoint_from_canonical(self): assert set(gc.sidecar_owner_types()) == {"location", "location_part"} assert not set(gc.sidecar_owner_types()) & set(gc.canonical_ref_owner_types()) def test_every_materializable_owner_has_exactly_one_executor(self): buckets = [set(gc.owners_enforced_by(g)) for g in gc.ENFORCEMENT_GATES] assert sum(len(b) for b in buckets) == len(gc.MATERIALIZABLE_OWNER_TYPES) assert set().union(*buckets) == set(gc.MATERIALIZABLE_OWNER_TYPES) # ── PostgreSQL 위의 세 소비자 ───────────────────────────────────────── def _seed(db, pid: str, eid: str) -> None: uid = f"one-exec-{uuid.uuid4()}" db.execute(sql_text( "INSERT INTO user_account (id, username, display_name, password_hash, " "role, is_active, created_at, updated_at) VALUES " "(:uid, :un, 't', 'x', 'creator', 1, '2026-01-01', '2026-01-01')" ), {"uid": uid, "un": f"u_{uid}"}) db.execute(sql_text( "INSERT INTO project_registry (id, name, created_by, created_at, updated_at) " "VALUES (:pid, 'one-exec', :uid, '2026-01-01', '2026-01-01')" ), {"pid": pid, "uid": uid}) db.execute(sql_text( "INSERT INTO episode (id, project_id, title, episode_number, source_filename, " "source_path, status, created_at, updated_at) VALUES " "(:eid, :pid, '1화', 1, 'x.txt', 'x/x.txt', 'analyzed', '2026-01-01', '2026-01-01')" ), {"eid": eid, "pid": pid}) def _canon(db, pid: str, short_id: str, etype: str) -> str: cid = str(uuid.uuid4()) db.execute(sql_text( "INSERT INTO entity_canon (id, project_id, short_id, name, entity_type, " "description, stable_traits, metadata_json, t2i_prompt, status, created_at, " "updated_at) VALUES (:cid, :pid, :s, :n, :e, '', '[]', '{}', '', 'active', " "'2026-01-01', '2026-01-01')" ), {"cid": cid, "pid": pid, "s": short_id, "n": f"{etype} {short_id}", "e": etype}) return cid def _link(db, pid: str, eid: str, cid: str) -> None: db.execute(sql_text( "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id, " "t2i_appearance_count) VALUES (:lid, :cid, :pid, :eid, 5)" ), {"lid": str(uuid.uuid4()), "cid": cid, "pid": pid, "eid": eid}) def _ref_asset(db, pid: str, eid: str, cid: str, rel: str) -> str: aid = str(uuid.uuid4()) db.execute(sql_text( "INSERT INTO image_asset (id, project_id, episode_id, entity_id, asset_type, " "file_path, is_primary, status, created_at) VALUES " "(:aid, :pid, :eid, :cid, 'reference', :fp, 1, 'ok', '2026-01-01')" ), {"aid": aid, "pid": pid, "eid": eid, "cid": cid, "fp": rel}) return aid @pytest.fixture def five_owners(pg_session, tmp_path: Path, monkeypatch): """다섯 갈래가 한 에피소드에 다 나온다 — 그리고 **옛 LP 참조 자산**이 하나 있다.""" monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path / "projects")) monkeypatch.setattr("app.core.file_paths.PROJECT_ROOT", tmp_path) pid, eid = f"p-{uuid.uuid4()}", str(uuid.uuid4()) _seed(pg_session, pid, eid) ids = {} for short_id, etype in (("C01", "character"), ("P01", "prop"), ("L01", "location"), ("LP01", "location_part"), ("O01", "outlook")): ids[short_id] = _canon(pg_session, pid, short_id, etype) _link(pg_session, pid, eid, ids[short_id]) # ★음성 대조 재료 — 지난 판이 만들어 둔 LP canonical ref 가 있다고 치자 for short_id in ("C01", "P01", "LP01"): rel = f"refs/{short_id}.png" f = tmp_path / rel f.parent.mkdir(parents=True, exist_ok=True) f.write_bytes(b"\x89PNG " + short_id.encode()) _ref_asset(pg_session, pid, eid, ids[short_id], rel) pg_session.commit() return pid, eid, ids class TestTheGateAndReadinessCountOnlyPolicyOwners: def test_the_gate_targets_are_character_and_prop_only(self, pg_session, five_owners): from app.core.pipeline_gate import episode_reference_entities, reference_targets pid, eid, ids = five_owners got = {e.short_id for e in episode_reference_entities(pg_session, pid, eid)} assert got == {"C01", "P01"}, got assert {e.short_id for e in reference_targets(pg_session, pid, eid)} == {"C01", "P01"} def test_nothing_is_missing_and_the_lp_is_not_in_the_denominator(self, pg_session, five_owners): from app.core.pipeline_gate import get_pipeline_status, missing_reference_entities pid, eid, _ = five_owners assert missing_reference_entities(pg_session, pid, eid) == [] st = get_pipeline_status(pg_session, pid, eid)["reference_images"] assert st["entity_total"] == 2 and st["complete"] is True, st def test_readiness_expects_character_and_prop_only(self, pg_session, five_owners): from app.core.asset_readiness import compute_episode_asset_readiness pid, eid, ids = five_owners card = compute_episode_asset_readiness(pg_session, pid, eid) expected = {x for x in card.expected_ids if x.startswith("reference:")} assert expected == {f"reference:{ids['C01']}", f"reference:{ids['P01']}"}, expected def test_the_three_consumers_agree_with_the_contract(self, pg_session, five_owners): """★세 소비자의 owner 집합 == 계약 — 끝점에서 같은 답.""" from app.core.asset_readiness import compute_episode_asset_readiness from app.core.pipeline_gate import episode_reference_entities from app.models.project import EntityCanon pid, eid, ids = five_owners by_id = {c.id: c.entity_type for c in pg_session.query(EntityCanon).filter(EntityCanon.project_id == pid).all()} gate = {e.entity_type for e in episode_reference_entities(pg_session, pid, eid)} card = compute_episode_asset_readiness(pg_session, pid, eid) ready = {by_id[x.split(":", 1)[1]] for x in card.expected_ids if x.startswith("reference:")} assert gate == ready == set(gc.canonical_ref_owner_types()), (gate, ready) class TestTheFinalRefMapDoesNotLeakSidecarOwners: def test_an_old_lp_reference_asset_does_not_reach_the_scene(self, pg_session, five_owners): """★음성 대조 — LP 참조 자산이 DB 에 있어도 최종 입력에 안 실린다.""" from app.services.scene_reference_service import SceneReferenceService pid, eid, ids = five_owners svc = SceneReferenceService(pg_session, pid) visible = svc.get_visible_entities( '[{"id": "%s"}, {"id": "%s"}, {"id": "%s"}, {"id": "%s"}]' % (ids["C01"], ids["P01"], ids["LP01"], ids["L01"])) assert {v["short_id"] for v in visible} == {"C01", "P01", "LP01", "L01"} everything = svc.get_reference_image_map(visible) assert ids["LP01"] in everything, "★재료가 있어야 음성 대조다 — LP 자산이 안 읽혔다" got = svc.get_ref_image_map_excluding_locations(visible) assert set(got) == {ids["C01"], ids["P01"]}, set(got) assert ids["LP01"] not in got and ids["L01"] not in got def test_unknown_or_missing_type_is_left_alone(self, pg_session, five_owners, monkeypatch): """종류 미상은 종전처럼 남는다 — 넓히지 않는다(기존 계약 보존).""" from app.services.scene_reference_service import SceneReferenceService pid, _, _ = five_owners svc = SceneReferenceService(pg_session, pid) monkeypatch.setattr(svc, "get_reference_image_map", lambda ve: {"x1": b"A", "lp": b"B"}) got = svc.get_ref_image_map_excluding_locations( [{"id": "x1"}, {"id": "lp", "entity_type": "location_part"}]) assert got == {"x1": b"A"}