"""confined fp 배선 통합 회귀 (Codex 재리뷰 BLOCK-1 요구 — spy 고정). target_scope 하네스 패턴 재사용: 실제 run_still_recipe_generation 루프를 최소 CP fixture 로 진입시키고, 생성은 sentinel 로 차단해 **어느 경로에 도달했는가**만 관찰한다. 잠그는 계약: - active=true: plate 선택(select_plate_for_shot)·reconcile(resolve_conti_ plate_authority) 호출 0 + plate 결손류 mark_failed 0 + confined branch 도달(run_multiroll_select 가 fp 라벨 참조·confined 프롬프트를 받음) - active=false(판별 거부): legacy 경로 도달(confined 프롬프트 아님) - OFF: 판별 호출 자체가 0 """ from __future__ import annotations import json from unittest.mock import MagicMock, patch PID, EID = "SAMPLE_P", "SAMPLE_E" def _write_cp(tmp_path, step_id, data): d = tmp_path / PID / "checkpoints" / "episodes" / EID / step_id d.mkdir(parents=True, exist_ok=True) (d / "manifest.json").write_text( json.dumps({"status": "completed", "data": data}, ensure_ascii=False), encoding="utf-8") def _still(sid, si, shi): return { "id": sid, "still_index": 0, "scene_index": si, "shot_index": shi, "screenplay_scene_heading": f"S#{si}. SAMPLE", "beat_title": "", "still_frame_prompt": "SAMPLE still prompt", "camera_json": "{}", "lighting_json": "{}", "visible_entities_json": "[]", "dependent_scene_id": None, } def _db(): 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 = None q.count.return_value = 0 db = MagicMock() db.query.return_value = q return db _FLAG_PATCHES = [ ("still_bgfirst_enabled", False), ("still_bgfirst_full_enabled", False), ("still_variants_enabled", False), ("still_conti_ab_enabled", False), ("multiroll_fix_rejudge_enabled", False), ("multiroll_gpt_composition_enabled", False), ("still_recipe_camera_frame_enabled", False), ("still_recipe_lighting_enabled", False), ("outdoor_lane_pipe_enabled", False), ("outdoor_lane_plan_enabled", False), ("background_share_plan_enabled", False), ("still_lane_prev_bgfirst_enabled", False), # 2026-08-20: 이 둘이 없으면 하네스가 기계의 .env 를 물어(둘 다 켜짐) # 걷기가 **실제 유료 LLM 호출**을 내보낸다 — 시험을 돌릴 때마다 돈이 # 나간다(실측: 전체 시험 한 바퀴에 Gemini 168건). 이 파일의 시험들은 # 간판·시대를 겨누지 않으므로 기준선에서 끈다. # 감시: .venv/bin/python -m pytest ... -p tests.netprobe ("signage_author_enabled", False), ("era_research_enabled", False), ] def _run(tmp_path, *, flag_on, applies, classify_shots, plate_on=False): # plate_on: active 시나리오에서만 켠다 — plate 선택이 켜져 있어도 # confined 가 그 경로를 건너뛰는지가 관찰 대상. legacy 시나리오에선 # plate 호출이 정상이라 금지 spy 와 양립하지 않으므로 끈다. _write_cp(tmp_path, "shot_ref_classify", { "shots": classify_shots, "scenes": {}, "world_anchor_en": ""}) _write_cp(tmp_path, "shot_continuity", {"pose_canon": []}) _write_cp(tmp_path, "shot_conti_light", {"contis": {}}) from app.services.still_recipe_service import ( run_still_recipe_generation, ) progress = MagicMock() scene_cp = MagicMock() seen = {"apt": 0, "base": 0, "shot": 0, "runs": []} def fake_apt(tag, brief, **kw): seen["apt"] += 1 return {"applies": applies, "reason_ko": "SAMPLE"} def fake_base(space_tag, location_text, out_path, **kw): seen["base"] += 1 out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_bytes(b"\x89PNG\r\n\x1a\n" + b"0" * 8) return out_path def fake_shot(tag, base_fp, sections, out_path, **kw): seen["shot"] += 1 out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_bytes(b"\x89PNG\r\n\x1a\n" + b"0" * 8) return out_path, {"reads": {}, "mismatches": [], "scene_description_en": "SAMPLE LAYOUT", "fixed": False} def spy_run(**kw): seen["runs"].append(kw) raise RuntimeError("SAMPLE generation blocked") plate_spy = MagicMock( side_effect=AssertionError("plate select must not run")) reconcile_spy = MagicMock( side_effect=AssertionError("plate reconcile must not run")) patches = [ patch("app.core.config.settings.projects_dir", str(tmp_path)), patch("app.core.config.settings.still_confined_fp_enabled", flag_on, create=True), patch("app.core.config.settings.still_plate_select_enabled", plate_on, create=True), patch("app.modules.pipeline.confined_fp.judge_fp_applicability", side_effect=fake_apt), patch("app.modules.pipeline.confined_fp.produce_base_fp", side_effect=fake_base), patch("app.modules.pipeline.confined_fp.produce_shot_fp", side_effect=fake_shot), patch("app.modules.pipeline.plate_select.select_plate_for_shot", plate_spy), patch( "app.modules.pipeline.still_recipe" ".resolve_conti_plate_authority", reconcile_spy), 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()), patch("app.modules.pipeline.multiroll_select.run_multiroll_select", side_effect=spy_run), ] + [ patch(f"app.core.config.settings.{name}", val, create=True) for name, val in _FLAG_PATCHES ] from contextlib import ExitStack with ExitStack() as stack: for pch in patches: stack.enter_context(pch) run_still_recipe_generation( db=_db(), project_id=PID, episode_id=EID, stills=[_still("st_1", 1, 1)], 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=progress, project_config=None, scene_dir=tmp_path / "scene", already_done_stills=set(), target_scenes=None, ) failed = [ (c.args[0], str(c.args[1]) if len(c.args) > 1 else "") for c in scene_cp.mark_failed.call_args_list ] return seen, failed _CONFINED_CLS = { "S1sh1": {"confined_structure": True, "place_en": "The pilot station inside the machine."}} def test_active_skips_plate_paths_and_reaches_confined_branch(tmp_path): seen, failed = _run( tmp_path, flag_on=True, applies=True, classify_shots=_CONFINED_CLS, plate_on=True) # plate 선택·reconcile 0회는 spy 의 AssertionError 부재로 증명 — # mark_failed 에 그 메시지가 없어야 한다 assert seen["apt"] == 1 and seen["base"] == 1 and seen["shot"] == 1 assert len(seen["runs"]) == 1 kw = seen["runs"][0] # confined branch 도달 증거: fp 라벨 참조 + confined 프롬프트(모든 롤 # 동일) + 브리프의 첨부 권위 꼬리 부재 ref_labels = [lab for lab, _ in kw["labeled_refs"]] assert any("FLOOR PLAN" in lab for lab in ref_labels) rp = kw.get("roll_prompts") or {} # 2026-08-13 #106 사용자 확정: a/b 구도 변주는 confined 포함 전 경로 — # 첫 롤=기준, 둘째 롤=변주 절 동반(서로 상이). 두 롤 모두 confined # 재료(도면 산문)는 공유한다. (구 계약 "모든 롤 동일"은 wave 이전.) assert rp and len(set(rp.values())) == len(rp) >= 1 for body in rp.values(): assert "SCENE LAYOUT" in body and "SAMPLE LAYOUT" in body assert "LOCATION PHOTOGRAPH" not in body # 실패는 sentinel 1건뿐 — plate 결손류 fail-closed 없음 assert len(failed) == 1 and "SAMPLE generation blocked" in failed[0][1] def test_apt_false_falls_back_to_legacy(tmp_path): seen, failed = _run( tmp_path, flag_on=True, applies=False, classify_shots=_CONFINED_CLS) assert seen["apt"] == 1 and seen["base"] == 0 and seen["shot"] == 0 assert len(seen["runs"]) == 1 kw = seen["runs"][0] ref_labels = [lab for lab, _ in kw["labeled_refs"]] assert not any("FLOOR PLAN" in lab for lab in ref_labels) assert len(failed) == 1 # sentinel 만 def test_flag_off_never_judges(tmp_path): seen, failed = _run( tmp_path, flag_on=False, applies=True, classify_shots=_CONFINED_CLS) assert seen["apt"] == 0 and seen["base"] == 0 and seen["shot"] == 0 assert len(seen["runs"]) == 1 # 기존 경로 도달 assert len(failed) == 1