"""이미지 스텝 넷(ref_image_gen · composite_image_gen · character_state_variant · scene_image_pipeline)의 계측이 **읽은 대로** 계약표에 있다 — 글 문(dual LVM)과 이미지 문은 다른 문이고, 단위 상한은 fixture 가 선언하되 production 산출과 대조하는 문이 있다. ★2026-09-02 실측: 넷이 unverified 자리표시(1)라 lane-off dry 가 글 문 165>150 에서 섰고, 이미지 문은 표가 아예 없었다. Codex 정정: composite 는 phase2(아웃룩) + phase3(쌍) 두 생성이라 이미지 4장으로 접으면 안 된다. """ from __future__ import annotations import json import pytest from tools.grounding_audit import canary_run as cr IMAGE_STEPS = ("ref_image_gen", "composite_image_gen", "character_state_variant", "scene_image_pipeline") @pytest.fixture def lvm_off(monkeypatch): from app.core.config import settings monkeypatch.setattr(settings, "scene_lvm_validation_mode", "off") monkeypatch.setattr(settings, "scene_variation_count", 2) monkeypatch.setattr(settings, "scene_single_frame_readback_enabled", False) class TestTheFourStepsHaveVerifiedUnits: def test_none_of_them_is_unverified(self): assert cr.unverified_units_in(IMAGE_STEPS) == [] assert cr.metering_unit_of("ref_image_gen") == cr.UNIT_ENTITY assert cr.metering_unit_of("composite_image_gen") == cr.UNIT_OUTLOOK_PAIR assert cr.metering_unit_of("character_state_variant") == cr.UNIT_STATE_VARIANT assert cr.metering_unit_of("scene_image_pipeline") == cr.UNIT_SHOT def test_text_calls_are_dual_lvm(self, lvm_off): """검증 ask_both(2) + severe 비교 ask_both(2) — 정상 2 · hard 4.""" assert (cr.normal_calls_per_unit_of("ref_image_gen"), cr.calls_per_unit_of("ref_image_gen")) == (2, 4) assert (cr.normal_calls_per_unit_of("composite_image_gen"), cr.calls_per_unit_of("composite_image_gen")) == (4, 8) assert (cr.normal_calls_per_unit_of("character_state_variant"), cr.calls_per_unit_of("character_state_variant")) == (2, 4) def test_scene_text_is_zero_with_lvm_off_and_follows_settings_when_on(self, lvm_off, monkeypatch): from app.core.config import settings # ★still 마다 번역 1(hard 3) — 실측 2026-09-02 (0 으로 적었다가 문에 걸림) assert (cr.normal_calls_per_unit_of("scene_image_pipeline"), cr.calls_per_unit_of("scene_image_pipeline")) == (1, 3) monkeypatch.setattr(settings, "scene_single_frame_readback_enabled", True) assert (cr.normal_calls_per_unit_of("scene_image_pipeline"), cr.calls_per_unit_of("scene_image_pipeline")) == (3, 7) assert cr.image_calls_per_unit_of("scene_image_pipeline") == (2, 26) monkeypatch.setattr(settings, "scene_lvm_validation_mode", "sample") assert (cr.normal_calls_per_unit_of("scene_image_pipeline"), cr.calls_per_unit_of("scene_image_pipeline")) == (7, 15) def test_image_calls_per_unit(self, lvm_off): assert cr.image_calls_per_unit_of("ref_image_gen") == (1, 5) assert cr.image_calls_per_unit_of("composite_image_gen") == (2, 10) assert cr.image_calls_per_unit_of("character_state_variant") == (1, 5) assert cr.image_calls_per_unit_of("scene_image_pipeline") == (2, 24) assert cr.image_calls_per_unit_of("scene_detail") == (0, 0) def test_the_dual_vlm_is_really_two_calls(self): """★래칫 — 검증이 `ask_both` 를 지난다(한 모델로 바뀌면 2 가 거짓이 된다).""" import inspect from app.modules.pipeline import ref_image_pipeline as rp src = inspect.getsource(rp) assert 'ask_both("ref_validation"' in src and 'ask_both("ref_comparison"' in src class TestThePlanUsesTheFixtureCaps: DIMS = {"scenes": 4, "shots": 10, "fan_out_cap": 10, "entity_cap": 20, "outlook_pair_cap": 8, "state_variant_cap": 0} def test_logical_units(self): assert cr.logical_cap_of("ref_image_gen", self.DIMS) == 20 assert cr.logical_cap_of("composite_image_gen", self.DIMS) == 8 assert cr.logical_cap_of("character_state_variant", self.DIMS) == 0 assert cr.logical_cap_of("scene_image_pipeline", self.DIMS) == 10 def test_a_missing_cap_is_refused(self): with pytest.raises(cr.ScopeMismatch, match="OUTLOOK_PAIR_CAP"): cr.logical_cap_of("composite_image_gen", {**self.DIMS, "outlook_pair_cap": None}) with pytest.raises(cr.ScopeMismatch, match="STATE_VARIANT_CAP"): cr.logical_cap_of("character_state_variant", {**self.DIMS, "state_variant_cap": None}) def test_image_plan_totals(self, lvm_off): got = cr.image_plan(self.DIMS, IMAGE_STEPS) by = {r["step"]: r for r in got["rows"]} assert (by["ref_image_gen"]["normal"], by["ref_image_gen"]["hard"]) == (20, 100) assert (by["composite_image_gen"]["normal"], by["composite_image_gen"]["hard"]) == (16, 80) assert (by["character_state_variant"]["normal"], by["character_state_variant"]["hard"]) == (0, 0) assert (by["scene_image_pipeline"]["normal"], by["scene_image_pipeline"]["hard"]) == (20, 240) assert (got["normal_total"], got["hard_total"]) == (56, 420) def test_the_fixtures_declare_both_caps(self): for fx in ("period_episode", "modern_episode", "canary_one_scene"): d = cr.fixture_dimensions(fx) assert d["outlook_pair_cap"] is not None and d["state_variant_cap"] is not None, fx class TestTheCapsAreCheckedAgainstProductionOutput: def _root(self, tmp_path, monkeypatch, *, outlooks=None, shots=None): monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path)) rid = "2dabc123abc1" ep = tmp_path / f"canary_{rid}" / "projects" / "p" / "checkpoints" / "episodes" / "e" if outlooks is not None: (ep / "outlook_phase3").mkdir(parents=True, exist_ok=True) (ep / "outlook_phase3" / "manifest.json").write_text(json.dumps( {"status": "completed", "data": {"outlooks": outlooks}}), encoding="utf-8") if shots is not None: (ep / "shot_staging").mkdir(parents=True, exist_ok=True) (ep / "shot_staging" / "manifest.json").write_text(json.dumps( {"status": "completed", "data": {"shots": shots}}), encoding="utf-8") return rid def test_outlook_pairs_within_cap_pass(self, tmp_path, monkeypatch): rid = self._root(tmp_path, monkeypatch, outlooks=[{"short_id": f"O0{i}"} for i in range(1, 5)]) got = cr.assert_outlook_pair_cap_covers(rid, {"outlook_pair_cap": 8}, set(), ["composite_image_gen"]) assert got["checked"] and got["outlooks"] == 4 def test_outlook_pairs_over_cap_refuse(self, tmp_path, monkeypatch): rid = self._root(tmp_path, monkeypatch, outlooks=[{"short_id": f"O0{i}"} for i in range(1, 5)]) with pytest.raises(cr.ScopeMismatch, match="OUTLOOK_PAIR_CAP"): cr.assert_outlook_pair_cap_covers(rid, {"outlook_pair_cap": 3}, set(), ["composite_image_gen"]) def test_a_done_step_is_not_checked(self, tmp_path, monkeypatch): rid = self._root(tmp_path, monkeypatch, outlooks=[{"short_id": "O01"}] * 9) got = cr.assert_outlook_pair_cap_covers(rid, {"outlook_pair_cap": 1}, {"composite_image_gen"}, ["composite_image_gen"]) assert got["checked"] is False def test_state_variants_use_the_production_predicate(self, tmp_path, monkeypatch): from app.core.subject_state import is_immobilized_state alive = [s for s in ("normal", "alive", "standing") if not is_immobilized_state(s)][:1] assert alive, "★시험 재료 — 안 눕는 상태 하나" shots = [{"character_angles": [{"character": "A", "subject_state": alive[0]}, {"character": "B", "subject_state": None}]}] rid = self._root(tmp_path, monkeypatch, shots=shots) got = cr.assert_state_variant_cap_covers(rid, {"state_variant_cap": 0}, set(), ["character_state_variant"]) assert got["checked"] and got["state_variants"] == 0 def test_an_immobilized_state_over_cap_refuses(self, tmp_path, monkeypatch): from app.core import subject_state as ss # ★어떤 낱말이 눕는 상태인지는 production 술어가 안다 — 여기서 글자를 고르지 않는다 monkeypatch.setattr(ss, "is_immobilized_state", lambda s: s == "X") monkeypatch.setattr(cr, "assert_state_variant_cap_covers", cr.assert_state_variant_cap_covers) shots = [{"character_angles": [{"character": "A", "subject_state": "X"}]}] rid = self._root(tmp_path, monkeypatch, shots=shots) with pytest.raises(cr.ScopeMismatch, match="STATE_VARIANT_CAP"): cr.assert_state_variant_cap_covers(rid, {"state_variant_cap": 0}, set(), ["character_state_variant"]) class TestExpectedUnitsComeFromProductionCheckpoints: """★상한(fixture)과 별개로 **이 run 이 실제로 살 수**를 CP 에서 읽는다 — 승인은 이 수 + 여유.""" def _root(self, tmp_path, monkeypatch): monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path)) rid = "3dabc123abc1" base = tmp_path / f"canary_{rid}" / "projects" / "p" ep = base / "checkpoints" / "episodes" / "e" (ep / "entity_detail").mkdir(parents=True, exist_ok=True) (ep / "entity_detail" / "manifest.json").write_text(json.dumps({"status": "completed", "data": { "entity_queue": [["a", "character", "C01"], ["b", "character", "C02"], ["c", "prop", "P01"], ["d", "location", "L01"], ["e", "location_part", "LP01"], ["f", "outlook", "O01"]]}})) (ep / "outlook_phase3").mkdir(parents=True, exist_ok=True) (ep / "outlook_phase3" / "manifest.json").write_text(json.dumps({"status": "completed", "data": { "outlooks": [{"short_id": "O01"}, {"short_id": "O02"}]}})) (ep / "shot_staging").mkdir(parents=True, exist_ok=True) (ep / "shot_staging" / "manifest.json").write_text(json.dumps({"status": "completed", "data": {"shots": []}})) (ep / "scene_detail").mkdir(parents=True, exist_ok=True) (ep / "scene_detail" / "manifest.json").write_text(json.dumps({"status": "completed", "data": { "scenes": [{}, {}, {}]}})) return rid, base def test_counts_follow_the_checkpoints(self, tmp_path, monkeypatch, lvm_off): rid, base = self._root(tmp_path, monkeypatch) got = cr.expected_units_of(rid) assert got == {"ref_image_gen": 3, "composite_image_gen": 2, "character_state_variant": 0, "scene_image_pipeline": 3}, got def test_already_generated_references_are_subtracted(self, tmp_path, monkeypatch, lvm_off): rid, base = self._root(tmp_path, monkeypatch) img = base / "checkpoints" / "images" / "e" img.mkdir(parents=True, exist_ok=True) (img / "reference_checkpoint.json").write_text(json.dumps({"stage": "reference", "completed": {"x": {}, "y": {}}})) assert cr.expected_units_of(rid)["ref_image_gen"] == 1 def test_expected_image_and_text_totals(self, tmp_path, monkeypatch, lvm_off): rid, _ = self._root(tmp_path, monkeypatch) got = cr.expected_calls_of(rid) # ref 3×(img 1 · text 2) + composite 2×(img 2 · text 4) + scene 3×(img 2 · text 0) assert got["images"] == 3 + 4 + 6 assert got["text"] == 6 + 8 + 4 # ★scene 은 still 마다 글 0 이지만 진입 시 world guide 재생성 1 (실측 2026-09-02) assert got["by_step"]["scene_image_pipeline"] == {"units": 3, "images": 6, "text": 3 + 1} class TestPerRunExtraTextCalls: def test_scene_image_pipeline_has_one_world_guide_call_per_run(self): """★실측 2026-09-02: 0 으로 적어 글 문이 world guide 재생성에서 세웠다.""" assert cr.extra_calls_per_run_of("scene_image_pipeline") == 1 assert cr.extra_calls_per_run_of("ref_image_gen") == 0 def test_the_plan_adds_it_to_normal_and_hard(self, lvm_off): sc = {"mode": "v2_chunk", "fixture": "period_episode", "target": "scene_image_pipeline", "background": "off", "still_recipe": "off", "outdoor": "off"} from app.core.config import settings for k, v in (("background_mode", "off"), ("still_recipe_mode", "off"), ("outdoor_lane_plan_enabled", False), ("outdoor_lane_pipe_enabled", False), ("outdoor_direct_compose_enabled", False), ("outdoor_map_conti_enabled", False)): pass built = cr.build_plan(cr.scenario(**{k: sc[k] for k in ("mode", "fixture", "target")})) assert built["normal"]["scene_image_pipeline"] == 10 * 1 + 1 assert built["caps"]["scene_image_pipeline"] == 10 * 3 + 1