"""계측 단위의 정본은 **명시 계약표**다 — manifest `fan_out` 도 AST 도 아니다 (Codex 재리뷰 2026-09-02). 실측 셋: `shot_director`·`scene_camera_flow` 는 `fan_out=False` 인데 씬마다 부르고, `shot_staging` 은 스텝 파일엔 호출이 없지만 helper 가 BATCH_SIZE 마다 부른다. ★모르는 단위는 single 로 접지 않고 **살 자리면 live 가 선다**. """ from __future__ import annotations import inspect import math import textwrap import pytest from tools.grounding_audit import canary_cost_table as ct from tools.grounding_audit import canary_run as cr class TestEveryMeteredStepHasADeclaredUnit: @pytest.mark.parametrize("mode", ["v2_chunk", "legacy"]) @pytest.mark.parametrize("target", ["scene_detail", "world_guide", "scene_image_pipeline"]) def test_closure_steps_are_all_in_the_table(self, mode, target): plan = ct.execution_plan(config=cr.fixture_config(mode), target=target) missing = [s for s in plan.get("applied_metered") or [] if s not in cr.METERING_UNITS] assert not missing, f"★계약표에 없는 계측 스텝: {missing}" def test_an_unknown_step_stops(self): with pytest.raises(cr.ScopeMismatch, match="single 로 접지 않는다"): cr.metering_unit_of("no_such_step") class TestTheUnitsThatWereReadMatchTheCode: def test_shot_staging_batches_by_the_production_constant(self): from app.modules.pipeline import shot_staging as ss assert cr.shot_staging_batch_size() == ss.BATCH_SIZE src = inspect.getsource(ss.run_shot_staging) assert "BATCH_SIZE" in src and "call_structured(" in src d = cr.fixture_dimensions("period_episode") assert cr.logical_cap_of("shot_staging", d) == max(1, math.ceil(d["shots"] / ss.BATCH_SIZE)) def test_world_guide_makes_one_call(self): from app.modules import world_guide_generator as wg src = textwrap.dedent(inspect.getsource(wg.WorldGuideGenerator.generate)) assert src.count("call_structured(") == 1 assert cr.metering_unit_of("world_guide") == cr.UNIT_SINGLE def test_scene_detail_is_per_shot(self): from app.core.steps import detail_steps as ds src = textwrap.dedent(inspect.getsource(ds.SceneDetailStep._run_shots_with_retry)) assert "submit(" in src assert cr.metering_unit_of("scene_detail") == cr.UNIT_SHOT class TestUnverifiedUnitsFailClosedBeforeTheProvider: def test_the_stage_one_remaining_steps_are_all_verified(self): plan = ct.execution_plan(config=cr.fixture_config("v2_chunk"), target="world_guide") remaining_after_reference = ["scene_camera_flow", "shot_staging", "scene_consistency", "visual_continuity_anchor", "scene_detail", "world_guide"] for s in remaining_after_reference: assert s in (plan.get("applied_metered") or []) or s in cr.METERING_UNITS assert cr.metering_unit_of(s) != cr.UNIT_UNVERIFIED, s def test_run_refuses_a_live_with_an_unverified_step_still_to_buy(self): src = textwrap.dedent(inspect.getsource(cr.run)) i_plan = src.index("built = build_plan(sc, run_id=run_id)") i_ref = src.index("계측 단위를 안 읽은 스텝") i_boot = src.index("boot = cbs.bootstrap(") assert i_plan < i_ref < i_boot, "★미확인 단위 문이 부트스트랩보다 뒤에 있다" assert "if live and _unv" in src def test_unverified_units_in_reports_only_unverified(self): # ★entity_detail 은 2026-09-02 에 읽어서 single 로 적었다 — 아직 안 읽은 것으로 잰다 got = cr.unverified_units_in(["scene_camera_flow", "background_render", "world_guide"]) assert got == ["background_render"] class TestTheGroupCapComesFromProductionSettings: """★Codex 재리뷰 (2026-09-02): anchor 는 prop 묶음(8)과 고정 인물 묶음(4)을 독립 상한으로 산다 — fixture fan_out_cap(10)으로 접으면 12 를 10 으로 낮게 센다.""" def test_the_cap_is_the_sum_of_the_two_production_caps(self, monkeypatch): from app.core.config import settings monkeypatch.setattr(settings, "visual_continuity_anchor_group_cap", 8) monkeypatch.setattr(settings, "immobilized_subject_continuity_enabled", True) monkeypatch.setattr(settings, "immobilized_subject_continuity_cap", 4) d = cr.fixture_dimensions("period_episode") assert cr.group_cap_of("visual_continuity_anchor") == 12 assert cr.logical_cap_of("visual_continuity_anchor", d) == 12 > d["fan_out_cap"] def test_the_second_lane_only_counts_when_enabled(self, monkeypatch): from app.core.config import settings monkeypatch.setattr(settings, "visual_continuity_anchor_group_cap", 8) monkeypatch.setattr(settings, "immobilized_subject_continuity_enabled", False) monkeypatch.setattr(settings, "immobilized_subject_continuity_cap", 4) assert cr.group_cap_of("visual_continuity_anchor") == 8 def test_an_unlisted_group_step_stops(self): with pytest.raises(cr.ScopeMismatch, match="group 상한"): cr.group_cap_of("some_other_group_step") def test_production_selects_seeds_with_both_caps(self): """★래칫 — 스텝이 실제로 두 상한 함수를 부른다.""" from app.core.steps import visual_continuity_anchor_step as vs src = inspect.getsource(vs) assert "select_seeds_with_cap(" in src and "select_immobilized_seeds_with_cap(" in src class TestTheEntityCapIsDeclaredByTheFixture: """★Codex 재리뷰 (2026-09-02): entity_t2i 를 엔티티 단위라 적고 상한은 max(씬,샷)=10 을 썼는데 canary 실제 엔티티는 15 였다.""" def test_period_fixture_declares_at_least_the_measured_count(self): from tests.grounding.fixtures import period_episode as P assert P.ENTITY_CAP >= 16 d = cr.fixture_dimensions("period_episode") assert d["entity_cap"] == P.ENTITY_CAP assert cr.logical_cap_of("entity_t2i", d) == P.ENTITY_CAP > d["fan_out_cap"] def test_a_fixture_without_the_declaration_stops(self): with pytest.raises(cr.ScopeMismatch, match="ENTITY_CAP"): cr.logical_cap_of("entity_t2i", {"scenes": 4, "shots": 10, "fan_out_cap": 10, "entity_cap": None}) def test_every_canary_fixture_declares_it(self): from tools.grounding_audit import canary_bootstrap as cbs for name in cbs.FIXTURES: assert cr.fixture_dimensions(name)["entity_cap"], name class TestCallsPerUnitFollowProduction: """★Codex 재리뷰 (2026-09-02): 단위 안의 허용 호출도 계약이다 — 정상 1 과 hard cap 을 갈라 적는다.""" def test_shot_staging_hard_cap_is_batches_times_max_attempts(self): from app.modules.pipeline import shot_staging as ss d = cr.fixture_dimensions("period_episode") assert cr.calls_per_unit_of("shot_staging") == ss.MAX_ATTEMPTS assert cr.hard_cap_of("shot_staging", d) == cr.logical_cap_of("shot_staging", d) * ss.MAX_ATTEMPTS assert "MAX_ATTEMPTS" in inspect.getsource(ss.run_shot_staging) def test_entity_t2i_attempts_three_in_source(self): from app.core.steps import entity_steps as es src = inspect.getsource(es.EntityT2iStep) assert "for attempt in range(3)" in src assert cr.calls_per_unit_of("entity_t2i") == 3 def test_scene_detail_structure_matches_the_declared_twenty_eight(self): """(1 기본 + 1 수정 + 2 variation × (judge 1 + repair ≤2 + verify 1 + 재시도 repair 1 + verify 1)) × 2""" from app.core.steps import detail_steps as ds one = textwrap.dedent(inspect.getsource(ds.SceneDetailStep._analyze_one)) run = textwrap.dedent(inspect.getsource(ds.SceneDetailStep._run_shots_with_retry)) rep = textwrap.dedent(inspect.getsource(ds.SceneDetailStep._attempt_owned_redraw_repair)) assert one.count("call_structured(") == 2, "★기본 1 + 수정 재호출 1" assert "run_owned_judge(" in one and "_attempt_owned_redraw_repair(" in one assert "_plan = (_upper,) if retried_model else (self.project_config, _upper)" in rep, "★repair 모델 둘" assert rep.count("run_owned_repair(") == 1 and rep.count("run_owned_judge(") == 1, "★repair + verify" assert "retried_model=True" in rep, "★상위 모델 재시도 한 번" assert "retry_tasks" in run and run.count("submit(") == 2, "★실패 샷 한 번 더" assert cr.calls_per_unit_of("scene_detail") == 28 def test_unlisted_steps_default_to_one(self): assert cr.calls_per_unit_of("scene_camera_flow") == 1 def test_build_plan_splits_normal_from_hard(self): sc = cr.scenario(mode="v2_chunk", fixture="period_episode", target="world_guide") built = cr.build_plan(sc=sc) t = built["totals"] assert t["hard_cap_total"] > t["logical_cap_total"] assert built["normal"]["scene_detail"] == 10 and built["caps"]["scene_detail"] == 280 row = next(r for r in built["rows"] if r["step"] == "scene_detail") assert row["normal_expected"] == 10 and row["calls_per_unit"] == 28 class TestTheEntityCapMustCoverTheRealQueue: def _run(self, tmp_path, monkeypatch, queue_len): root = tmp_path / "run" ep = root / "projects" / "p" / "checkpoints" / "episodes" / "e" / "entity_detail" ep.mkdir(parents=True) import json (ep / "manifest.json").write_text(json.dumps( {"status": "completed", "data": {"entity_queue": [["n", "prop", f"P{i:02d}"] for i in range(queue_len)]}}), encoding="utf-8") monkeypatch.setattr(cr.ci, "root_dir", lambda _rid: root) def test_a_queue_inside_the_cap_passes(self, tmp_path, monkeypatch): self._run(tmp_path, monkeypatch, 15) got = cr.assert_entity_cap_covers_queue("r", {"entity_cap": 20}, set(), ["entity_t2i", "world_guide"]) assert got == {"checked": True, "queue": 15, "entity_cap": 20, "steps": ["entity_t2i"]} def test_a_queue_over_the_cap_stops(self, tmp_path, monkeypatch): """★양성 대조 — 원고를 키우고 상수를 안 바꾸면 여기서 선다.""" self._run(tmp_path, monkeypatch, 21) with pytest.raises(cr.ScopeMismatch, match="ENTITY_CAP"): cr.assert_entity_cap_covers_queue("r", {"entity_cap": 20}, set(), ["entity_t2i"]) def test_a_done_entity_step_is_not_measured(self, tmp_path, monkeypatch): self._run(tmp_path, monkeypatch, 99) got = cr.assert_entity_cap_covers_queue("r", {"entity_cap": 20}, {"entity_t2i"}, ["entity_t2i"]) assert got["checked"] is False class TestTheRepairChainIsMeasuredDirectly: """★Codex GO 조건 1 — nested repair + 재귀 ID-repair + verify 를 **직접 돌려 센다**. 한 variation 의 redraw 갈래 최대 = repair(모델 둘 · 첫 시도 실패) 2 + verify 1 + ID 거절 뒤 상위 모델 재귀(repair 1 + verify 1) = **5**. 실제 `_attempt_owned_redraw_repair` 를 가짜 repair/judge 로 돌린다 — 이름을 세는 것이 아니다. """ def _step(self, monkeypatch): from app.core.steps import detail_steps as ds s = ds.SceneDetailStep.__new__(ds.SceneDetailStep) s.project_config = {} monkeypatch.setattr(s, "build_opik_metadata", lambda *a, **k: {}, raising=False) return s def _drive(self, monkeypatch, *, first_repair_fails: bool, id_bad_once: bool): from app.core.steps import _owned_repair as orp, _owned_judge as ojd from app.core import visible_entities_validator as vev calls = {"repair": 0, "judge": 0} def fake_repair(**kw): calls["repair"] += 1 if first_repair_fails and calls["repair"] == 1: raise RuntimeError("repair LLM 경계 실패(가짜)") return {"t2i_prompt": "repaired prompt", "owned_object_usage": []} def fake_judge(**kw): calls["judge"] += 1 return [] # verify 통과(redraw 없음) monkeypatch.setattr(orp, "run_owned_repair", fake_repair) monkeypatch.setattr(ojd, "run_owned_judge", fake_judge) monkeypatch.setattr(vev, "validate_visible_entities_contract", lambda *a, **k: None) seen = {"n": 0} def check_prompts(_result): seen["n"] += 1 return {"C99"} if (id_bad_once and seen["n"] == 1) else set() s = self._step(monkeypatch) variation = {"t2i_prompt": "original", "owned_object_usage": []} got = s._attempt_owned_redraw_repair( variation=variation, original_t2i="original", owned=["P01"], judge_violations=[{"verdict": "redraw_violation", "object": "P01"}], cam_dir="", result={"t2i_variations": [variation]}, name_by_short_id={}, check_prompts_fn=check_prompts, si=1, shot_idx=1) return got, calls def test_the_worst_path_is_five_calls(self, monkeypatch): got, calls = self._drive(monkeypatch, first_repair_fails=True, id_bad_once=True) assert got is not None, "★재귀 뒤 성공해야 한다" assert calls == {"repair": 3, "judge": 2}, calls # 2 + 1 + (1 + 1) assert sum(calls.values()) == 5 def test_the_plain_path_is_two_calls(self, monkeypatch): """★양성 대조 — 첫 repair 성공 · ID 통과면 repair 1 + verify 1.""" got, calls = self._drive(monkeypatch, first_repair_fails=False, id_bad_once=False) assert got is not None and calls == {"repair": 1, "judge": 1} def test_the_declared_twenty_eight_is_built_from_five(self): """(기본 1 + 수정 1 + 2 variation × (judge 1 + repair 갈래 5)) × 실패 샷 재시도 2""" assert (1 + 1 + 2 * (1 + 5)) * 2 == 28 == cr.calls_per_unit_of("scene_detail")