"""화 범위로 바꾸면서 **끊긴 배선** 둘 (2026-09-04 실측). ## 왜 이 파일이 따로 있나 「이 화의 것만 본다」로 바꾸려면 소비자 스무 곳이 정본 함수를 불러야 한다. 그 과정에서 두 곳이 **부르기만 하고 배선을 안 했다.** 둘 다 조용히 죽는 자리라 시험이 없으면 아무도 모른다: 1. `pipeline_gate.get_pipeline_status` 가 `active_episode_canon_ids` 를 쓰는데 **그 함수 안에 import 가 없다** — 같은 파일의 다른 두 함수에만 있었다. → `NameError`. UI 파이프라인 상태 조회가 통째로 죽는다. 2. `scene_reference_service.build_entity_text_map` 이 `episode_id` 를 쓰는데 **파라미터에 없다** — 역시 `NameError` 인데, 이 자리는 `except Exception` 안이라 **경고 한 줄 찍고 넘어간다.** 그러면 `C##O##` 합성 키가 하나도 안 만들어지고 씬 프롬프트가 조용히 나빠진다. ★두 결함 다 **리그레션이 잡았다.** 「고쳤다」가 아니라 「도는가」를 물어야 나오는 부류다. Lane: ``-m pg`` (1번은 실제 세션이 필요하다). """ from __future__ import annotations import ast import pathlib import uuid import pytest from sqlalchemy import text as sql_text pytestmark = pytest.mark.pg BACKEND = pathlib.Path(__file__).resolve().parents[2] def test_파이프라인_상태_조회가_이름_오류로_안_죽는다(pg_session): """★`get_pipeline_status` 가 `active_episode_canon_ids` 를 쓰는데 그 함수 안에 import 가 없으면 **부르는 순간** NameError 다. """ from app.core.pipeline_gate import get_pipeline_status pid, eid = f"p-{uuid.uuid4()}", str(uuid.uuid4()) uid = f"u-{uuid.uuid4()}" pg_session.execute(sql_text( "INSERT INTO user_account (id, username, display_name, password_hash, " "role, is_active, created_at, updated_at) VALUES " "(:u, :u, 't', 'x', 'creator', 1, 'x', 'x')"), {"u": uid}) pg_session.execute(sql_text( "INSERT INTO project_registry (id, name, created_by, created_at, " "updated_at) VALUES (:p, 'wiring', :u, 'x', 'x')"), {"p": pid, "u": uid}) pg_session.execute(sql_text( "INSERT INTO episode (id, project_id, title, episode_number, " "source_filename, source_path, status, created_at, updated_at) VALUES " "(:e, :p, '1화', 1, 'x', 'x', 'analyzed', 'x', 'x')"), {"e": eid, "p": pid}) cid = str(uuid.uuid4()) pg_session.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 " "(:c, :p, 'C01', '민수', 'character', '', '[]', " "'{\"location\": null, \"visual_identity\": null}', 't', 'active', " "'x', 'x')"), {"c": cid, "p": pid}) pg_session.execute(sql_text( "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id) " "VALUES (:l, :c, :p, :e)"), {"l": str(uuid.uuid4()), "c": cid, "p": pid, "e": eid}) pg_session.commit() status = get_pipeline_status(pg_session, pid, eid) # ★여기서 죽었다 assert isinstance(status, dict) and status, "상태를 아예 못 냈다" def _call_keywords(path: str, callee: str) -> list: """`path` 안에서 `.(...)` 호출들이 쓴 **키워드 이름 목록**. ★인자 **개수**로 재면 안 된다 — 다른 인자가 하나 늘어도 통과한다. 무엇을 넘겼는지를 이름으로 본다. """ tree = ast.parse((BACKEND / path).read_text(encoding="utf-8")) out = [] for n in ast.walk(tree): if (isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and n.func.attr == callee): out.append([k.arg for k in n.keywords]) return out def test_합성_키_빌더에_화가_실제로_넘어간다(): """★서명에 칸이 생겨도 **부르는 쪽이 안 넘기면** 아무것도 안 달라진다. `episode_id` 는 기본값이 `None` 이라 안 넘겨도 조용히 돈다 — 그리고 `C##O##` 키가 하나도 안 나온다. 그래서 **호출부**를 잠근다. """ import inspect from app.services.scene_reference_service import SceneReferenceService sig = inspect.signature(SceneReferenceService.build_entity_text_map) par = sig.parameters.get("episode_id") assert par is not None, f"서명에 화가 없다: {list(sig.parameters)}" # ★기본값이 있으면 안 넘겨도 조용히 돌고, 그때 exact 조회가 # `episode_id IS NULL` 이 되어 legacy 행을 돌리거나 빈 배열이 된다. assert par.default is inspect.Parameter.empty, ( "화에 기본값이 있다 — 안 넘겨도 조용히 도는 우회가 열려 있다") assert par.kind is inspect.Parameter.KEYWORD_ONLY, ( "화가 키워드 전용이 아니다 — 자리로 넘기면 다음 사람이 순서를 틀린다") for path in ("app/services/scene_image_service.py", "app/services/scene_generation_coordinator.py"): calls = _call_keywords(path, "build_entity_text_map") assert calls, f"{path}: 호출이 아예 없다 — 이 시험이 낡았다" assert all("episode_id" in kw for kw in calls), ( f"{path}: 화를 키워드로 안 넘기는 호출이 있다 ({calls})") def test_화_없이_부르면_선다(): """★조용히 비는 대신 **선다** — try 밖이라 로그로도 안 묻힌다.""" from app.core.errors import AppError from app.services.scene_reference_service import SceneReferenceService svc = SceneReferenceService.__new__(SceneReferenceService) svc._db, svc._project_id = None, "p" with pytest.raises(AppError) as ei: svc.build_entity_text_map([], episode_id="") assert ei.value.code == "scene_reference.episode_required" def _seed_project(session, pid, uid, *eids): session.execute(sql_text( "INSERT INTO user_account (id, username, display_name, password_hash, " "role, is_active, created_at, updated_at) VALUES " "(:u, :u, 't', 'x', 'creator', 1, 'x', 'x')"), {"u": uid}) session.execute(sql_text( "INSERT INTO project_registry (id, name, created_by, created_at, " "updated_at) VALUES (:p, 'wiring', :u, 'x', 'x')"), {"p": pid, "u": uid}) for n, eid in enumerate(eids, start=1): session.execute(sql_text( "INSERT INTO episode (id, project_id, title, episode_number, " "source_filename, source_path, status, created_at, updated_at) " "VALUES (:e, :p, :t, :n, 'x', 'x', 'analyzed', 'x', 'x')"), {"e": eid, "p": pid, "t": f"{n}화", "n": n}) def _canon(session, pid, sid, name, etype): cid = str(uuid.uuid4()) meta = ('{"location": null, "visual_identity": null}' if etype == "character" else "{}") session.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 " f"(:c, :p, :s, :n, :e, '', '[]', '{meta}', 't', 'active', 'x', 'x')"), {"c": cid, "p": pid, "s": sid, "n": name, "e": etype}) return cid def test_상태_조회의_합성_진행률이_이_화_것만_센다(pg_session, tmp_path, monkeypatch): """★★★반례 (Codex 가 준 것): EP1 은 C01+O01, EP2 는 C01+O02. 자산은 **EP1 것만** 있다. 그러면 EP2 는 `combo_total=1`·`composite_done=0` 이어야 한다. 앞 판은 분모(`CharacterOutlook` 전체)도 분자(reference asset 전체 count)도 **프로젝트 범위**라, EP1 합성 하나만 있어도 EP2 가 「다 됐다」로 보였다. 바로 위 주석의 「게이트와 같은 물음」에도 어긋난 자리다. """ from app.core.pipeline_gate import get_pipeline_status pid, uid = f"p-{uuid.uuid4()}", f"u-{uuid.uuid4()}" ep1, ep2 = str(uuid.uuid4()), str(uuid.uuid4()) _seed_project(pg_session, pid, uid, ep1, ep2) c01 = _canon(pg_session, pid, "C01", "정순", "character") o01 = _canon(pg_session, pid, "O01", "앞치마", "outlook") o02 = _canon(pg_session, pid, "O02", "외투", "outlook") for cid, eid in ((c01, ep1), (o01, ep1), (c01, ep2), (o02, ep2)): pg_session.execute(sql_text( "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id) " "VALUES (:l, :c, :p, :e)"), {"l": str(uuid.uuid4()), "c": cid, "p": pid, "e": eid}) for eid, oid in ((ep1, o01), (ep2, o02)): pg_session.execute(sql_text( "INSERT INTO character_outlook (id, project_id, episode_id, " "character_id, outlook_id, created_at) " "VALUES (:i, :p, :e, :c, :o, 'x')"), {"i": str(uuid.uuid4()), "p": pid, "e": eid, "c": c01, "o": oid}) # EP1 합성만 실재한다 — **파일까지**. DB 는 상대 경로만 받는다 # (`ck_image_asset_file_path_relative`), 그래서 root 를 옮겨 심는다. monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path / "projects")) rel = f"projects/{pid}/images/{ep1}/reference/ep1.png" png = tmp_path / rel png.parent.mkdir(parents=True, exist_ok=True) png.write_bytes(b"\x89PNG\r\n\x1a\n") pg_session.execute(sql_text( "INSERT INTO image_asset (id, project_id, episode_id, asset_type, " "entity_id, file_path, prompt_used, status, is_primary, created_at) " "VALUES (:i, :p, :e, 'reference', :c, :f, :pu, 'generated', 1, 'x')"), {"i": str(uuid.uuid4()), "p": pid, "e": ep1, "c": c01, "f": rel, "pu": f"[composite:{c01}:{o01}] 정순+앞치마"}) pg_session.commit() st1 = get_pipeline_status(pg_session, pid, ep1)["composite_images"] assert (st1["combo_total"], st1["composite_done"]) == (1, 1), st1 st2 = get_pipeline_status(pg_session, pid, ep2)["composite_images"] assert st2["combo_total"] == 1, f"EP2 분모에 다른 화 쌍이 들어왔다: {st2}" assert st2["composite_done"] == 0, ( f"EP1 합성을 EP2 것으로 셌다: {st2}")