"""★★★상한이 **production 호출 구조**를 따르나 (Codex BLOCK 2026-09-02). manifest 의 `fan_out` 만 보면 틀린다 — `shot_director` 는 `fan_out=False` 인데 `direct_shots()` 가 **씬마다** LLM 을 부른다. 앞 판은 그것을 논리 1 로 세어 스텝 문을 2 로 걸었고, 그러면 **정상 경로에서도 세 번째 씬이 반드시 거절**된다. 수와 문이 같은 SOT(`logical_cap_of`)를 쓰는지, 그리고 그 SOT 가 손으로 적은 목록이므로 **코드와 맞는지** 여기서 잠근다. """ from __future__ import annotations import textwrap import ast import inspect import pytest from tools.grounding_audit import canary_run as cr @pytest.fixture(autouse=True) def _environ_is_restored(): """★★`cr.run()` 은 격리 자식을 위해 `os.environ` 을 **직접** 바꾼다 (`GROUNDING_MODE=v2_chunk` 등). 시험 프로세스 안에서 그것이 남으면 뒤 시험이 빈 config 를 `v2_chunk` 로 읽는다 — 실제로 `test_legacy_and_v2_hashes_do_not_move` 가 전체 판에서만 붉었다(혼자 돌면 초록). 순서 의존은 이렇게 생긴다. 되돌린다. """ import os before = dict(os.environ) try: yield finally: os.environ.clear() os.environ.update(before) def _loops_scenes_and_calls_llm(fn) -> bool: """그 함수 안에 **씬을 도는 for** 가 있고 그 안에서 LLM 을 부르나.""" names = ("_resolve_scene_llm", "call_structured", "_select_for_scene", "submit") def _calls_llm(node) -> bool: return any(isinstance(n, ast.Call) and ( getattr(n.func, "id", "") or getattr(n.func, "attr", "")) in names for n in ast.walk(node)) import textwrap # ★메서드 소스는 들여쓰기가 남아 `ast.parse` 가 죽는다 — 벗겨서 읽는다 tree = ast.parse(textwrap.dedent(inspect.getsource(fn))) for node in ast.walk(tree): # ★`for` 문뿐 아니라 **comprehension** 도 loop 다 — `shot_selection` 은 # `{pool.submit(...): i for i, s in enumerate(scenes_to_select)}` 로 # fan-out 한다. `ast.For` 만 보면 「씬을 안 돈다」로 잘못 읽는다. if isinstance(node, ast.For): if "scene" in ast.unparse(node.iter) and _calls_llm(node): return True elif isinstance(node, (ast.DictComp, ast.ListComp, ast.SetComp, ast.GeneratorExp)): if any("scene" in ast.unparse(g.iter) for g in node.generators) \ and _calls_llm(node): return True return False class TestTheDeclaredUnitMatchesProduction: def test_shot_director_really_calls_per_scene(self): from app.modules.pipeline import shot_director as sd assert _loops_scenes_and_calls_llm(sd.direct_shots) def test_shot_selection_really_calls_per_scene(self): from app.core.steps.shot_selection_step import ShotSelectionStep as S assert _loops_scenes_and_calls_llm(S._execute) def test_every_declared_step_is_checked_here(self): """★목록에 스텝을 더하면 **이 시험도** 더해야 한다.""" assert set(cr.SCENE_UNIT_STEPS) == {"shot_selection", "shot_director", "scene_camera_flow", "scene_consistency", # ★2026-09-02 앞단 글 스텝 다섯 — 읽어서 씬(또는 그 이하 묶음)마다 한 번 "scene_summary", "beat_extract", "shot_extract", "shot_validator", "grounding_chunk"} class TestTheNumbersAndTheGateShareOneSource: def test_the_period_fixture_gives_four_and_four(self): sc = cr.scenario(mode="v2_chunk", fixture="period_episode", target="shot_director") built = cr.build_plan(sc) assert built["dimensions"]["scenes"] == 4 assert built["caps"]["shot_selection"] == 4 assert built["caps"]["shot_director"] == 4 def test_the_counted_caps_are_eight_and_eight(self): from tools.grounding_audit import canary_cost_table as ct sc = cr.scenario(mode="v2_chunk", fixture="period_episode", target="shot_director") built = cr.build_plan(sc) rows = {r["step"]: r for r in ct.plan_rows( logical_caps=built["caps"], contract=cr.LOCKED_CONTRACT, target="shot_director")} present = [s for s in cr.SCENE_UNIT_STEPS if s in rows] assert {"shot_selection", "shot_director"} <= set(present) for s in present: # ★씬마다 한 번 × 단위당 허용 호출(재시도) — beat/shot_extract 는 max_retry 5 → 6 assert rows[s]["logical_cap"] == 4 * cr.calls_per_unit_of(s), (s, rows[s]) assert rows[s]["counted_cap"] == 8 * cr.calls_per_unit_of(s), rows[s] def test_the_gate_uses_the_same_caps(self): """★`canary_pipeline` 의 스텝 문은 `caps` 를 그대로 곱한다 — 보고서 수만 고치고 문을 안 고치면 세 번째 씬이 또 거절된다.""" from tools.grounding_audit import canary_pipeline as cp src = inspect.getsource(cp.run_pipeline) assert "int(caps.get(s, 0)) * _per_logical(plan)" in src def test_no_second_cap_rule_survives(self): """★`build_plan` 이 단위를 **다시 적지 않는다**.""" src = inspect.getsource(cr.build_plan) assert 'M[s].get("fan_out")' not in src assert "logical_cap_of(" in src class TestTheOldRuleWouldHaveRefusedTheThirdScene: """★양성 대조 — 옛 규칙으로 세면 `shot_director` 문이 2 다.""" def test_manifest_alone_says_one(self): from app.core.step_manifest import STEP_MANIFEST as M assert not M["shot_director"].get("fan_out") def test_the_new_rule_says_scenes(self): assert cr.fan_out_unit_of("shot_director") == "scenes" assert cr.logical_cap_of("shot_director", {"scenes": 4, "fan_out_cap": 10}) == 4 # ★음성 대조 — 다른 스텝은 하던 대로 assert cr.fan_out_unit_of("scene_director") in ("one", "fan_out_cap") class TestANarrowedDryShowsOnlyWhatItRuns: """★closure 전체 수(69)를 보이면 이 판이 실제로 무엇을 사는지가 안 읽힌다. 좁힌 판은 **도는 스텝만의** 논리·슬롯 최악을 같은 `caps` 에서 잘라 낸다.""" def test_the_narrowed_block_carries_eight_and_sixteen(self, monkeypatch): monkeypatch.setattr(cr, "git_tip", lambda: { "tip": "f" * 40, "clean": True, "dirty_files": []}) monkeypatch.setattr(cr, "assert_resume_transition", lambda *_a, **_k: {"resume": False}) sc = cr.scenario(mode="v2_chunk", fixture="period_episode", target="shot_director") got = cr.run(live=False, run_id="deadbeefcafe", sc=sc, steps=["shot_selection", "shot_director"]) nw = got["plan"]["narrowed"] assert nw["steps"] == ["shot_selection", "shot_director"] assert nw["caps"] == {"shot_selection": 4, "shot_director": 4} assert nw["logical_total"] == 8 assert nw["counted_total"] == 16 assert {r["step"] for r in nw["rows"]} == set(nw["steps"]) # ★문 셋은 0 — 이 판이 사진을 살 길이 없다 assert got["plan"]["approved_search_requests"] == 0 assert got["plan"]["approved_download_operations"] == 0 assert got["plan"]["approved_image_calls"] == 0 def test_a_full_run_has_no_narrowed_block(self, monkeypatch): monkeypatch.setattr(cr, "git_tip", lambda: { "tip": "f" * 40, "clean": True, "dirty_files": []}) monkeypatch.setattr(cr, "assert_resume_transition", lambda *_a, **_k: {"resume": False}) sc = cr.scenario(mode="v2_chunk", fixture="period_episode", target="shot_director") got = cr.run(live=False, run_id="deadbeefcaf1", sc=sc) assert got["plan"]["narrowed"] is None class TestTheAfternoonPairAreAlsoPerScene: """★2026-09-02 오후 실측 — `scene_camera_flow` 가 cap 2 에서 12건 거절돼 failed (attempt 0dfb50f3f95b · 유료 5). 둘 다 씬마다 task 를 만들어 ThreadPool 로 per-scene worker 를 부른다. 손 목록에 넣고 **그 함수를 직접 보는** 래칫.""" @pytest.mark.parametrize("step,module,worker", [ ("scene_camera_flow", "app.core.steps.scene_camera_flow_step", "_process_scene"), ("scene_consistency", "app.core.steps.scene_consistency_step", "_process_one_scene"), ]) def test_the_worker_calls_the_llm_and_execute_submits_it(self, step, module, worker): import importlib mod = importlib.import_module(module) cls = next(v for v in vars(mod).values() if isinstance(v, type) and v.__name__.endswith("Step") and hasattr(v, worker)) wsrc = textwrap.dedent(inspect.getsource(getattr(cls, worker))) assert "call_structured(" in wsrc, f"★{worker} 가 LLM 을 안 부른다" esrc = textwrap.dedent(inspect.getsource(cls._execute)) assert worker in esrc and "submit(" in esrc, f"★_execute 가 {worker} 를 submit 하지 않는다" assert step in cr.SCENE_UNIT_STEPS assert cr.fan_out_unit_of(step) == "scenes" def test_the_declared_scene_set_is_exactly_the_four(self): assert set(cr.SCENE_UNIT_STEPS) == {"shot_selection", "shot_director", "scene_camera_flow", "scene_consistency", # ★2026-09-02 앞단 글 스텝 다섯 — 읽어서 씬(또는 그 이하 묶음)마다 한 번 "scene_summary", "beat_extract", "shot_extract", "shot_validator", "grounding_chunk"} def test_group_unit_steps_are_bounded_by_the_production_caps(self): assert cr.fan_out_unit_of("visual_continuity_anchor") == "group" d = cr.fixture_dimensions("period_episode") assert cr.logical_cap_of("visual_continuity_anchor", d) == cr.group_cap_of("visual_continuity_anchor") def test_stage_one_closure_units_are_declared(self): """★1단계 남은 스텝들의 단위가 전부 **알려진** 값이다 — 「one」 으로 떨어지는 것은 LLM 을 안 부르거나 한 번만 부르는 스텝뿐.""" want = {"scene_camera_flow": "scenes", "scene_consistency": "scenes", "scene_detail": "fan_out_cap", "visual_continuity_anchor": "group", "shot_staging": "batch", "world_guide": "one"} assert {s: cr.fan_out_unit_of(s) for s in want} == want class TestTheEarlyTextStepsFanOutPerSceneOrLess: """★2026-09-02: 앞단 다섯을 씬 단위로 적었다 — production 이 씬(또는 씬 묶음·구간)마다 submit 하는지 소스로 잰다. 묶음·구간은 씬 수 이하라 씬 상한이 위쪽 상한이다.""" def _src(self, obj): import inspect, textwrap return textwrap.dedent(inspect.getsource(obj)) def test_scene_summary_submits_per_scene(self): try: from app.modules import scene_summarizer as m except ImportError: from app.modules.pipeline import scene_summarizer as m assert "submit(" in self._src(m) def test_beat_and_shot_extract_submit_per_bundle_of_scenes(self): from app.core.steps import beat_shot_steps as m src = self._src(m) assert "_build_bundles(" in src and "submit(" in src assert "max_retry = 5" in src # ★hard 6 의 근거 def test_shot_validator_submits_per_scene(self): from app.core.steps.shot_validator_step import ShotValidatorStep as S src = self._src(S) assert "_validate_scene" in src and "submit(" in src def test_grounding_chunk_submits_per_chunk(self): from app.core.steps import grounding_chunk_step as m assert "ThreadPoolExecutor" in self._src(m) def test_they_are_scene_units_in_the_contract(self): for step in ("scene_summary", "beat_extract", "shot_extract", "shot_validator", "grounding_chunk"): assert cr.fan_out_unit_of(step) == "scenes", step