"""후보별 **최종** 프롬프트와 참조가 같은 말을 하는가 — 서비스 끝점. 감사 P0-A 재리뷰. 앞 판의 `test_location_authority_matches_refs.py` 는 `build_still_prompt` 를 손으로 부르고 `plate_attached` 도 직접 넣어 **서비스 배선 오류를 못 잡았다** — 서비스의 `if bgfirst_used:` 를 옛 조건으로 되돌려도 초록이었다. 여기서는 `run_still_recipe_generation` 을 실제로 태우고, 생성 직전(`run_multiroll_select`) 경계에서 후보별 `roll_prompts` · `roll_refs` 를 그대로 잡는다. 재는 계약 셋: (a) lane_prev 가 **꺼져 있어도** ordinary bgfirst 후보 A 가 SHOT BACKGROUND 체인 프롬프트를 쓴다 (LOCATION PHOTOGRAPH 를 가리키지 않는다) (b) 후보의 사진 주장이 그 후보의 **실제 참조 라벨**과 일치한다 (c) `handled_by` 가 체인 프롬프트에 남는다 (재조립이 인자를 안 잃는다) """ from contextlib import ExitStack from typing import Optional from unittest.mock import MagicMock, patch import pytest from tests.unit.test_still_recipe_bgfirst import _write_cp _PHOTO_CLAIM = "the attached LOCATION PHOTOGRAPH" _LOC_LABEL = "LOCATION PHOTOGRAPH" _BG_LABEL = "SHOT BACKGROUND" _HAND_CLAUSE = "THE HAND THAT IS DOING THIS" def _run_bgfirst_candidate( tmp_path, monkeypatch, *, lane_prev: bool = False, handled_by: str = "", authority: str = "plate", ): """표준 non-lane 콘티 샷 하나를 ordinary bgfirst 로 태운다. lane fixture(`_run_lane_shot`)는 lane CP·Step1 sentinel 에 특화돼 후보 B 까지 못 간다. 여기서는 콘티만 붙여 bgfirst 대상이 되게 하고, 생성 경계에서 후보별 프롬프트·참조를 잡은 뒤 세운다. """ from app.core.config import settings from app.modules.pipeline import still_recipe as sr_mod from app.services.still_recipe_service import ( run_still_recipe_generation, ) for on in ("still_bgfirst_enabled", "still_bgfirst_full_enabled"): monkeypatch.setattr(settings, on, True, raising=False) monkeypatch.setattr( settings, "still_lane_prev_bgfirst_enabled", lane_prev, raising=False) monkeypatch.setattr(settings, "still_recipe_mode", "v1", raising=False) for off in ("still_variants_enabled", "still_plate_select_enabled", "still_conti_ab_enabled", "multiroll_fix_rejudge_enabled", "multiroll_gpt_composition_enabled", "still_recipe_camera_frame_enabled", "still_recipe_lighting_enabled", "still_recipe_conduct_enabled"): monkeypatch.setattr(settings, off, False, raising=False) # groupbg 묶음은 share_plan 이 있어야 진입한다 monkeypatch.setattr(settings, "background_share_plan_enabled", authority == "groupbg", raising=False) monkeypatch.setattr(settings, "projects_dir", str(tmp_path)) monkeypatch.setattr(settings, "openai_api_key", "SAMPLE_FIXTURE_KEY") conti = tmp_path / "SAMPLE_FIXTURE_conti.png" conti.write_bytes(b"SAMPLE_FIXTURE_conti") plate = tmp_path / "SAMPLE_FIXTURE_plate.png" plate.write_bytes(b"SAMPLE_FIXTURE_plate") _write_cp(tmp_path, "shot_ref_classify", { "shots": {"S1sh1": { "person_visible": True, "place_en": "SAMPLE FIXTURE PLACE", **({"handled_by": handled_by} if handled_by else {}), }}, "scenes": {"1": {"place_en": "SAMPLE FIXTURE PLACE", "time_of_day_en": "day"}}, "world_anchor_en": "", }) _write_cp(tmp_path, "shot_continuity", {"pose_canon": []}) # lane 이 아닌 **일반 콘티** — 이것이 bgfirst 대상 조건이다. # `conti_pack` 은 서비스가 먼저 검사한다(팩 불일치면 422). from app.modules.pipeline.shot_conti_light import ( resolve_prompt_version as _conti_pack_version, ) _write_cp(tmp_path, "shot_conti_light", { "contis": {"S1sh1": {"status": "ok", "image_path": str(conti), "asset_id": "SAMPLE_FIXTURE_CONTI_UUID"}}, "conti_pack": _conti_pack_version("5"), "lane_conti": {}, }) # ★authority 갈래 (Codex 재리뷰 BLOCK): `_authority_kind` 는 # **plate 가 있으면 무조건 "plate"** 이고 groupbg 는 `plate is None` # 일 때만 진입한다(:4235). 앞 판은 plate 를 배정해 놓고 「groupbg 후보 # B 회귀 잠금」이라 적었는데 **groupbg 를 아예 안 태우고 있었다.** if authority == "groupbg": # 플레이트 없음 + share_plan 그룹 배정 = groupbg 묶음 조건 _write_cp(tmp_path, "background_render", {"groups": {}}) _write_cp(tmp_path, "background_share_plan", {"plan": { "shot_plans": {}, "share_groups": [{ "group_key": "SAMPLE_FIXTURE_GROUP", "shot_tags": ["S1sh1"], "evidence": [{"quote_ko": "SAMPLE FIXTURE 인용"}], }], }}) else: _write_cp(tmp_path, "background_render", {"groups": { "SAMPLE_FIXTURE_BG": { "status": "ok", "png_path": str(plate), # bgfirst 는 플레이트 lineage 를 fail-closed 로 요구한다 "asset_id": "SAMPLE_FIXTURE_PLATE_UUID", "shot_ids": ["S1_Shot1"]}}}) seen: dict = {"rolls": [], "prompt_kw": []} real_prompt = sr_mod.build_still_prompt def _spy_prompt(**kw): seen["prompt_kw"].append(dict(kw)) return real_prompt(**kw) def _capture_rolls(**kw): # ★끝점 — 후보별 프롬프트와 참조가 **여기서** 짝지어져 나간다. seen["rolls"].append({ "prompt": kw.get("prompt") or "", "roll_prompts": dict(kw.get("roll_prompts") or {}), "roll_refs": { k: [lbl for lbl, _ in v] for k, v in (kw.get("roll_refs") or {}).items() }, "labeled_refs": [lbl for lbl, _ in (kw.get("labeled_refs") or [])], }) raise RuntimeError("SAMPLE_FIXTURE gen blocked") still = { "id": "SAMPLE_FIXTURE_STILL", "still_index": 0, "scene_index": 1, "shot_index": 1, "screenplay_scene_heading": "S#1. SAMPLE", "beat_title": "", "still_frame_prompt": "SAMPLE still prompt", "camera_json": "{}", "lighting_json": "{}", "visible_entities_json": "[]", "dependent_scene_id": None, } db = MagicMock() # bgfirst 는 플레이트 lineage(asset UUID)를 fail-closed 로 요구한다 — # 조회가 비면 후보 조립 **전에** 선다. 조립을 재려면 채워야 한다. # ★조회마다 기대 shape 가 다르다 — 하나로 못 덮는다. # · `db.query(ImageAsset.id)` → `row[0]` 으로 읽으니 **튜플** # · `db.query(ImageAsset)` → `.generation_model` 등 **행 객체** # MagicMock 을 그대로 두면 안 정한 속성이 JSON 직렬화에 들어가 선다. class _AssetRow: id = "SAMPLE_FIXTURE_PLATE_UUID" file_path = str(plate) generation_model = "SAMPLE_FIXTURE_MODEL" asset_type = "background" is_intermediate = False prompt = "" reference_image_ids = "[]" def _mk_q(first_value): q = MagicMock() for _m in ("filter", "filter_by", "join", "order_by", "options"): getattr(q, _m).return_value = q q.all.return_value = [] q.first.return_value = first_value return q def _query(*args, **_kw): col_only = bool(args) and getattr(args[0], "key", None) == "id" return _mk_q(("SAMPLE_FIXTURE_PLATE_UUID",) if col_only else _AssetRow()) db.query.side_effect = _query scene_cp = MagicMock() with ExitStack() as stack: for pch in ( patch("app.modules.pipeline.still_recipe.build_still_prompt", _spy_prompt), patch("app.modules.pipeline.multiroll_gemini.make_nb2_gen_fn", return_value=MagicMock()), patch("app.modules.pipeline.multiroll_gemini." "make_gemini_judge_fn", return_value=MagicMock()), patch("app.modules.pipeline.multiroll_gemini." "make_gemini_critique_fn", return_value=MagicMock()), # Step1(배경 재투영)은 **성공한 척** 한다 — 여기서 세우면 # 후보 조립(Step2)까지 못 가서 재려는 것을 못 잰다. patch("app.modules.llm.gpt_image_primitive.call_gpt_image_bytes", return_value=b"SAMPLE_FIXTURE_PNG"), # ★유료 텍스트 호출을 막는다. 이 fixture 는 프롬프트 조립만 # 재므로 어떤 분류·판정도 실제로 부를 필요가 없다. patch("app.modules.llm.llm_client.call_structured", side_effect=RuntimeError("SAMPLE_FIXTURE llm blocked")), patch("app.modules.llm.llm_client.call_text", side_effect=RuntimeError("SAMPLE_FIXTURE llm blocked")), patch("app.modules.pipeline.multiroll_select." "run_multiroll_select", _capture_rolls), ): stack.enter_context(pch) run_still_recipe_generation( db=db, project_id="SAMPLE_FIXTURE_PROJECT", episode_id="SAMPLE_FIXTURE_EPISODE", stills=[still], stills_orm=[], entity_lookup={}, ref_image_map={}, reference_svc=MagicMock(), scene_ref_image_map={}, scene_ref_asset_id_map={}, staging_map={}, scene_cp=scene_cp, persistence_svc=MagicMock(), progress=MagicMock(), project_config=None, scene_dir=tmp_path / "scene", already_done_stills=set(), ) seen["failed"] = [ str(c.args[1]) for c in scene_cp.mark_failed.call_args_list] return seen def _photo_claiming(prompt: str) -> bool: return _PHOTO_CLAIM in prompt def test_every_candidate_prompt_agrees_with_its_own_refs( tmp_path, monkeypatch): """(b) 후보가 「사진이 붙어 있다」고 말하면 그 후보 참조에 실제로 있다. ★이것이 groupbg 후보 B 결함의 회귀 잠금이다 — 앞 판은 프롬프트가 「없다」고 하는데 refs_b 에는 LOCATION PHOTOGRAPH 가 붙었다. """ seen = _run_bgfirst_candidate(tmp_path, monkeypatch, authority="groupbg") assert seen["rolls"], f"생성 경계 도달 전 정지: {seen['failed']}" # ★이 fixture 가 **정말 groupbg 를 탔는지** 먼저 못박는다 — # plate 를 주면 `_authority_kind` 가 "plate" 가 되어 groupbg 를 아예 # 안 태우면서 초록이 된다(앞 판이 그랬다). main_calls = [kw for kw in seen["prompt_kw"] if not kw.get("chain_bg_mode")] assert main_calls, "main 조립이 없다" assert main_calls[0].get("plate_attached") is True, ( "groupbg 갈래인데 plate_attached 가 True 가 아니다 — " "`or bgfirst_used` 가 빠졌거나 fixture 가 plate 를 줬다") mismatches = [] for call in seen["rolls"]: for label, prompt in (call["roll_prompts"] or {"_single": call["prompt"]}).items(): labels = call["roll_refs"].get(label) or call["labeled_refs"] claims = _photo_claiming(prompt) has = any(_LOC_LABEL in lb for lb in labels) if claims != has: mismatches.append((label, claims, has, labels)) assert mismatches == [], ( "후보의 사진 주장과 실제 참조가 어긋난다 " "(label, 주장, 실제, 라벨): " + repr(mismatches)) def test_ordinary_bgfirst_uses_chain_prompt_without_lane_prev( tmp_path, monkeypatch): """(a) lane_prev 가 꺼져 있어도 ordinary bgfirst 는 체인 프롬프트다. ★서비스의 `if bgfirst_used:` 를 옛 조건(`lane_chain or prev_sel`)으로 되돌리면 이 시험이 빨강이 된다 — 그때는 재조립이 안 돌아 기본 프롬프트(LOCATION PHOTOGRAPH 주장)가 Step2 로 들어간다. """ seen = _run_bgfirst_candidate(tmp_path, monkeypatch, lane_prev=False) chain_calls = [kw for kw in seen["prompt_kw"] if kw.get("chain_bg_mode")] assert chain_calls, ( "lane_prev OFF 인 ordinary bgfirst 에서 체인 재조립이 안 돌았다 — " f"조립된 프롬프트 {len(seen['prompt_kw'])}건 모두 chain_bg_mode 가 " f"꺼져 있다. 정지 사유={seen['failed']}") def test_chain_reassembly_keeps_handled_by(tmp_path, monkeypatch): """(c) 체인 재조립이 `handled_by` 를 잃지 않는다. main 조립은 넘기는데 재조립만 빠뜨려 「THE HAND THAT IS DOING THIS」 절이 통째로 사라지던 자리. """ seen = _run_bgfirst_candidate( tmp_path, monkeypatch, handled_by="SAMPLE_FIXTURE_PERSON") chain_calls = [kw for kw in seen["prompt_kw"] if kw.get("chain_bg_mode")] assert chain_calls, f"체인 재조립 미도달: {seen['failed']}" # ★「인자가 왔는가」만 재면 값이 비어도 초록이다 — main 조립이 실제로 # 넘긴 값과 **같은지**를 잰다. 그래야 인자를 빠뜨린 것을 잡는다. main_calls = [kw for kw in seen["prompt_kw"] if not kw.get("chain_bg_mode")] assert main_calls, "main 조립이 없다 — fixture 가 경로를 못 탔다" main_hb = main_calls[0].get("handled_by") assert all(kw.get("handled_by") == main_hb for kw in chain_calls), ( f"체인 재조립의 handled_by 가 main 과 다르다 — main={main_hb!r}, " f"chain={[kw.get('handled_by') for kw in chain_calls]!r}")