"""run_still_recipe_generation 표적 scope 실행 하네스 (재재리뷰 TEST GAP-2). 실제 recipe ordered loop 를 최소 CP fixture 로 진입시켜 잠금: - 전체 stills 컨텍스트 유지 + **effective∩미완료만** 실행 도달 (표적 밖 sibling 은 진행률·실패 기록 어디에도 등장 0) - effective prev 체인 클로저가 실행 집합에 합류 - 표적 진행률 ordinal 연속(2/(n+2)부터, NARROW-3) 생성 자체는 sentinel 예외로 차단(per-still 격리 except → mark_failed) — 어떤 still 이 실행에 도달했는지가 관찰 대상. 시나리오 의존 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 _self_chain_query_mock(): 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_plate_select_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), ("still_recipe_conduct_enabled", False), ("outdoor_lane_pipe_enabled", False), ("outdoor_lane_plan_enabled", False), ("background_share_plan_enabled", False), # 2026-08-05 기본값 ON 승격이 이 하네스를 깨뜨렸다(fail-closed 422) — # 하네스는 최소 경로 고정이므로 명시로 끈다. ("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, *, classify_shots, target_scenes, stills): _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() patches = [ patch("app.core.config.settings.projects_dir", str(tmp_path)), 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()), # 생성 차단 sentinel — per-still 격리 except → mark_failed 로 # '실행 도달' 여부만 관찰 patch("app.modules.pipeline.multiroll_select.run_multiroll_select", side_effect=RuntimeError("SAMPLE generation blocked")), ] + [ patch(f"app.core.config.settings.{name}", val, create=True) for name, val in _FLAG_PATCHES ] from contextlib import ExitStack # safety 사다리 미발화 선언 — MagicMock 반환(truthy)이 provenance # 정정 분기를 오발화시켜 review_notes JSON 직렬화가 깨진다 persistence = MagicMock() persistence.safety_ladder_call_provenance.return_value = None with ExitStack() as stack: for pch in patches: stack.enter_context(pch) generated = run_still_recipe_generation( db=_self_chain_query_mock(), project_id=PID, episode_id=EID, stills=stills, 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=persistence, progress=progress, project_config=None, scene_dir=tmp_path / "scene", already_done_stills=set(), target_scenes=target_scenes, ) gen_msgs = [ c.args for c in progress.update.call_args_list if c.args and "레시피 스틸 생성 중" in c.args[0] ] failed_ids = [c.args[0] for c in scene_cp.mark_failed.call_args_list] return generated, gen_msgs, failed_ids def test_target_scope_only_effective_reaches_execution(tmp_path): stills = [_still("st_1", 1, 1), _still("st_2", 2, 1), _still("st_3", 3, 1)] generated, gen_msgs, failed_ids = _run( tmp_path, classify_shots={}, target_scenes=(2,), stills=stills) # 표적 밖 st_1/st_3 은 실행·진행률·실패 기록 어디에도 등장 0 assert failed_ids == ["st_2"] assert len(gen_msgs) == 1 and "S2sh1" in gen_msgs[0][0] # NARROW-3: 첫 샷=2/(exec_total+2)=2/3 assert (gen_msgs[0][1], gen_msgs[0][2]) == (2, 3) assert generated == 0 # sentinel 로 생성 차단 — 완료 0 def test_target_scope_prev_chain_joins_execution(tmp_path): """effective prev 체인(S2sh1→S1sh1)이 실행 집합에 합류, sibling 제외, ordinal 연속 2→3.""" stills = [_still("st_1", 1, 1), _still("st_2", 2, 1), _still("st_3", 3, 1)] generated, gen_msgs, failed_ids = _run( tmp_path, classify_shots={"S2sh1": {"prev": "S1sh1"}}, target_scenes=(2,), stills=stills) assert failed_ids == ["st_1", "st_2"] # 스토리 순서 assert [m[0].split("(")[-1].rstrip(")") for m in gen_msgs] == [ "S1sh1", "S2sh1"] assert [(m[1], m[2]) for m in gen_msgs] == [(2, 4), (3, 4)] assert generated == 0 # audit: requested=[st_2], dependency_added=[st_1] from app.modules.pipeline.scene_image_scope import ( load_scope_audit, scope_audit_path, ) audit = load_scope_audit( scope_audit_path(tmp_path, PID, EID), expected_scenes=(2,)) assert audit["requested_ids"] == ["st_2"] assert audit["dependency_added"] == ["st_1"] def test_no_target_runs_all_byte_identical(tmp_path): """target 미설정=전 still 실행 도달 (기존 경로 그대로).""" stills = [_still("st_1", 1, 1), _still("st_2", 2, 1)] generated, gen_msgs, failed_ids = _run( tmp_path, classify_shots={}, target_scenes=None, stills=stills) assert failed_ids == ["st_1", "st_2"] assert [(m[1], m[2]) for m in gen_msgs] == [(2, 4), (3, 4)]