"""격리 fixture canary 의 **하나뿐인 입구**. ★유료 0 · 아무것도 안 산다. Codex (2026-08-31) — 「실제 유료 실행 입구가 DB 생성 → migration → bootstrap → 38개 스텝을 **한 순서로** 호출하는지」. """ from __future__ import annotations import json import pytest from tools.grounding_audit import canary_isolation as ci from tools.grounding_audit import canary_run as cr @pytest.fixture def env(monkeypatch, tmp_path): """★★`backend/` 밖에서는 **재지 않고 건너뛴다** (숨기지 않고 사유를 적는다). pydantic-settings 가 `.env` 를 작업 디렉토리 기준으로 찾으므로, 저장소 뿌리에서 돌리면 settings 가 전부 기본값이 된다 — `background_mode` 가 False 로 읽혀 유료 스텝 여섯이 「건너뜀」이 되고, 그 상태로 잰 수는 아무 뜻이 없다. 조용히 다른 답을 내는 것보다 **안 재는 것**이 낫다. ★이 전제가 실제로 지켜지는지는 `TestItRefusesToMeasureFromTheWrongPlace` 가 따로 잰다 — 건너뛰기가 그 시험까지 덮지 않는다. """ from tools.grounding_audit import canary_cost_table as ct if not ct.settings_came_from_backend_env(): import pathlib pytest.skip(f"★`backend/` 에서 돌려야 잰다 — 지금 {pathlib.Path.cwd()}") monkeypatch.setenv("THEROAD_CANARY_TEMPLATE_URL", "postgresql://사용자:암호@어딘가:5432/theroad") monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path)) return tmp_path class TestTheOrderIsTheContract: def test_a_dry_run_stops_before_buying(self, env): got = cr.run(live=False) assert got["live"] is False assert "안 샀다" in got["note"] assert "database" not in got["stages"], "★dry 인데 DB 를 만들었다" assert "pipeline" not in got["stages"] def test_it_writes_what_it_did(self, env): got = cr.run(live=False) p = ci.root_dir(got["run_id"]) / "canary_run.json" saved = json.loads(p.read_text(encoding="utf-8")) assert saved["run_id"] == got["run_id"] assert saved["stages"]["isolation"]["db"].startswith("theroad_canary_") def test_the_five_stages_are_in_this_order(self): """★AST — 문 다섯이 **그 차례로** 불린다.""" import ast import inspect tree = ast.parse(inspect.getsource(cr.run).lstrip()) # ★`ast.walk` 는 **소스 차례가 아니다** — 줄 번호로 세운다 seq = [n.func.attr for n in sorted( (x for x in ast.walk(tree) if isinstance(x, ast.Call) and isinstance(x.func, ast.Attribute)), key=lambda x: (x.lineno, x.col_offset))] want = ["prepare_env", "create_database", "upgrade_command", "bootstrap", "run_pipeline"] got = [x for x in seq if x in want] assert got == want, f"★차례가 다르다: {got}" class TestTheApprovedCeilingIsTheGate: def test_the_gate_uses_the_approved_number(self, env): got = cr.run(live=False) p = got["plan"] assert p["emergency_counted"] == cr.APPROVED_EMERGENCY_COUNTED assert p["emergency_raw"] == cr.APPROVED_EMERGENCY_RAW def test_the_expected_is_inside_but_the_worst_case_is_above(self, env): """★★정지선은 **최악치 아래**에 일부러 둔다. `expected` 는 한 번씩 성공했을 때의 수라 승인 안이어야 하고, `worst_case` 는 tier·슬롯이 다 열렸을 때라 그 위여도 된다 — 실패가 쏟아지면 **일찍 멈추라고** 두는 정지선이지 상한 예측이 아니다. """ p = cr.run(live=False)["plan"] assert p["planning_logical"] <= p["emergency_counted"] assert p["worst_case_counted"] > p["emergency_counted"] assert "정지선이지 상한 예측이 아니다" in p["★ceiling_note"] def test_a_computation_above_the_approval_stops(self, env, monkeypatch): """★★승인은 내가 센 수 **위에** 있다 — 넘으면 조용히 안 넓힌다. ★2026-09-02: 승인은 **모드마다** 따로가 됐다. 옛 전역 `APPROVED_EMERGENCY_COUNTED` 는 live 사슬에서 빠졌으므로 그것을 갈아 끼워도 안 문다 — 실제로 읽는 자리를 갈아 끼운다. """ monkeypatch.setitem(cr.APPROVED_BY_MODE, cr.DEFAULT_CANARY_MODE, {"counted": 1, "raw": 1, "search": 0, "download": 0}) with pytest.raises(ci.IsolationRefused) as e: cr.run(live=False) assert "승인" in str(e.value) def test_an_unknown_step_stops_before_anything(self, env, monkeypatch): real = cr.build_plan def fake(_sc=None, **_kw): got = real() got["plan"]["applied_unknown"] = ["어떤_스텝"] return got monkeypatch.setattr(cr, "build_plan", fake) with pytest.raises(ci.IsolationRefused) as e: cr.run(live=False) assert "미확정" in str(e.value) class TestThePlanIsWhatWeAgreed: def test_nothing_applied_is_unknown(self, env): p = cr.run(live=False)["plan"] assert p["unknown"] == [] def test_the_free_steps_include_the_code_step(self, env): """★★`scene_save` 는 **코드 스텝**이라 비용 목록에서 빠졌었다. Codex 실측: closure 43 · 비용 목록 42 — 빠진 하나가 `scene_save` 다. 그것이 안 돌면 뒤 스텝이 제 체크포인트 없이 시작한다. """ p = cr.run(live=False)["plan"] assert set(p["free"]) == {"scene_save", "shot_dependency", "floor_plan_overlay_payload", "episode_reference_policy"} def test_the_grounding_steps_are_skipped_while_d_is_inert(self, env): p = cr.run(live=False)["plan"] assert {"grounding_a0", "grounding_plan", "grounding_research"} <= set(p["skipped"]) def test_the_bootstrap_cap_is_separate(self, env): p = cr.run(live=False)["plan"] assert p["bootstrap_cap"] == cr.BOOTSTRAP_CAP assert p["bootstrap_cap"] < p["emergency_counted"] class TestIsolationComesBeforeAnyAppImport: """★★★`build_plan()` 이 `app.core.database` 를 끌어온다 (실측 2026-08-31). 환경을 나중에 바꾸면 `SessionLocal` 이 **원본 URL 로 이미 만들어진 뒤**다 — 유료 주행이 원본 DB 에 붙은 세션을 들고 갔을 것이다. """ def test_the_env_is_set_before_the_plan_is_built(self): import ast import inspect tree = ast.parse(inspect.getsource(cr.run).lstrip()) seq = [(n.lineno, (n.func.attr if isinstance(n.func, ast.Attribute) else getattr(n.func, "id", ""))) for n in ast.walk(tree) if isinstance(n, ast.Call)] seq.sort() names = [x[1] for x in seq] assert names.index("prepare_env") < names.index("build_plan"), \ "★계획을 먼저 세우면 원본 URL 로 세션이 만들어진다" assert names.index("assert_database_module_not_loaded") < \ names.index("prepare_env"), "★문이 격리보다 뒤에 있다" def test_building_the_plan_really_pulls_the_db_module(self): """★양성 대조 — 위 차례가 **왜** 필요한지 보인다.""" import subprocess import sys as _s code = ("import sys; sys.path.insert(0,'.');" "from tools.grounding_audit import canary_run as c;" "a='app.core.database' in sys.modules;" "c.build_plan();" "b='app.core.database' in sys.modules;" "print(a,b)") got = subprocess.run([_s.executable, "-c", code], capture_output=True, text=True) assert "False True" in got.stdout, got.stdout + got.stderr class TestTheLockWrapsBothPaidStages: """★★★Codex (2026-08-31) — 잠금 한 scope 가 bootstrap 과 pipeline 을 **둘 다** 감싸야 한다. 옛 자리로 되돌리면 「bootstrap 은 샀고 pipeline 은 0」이 아니라 **provider 전체 0** 으로 실패해야 한다. """ def _live(self, env, monkeypatch, *, order): """부트스트랩 가짜가 **진짜 `_get_router_binding`** 을 부른다.""" from app.modules.llm import llm_client as lc from tools.grounding_audit import canary_bootstrap as cbs from tools.grounding_audit import canary_pipeline as cp from tools.grounding_audit import canary_request_lock as rl lc._router = None lc._binding = None monkeypatch.setenv("GEMINI_API_KEY", "가짜") monkeypatch.setattr(cr.ci, "create_database", lambda rid: {"db_name": "x"}) monkeypatch.setattr(cr.subprocess, "run", lambda *a, **k: type("P", (), { "returncode": 0, "stdout": "", "stderr": ""})()) def fake_boot(rid, *, live, cap, fixture=None): order.append(("bootstrap", rl.assert_router_locked( strict=False)["num_retries"])) lc._get_router_binding() # ★진짜 production 함수 return {"project_id": "p", "episode_id": "e"} def fake_pipe(**kw): order.append(("pipeline", rl.assert_router_locked( strict=False)["num_retries"])) return {"per_step": {}} monkeypatch.setattr(cbs, "bootstrap", fake_boot) monkeypatch.setattr(cp, "run_pipeline", fake_pipe) monkeypatch.setattr(cr.cbs, "bootstrap", fake_boot) monkeypatch.setattr(cr.cp, "run_pipeline", fake_pipe) monkeypatch.setattr(cr.ci, "assert_database_module_not_loaded", lambda: None) # ★실제 엔진 문도 이 시험의 관심이 아니다 — 격리는 제 시험이 따로 잠근다 monkeypatch.setattr(cr.ci, "assert_engine_is_canary", lambda rid: {"checked": False, "faked": True}) # ★단위 미확인 문도 이 시험의 관심이 아니다 — 계약표 시험이 따로 잠근다 monkeypatch.setattr(cr, "unverified_units_in", lambda steps: []) return cr.run(live=True) def test_both_stages_see_a_locked_router(self, env, monkeypatch): order = [] got = self._live(env, monkeypatch, order=order) assert [x[0] for x in order] == ["bootstrap", "pipeline"] assert [x[1] for x in order] == [0, 0], f"★잠금 밖에서 돌았다: {order}" assert got["stages"]["router_ready"]["num_retries"] == 0 assert got["stages"]["router_lock_after_bootstrap"]["locked"] is True assert got["stages"]["router_lock"]["locked"] is True def test_the_router_is_ready_before_the_first_purchase(self, env, monkeypatch): """★★첫 유료 호출 **전에** 짓고 확인한다 — 돈 쓴 뒤가 아니다.""" import ast import inspect tree = ast.parse(inspect.getsource(cr.run).lstrip()) seq = sorted(((n.lineno, n.func.attr if isinstance(n.func, ast.Attribute) else getattr(n.func, "id", "")) for n in ast.walk(tree) if isinstance(n, ast.Call))) names = [x[1] for x in seq] assert names.index("canary_request_lock") < \ names.index("prepare_router") < names.index("bootstrap") assert names.index("bootstrap") < names.index("run_pipeline") def test_a_dirty_router_stops_everything_with_zero_bought( self, env, monkeypatch): """★★★되돌린 판 — 재시도가 살아 있는 Router 가 이미 있으면 **아무도 못 산다**(bootstrap 도 pipeline 도 0).""" from app.modules.llm import llm_client as lc order = [] class _R: num_retries = 3 # ★단위 미확인 문은 이 시험의 관심이 아니다 monkeypatch.setattr(cr, "unverified_units_in", lambda steps: []) monkeypatch.setattr(cr.ci, "create_database", lambda rid: {"db_name": "x"}) monkeypatch.setattr(cr.subprocess, "run", lambda *a, **k: type("P", (), { "returncode": 0, "stdout": "", "stderr": ""})()) monkeypatch.setattr(cr.ci, "assert_database_module_not_loaded", lambda: None) monkeypatch.setattr(cr.cbs, "bootstrap", lambda *a, **k: order.append("bootstrap")) monkeypatch.setattr(cr.cp, "run_pipeline", lambda **k: order.append("pipeline")) lc._binding = lc._RouterBinding(slot="s", router=_R()) try: with pytest.raises(Exception) as e: cr.run(live=True) assert "새 프로세스" in str(e.value) assert order == [], f"★섰어야 하는데 돌았다: {order}" finally: lc._binding = None lc._router = None class TestTheCapsComeFromTheFixtureNotAConstant: """★★★fan_out 을 **고정 2**로 세면 원고가 바뀌어도 수가 안 따라온다. 실측 2026-09-01: 원고를 1씬2샷 → 2씬3샷으로 늘렸는데 `expected` 가 옛 판의 42 그대로였다 (Codex). """ def test_the_dimensions_are_read_from_the_manuscript(self): d = cr.fixture_dimensions() from tests.grounding.fixtures import canary_one_scene as fx assert d["scenes"] == len(fx.segments()) assert d["shots"] == sum(len(s["shots"]) for s in fx.shot_scenes()) assert d["fan_out_cap"] == max(d["scenes"], d["shots"]) def test_a_fan_out_step_gets_the_fixture_cap(self): from app.core.step_manifest import STEP_MANIFEST as M built = cr.build_plan() d = built["dimensions"] import math for s, c0 in built["caps"].items(): # ★caps 는 hard cap(단위 × 허용 호출)이다 — 단위 수로 되돌려 견준다 cpu = cr.calls_per_unit_of(s) # ★글 호출이 0 인 이미지 전용 스텝(floor_plan_render)은 hard cap 0 — 단위 수는 논리 상한에서 직접 읽는다 c = (c0 // cpu if cpu else cr.logical_cap_of(s, d)) if s != cr.CENTRAL_STEP else c0 # ★2026-09-02: 단위는 **명시 계약표**(`METERING_UNITS`)에서 온다 — # manifest 의 `fan_out` 만 보면 `shot_director` 를 1 로 세어 세 번째 # 씬에서 거절되고, `shot_staging` 은 helper 의 batch 를 못 본다 u = cr.metering_unit_of(s) if u == cr.UNIT_SCENE: assert c == d["scenes"], f"★{s}={c}" elif u == cr.UNIT_SHOT: assert c == d["fan_out_cap"], f"★{s}={c}" elif u == cr.UNIT_ENTITY: assert c == d["entity_cap"], f"★{s}={c}" elif u == cr.UNIT_GROUP: assert c == cr.group_cap_of(s), f"★{s}={c}" elif u == cr.UNIT_BATCH: assert c == max(1, math.ceil(d["shots"] / cr.shot_staging_batch_size())), f"★{s}={c}" elif u in (cr.UNIT_SINGLE, cr.UNIT_CENTRAL): assert c == 1 or s == cr.CENTRAL_STEP, f"★{s}={c}" elif u in (cr.UNIT_CHAIN_GROUP, cr.UNIT_FLOOR_PLAN, cr.UNIT_BACKGROUND): # ★배경 사슬 세 단위 (2026-09-03) — 원고 선언 상한 그대로 assert c == d[f"{u}_cap"], f"★{s}={c}" else: # unverified — 끝난 스텝의 역사값은 manifest 로 접는다 assert c == (d["fan_out_cap"] if M[s].get("fan_out") else 1), f"★{s}={c}" def test_growing_the_manuscript_moves_the_number(self, monkeypatch): """★원고가 커지면 `expected` 도 **따라 커진다**.""" before = cr.build_plan()["totals"]["logical_cap_total"] monkeypatch.setattr(cr, "fixture_dimensions", lambda _fx=None: {"scenes": 9, "shots": 9, "fan_out_cap": 9, "entity_cap": 12, # ★단위 상한은 선언이 없으면 선다 — 가짜 치수도 다 갖춘다 "outlook_pair_cap": 8, "state_variant_cap": 0, "outdoor_group_cap": 2, "chain_bg_group_cap": 1, "floor_plan_cap": 2, "background_cap": 4}) after = cr.build_plan()["totals"]["logical_cap_total"] assert after > before def test_the_expected_still_fits_the_approved_stop_line(self, env): """★기대치가 승인 정지선 안이어야 돈다 — 넘으면 선다.""" p = cr.run(live=False)["plan"] assert p["planning_logical"] <= p["emergency_counted"] assert p["fixture_dimensions"]["scenes"] == 2 class TestItStopsBeforeOpeningAnythingWhenThePlanLies: """★★★Codex 조건 ① (2026-09-01) — 선언과 production 술어가 다르면 **provider 도 장부 attempt 도 열기 전에** 선다. 실측 — `visual_continuity_anchor_enabled` 를 fixture 가 False 라 적었는데 `.env` 는 true 였다. 계획표는 「0원」이라 하고 pipeline 은 사러 갔다. """ def test_a_contradicting_declaration_stops_at_build_plan(self, monkeypatch): from app.core import applicability as ap from tools.grounding_audit import canary_cost_table as ct from tests.grounding.test_canary_cost_table import _no monkeypatch.setitem(ap.APPLICABILITY_VALIDATORS, "if_background_mode", _no) monkeypatch.setattr(cr, "fixture_config", lambda _mode, fixture=None: {"grounding_v2": False, "background_mode": True}) with pytest.raises(ct.ApplicabilityContradiction): cr.build_plan() def test_build_plan_runs_before_the_database_and_the_ledger(self): """★순서가 계약이다 — `build_plan()` 이 DB·부트스트랩·pipeline **앞**.""" import inspect src = inspect.getsource(cr.run) i_plan = src.index("built = build_plan(sc, run_id=run_id)") for later in ('got["stages"]["database"] = ci.create_database', "boot = cbs.bootstrap(", "cp.run_pipeline("): assert i_plan < src.index(later), f"{later} 가 계획보다 앞에 있다" def test_the_fixture_declares_only_what_it_can_set(self): """★★내가 **못 정하는 값은 안 적는다**. `settings` 를 읽는 규칙을 project_config 에 적어 두면, 적은 값이 지켜지는 줄 알고 계획을 세우게 된다. """ from app.core.applicability import APPLICABILITY_VALIDATORS from tools.grounding_audit import canary_cost_table as ct declared = cr.default_fixture_config() for key in declared: live = ct._production_says(f"if_{key}") if live is None: continue # runner 가 필요한 규칙 — 선언으로 푼다 assert bool(declared[key]) == live, ( f"{key} 는 production 이 {live} 라고 답한다 — 선언이 거짓말") assert "if_visual_continuity_anchor_enabled" in APPLICABILITY_VALIDATORS class TestItRefusesToMeasureFromTheWrongPlace: """★★★재는 도구가 **어디서 부르냐에 따라 다른 답**을 내면 그 수는 뜻이 없다. 실측 2026-09-01 — 저장소 뿌리에서 부르면 `.env` 를 못 찾아 settings 가 전부 기본값이 되고, `background_mode` 가 False 로 읽혀 유료 스텝 여섯이 통째로 「건너뜀」이 됐다. 조용히 틀린 답을 내는 대신 **선다**. """ def test_the_paid_entry_checks_it_first(self): import inspect from tools.grounding_audit import canary_cost_table as ct src = inspect.getsource(cr.run) i = src.index("ct.assert_backend_cwd()") assert i < src.index("built = build_plan(sc, run_id=run_id)") assert i < src.index("cbs.prepare_env(") assert hasattr(ct, "WrongWorkingDirectory") def test_it_stops_when_the_cwd_is_wrong(self, monkeypatch, tmp_path): from pathlib import Path as _P from tools.grounding_audit import canary_cost_table as ct monkeypatch.setattr(_P, "cwd", classmethod(lambda cls: tmp_path)) with pytest.raises(ct.WrongWorkingDirectory): ct.assert_backend_cwd() def test_it_stops_when_the_env_file_is_missing(self, monkeypatch, tmp_path): from tools.grounding_audit import canary_cost_table as ct monkeypatch.setattr(ct, "BACKEND", tmp_path) with pytest.raises(ct.WrongWorkingDirectory): ct.assert_backend_cwd() def test_module_paths_do_not_depend_on_the_cwd(self, monkeypatch, tmp_path): """★`step_module` 이 낸 길은 **어디서 부르든 열려야** 한다.""" from pathlib import Path as _P from tools.grounding_audit import canary_cost_table as ct want = ct.step_module("scene_detail") monkeypatch.setattr(_P, "cwd", classmethod(lambda cls: tmp_path)) assert ct.step_module("scene_detail") == want assert (ct.BACKEND / want).is_file() class TestTheStepGateUsesTheContractWeActuallyLocked: """★★★문이 **물리 최악보다 좁으면** 멀쩡한 주행이 거짓으로 선다. ★★그런데 이 시험을 처음 쓸 때 나는 `ct.execution_plan()` 을 **직접 불러** 쟀다. production 이 쓰는 것은 `cr.build_plan()` 이고, 거기서는 이미 `plan["contract"]` 를 넣고 있었다 — 즉 **없는 결함을 보고** 같은 계약을 두 번 넣는 중복을 만들 뻔했다 (Codex 2026-09-01). 그래서 여기서는 **넘겨지는 그 plan** 을 붙잡아 잰다. """ def test_build_plan_is_the_only_place_that_injects_the_contract(self): import inspect src = inspect.getsource(cr) assert src.count('["contract"] = dict(LOCKED_CONTRACT)') == 1, ( "계약을 넣는 자리는 **한 곳**이어야 한다") def test_the_plan_carries_the_locked_contract(self, env): got = cr.build_plan()["plan"] assert got["contract"] == cr.LOCKED_CONTRACT def test_the_gate_matches_the_locked_contract(self, env): """★2026-09-02 뒤집었다 — tier 를 닫았으니 한 논리당 **슬롯 수**다. 앞 판은 fallback 을 열어 둬 tier 3 × 슬롯 2 = 6 이었다. 이제 tier 1 이라 2 다. ★수를 못박지 않고 **계약에서 파생**시킨다 — 다음에 슬롯이 늘어도 이 시험이 거꾸로 서지 않는다. """ from tools.grounding_audit import call_bound_contract as cb from tools.grounding_audit import canary_cost_table as ct from tools.grounding_audit import canary_pipeline as cp built = cr.build_plan() per = cp._per_logical(built["plan"]) slots = ct.slots_for("gpt")["slots"] want = cb.layers(contract=cr.LOCKED_CONTRACT, slots=slots)["per_logical_counted"] assert per == want, f"per_logical 이 {per} 인데 계약은 {want} 다" # ★단위가 single(또는 manifest 로 접힌 unverified·fan_out=False)인 스텝으로 잰다 # — visual_continuity_anchor 는 2026-09-02 부터 묶음 단위(fan_out_cap)다 assert built["caps"]["entity_merge"] * per == want assert built["normal"]["scene_detail"] * per == 3 * want def test_the_pipeline_receives_that_very_plan(self, env): """★★사슬을 잇는다 — `build_plan()` 이 계약을 넣고, `run()` 은 **그 plan 을 그대로** 넘긴다. 유료 입구를 실제로 돌리지 않고 잰다 (돌리면 부트스트랩이 산다). """ import ast import inspect import textwrap tree = ast.parse(textwrap.dedent(inspect.getsource(cr.run))) call = next(n for n in ast.walk(tree) if isinstance(n, ast.Call) and ast.unparse(n.func).endswith("run_pipeline")) given = {k.arg: ast.unparse(k.value) for k in call.keywords} assert given["plan"] == 'built[\'plan\']', given["plan"] assert given["caps"] == 'built[\'caps\']', given["caps"] def test_the_locked_contract_pins_every_layer_below_the_door(self): """★2026-09-02 뒤집었다 — 문 아래 두 겹과 tier 를 **다** 잠근다.""" assert cr.LOCKED_CONTRACT["num_retries"] == 0 assert cr.LOCKED_CONTRACT["enable_fallback"] is False assert cr.LOCKED_CONTRACT["sdk_max_retries"] == 0 class TestTheReportNamesWhatIsReallyThere: """★★★보고가 증거보다 세면 안 된다 (2026-09-02, 하루에 두 번). 둘 다 「도구가 거짓말한 자리」다 — 코드는 맞게 하고 **말이 틀렸다**. """ def test_a_dirty_path_it_names_actually_exists(self, tmp_path): """★막는 문이 **없는 경로**를 댔다 — `'ackend/tools/...'`. 원인: porcelain 은 앞 두 칸이 상태 칸인데 `stdout.strip()` 이 ' M path' 의 **첫 칸을 먹어** `[3:]` 이 경로 첫 글자까지 잘랐다. """ import subprocess from tools.grounding_audit import canary_run as cr repo = tmp_path / "repo" (repo / "backend").mkdir(parents=True) run = lambda *a: subprocess.run(a, cwd=str(repo), check=True, capture_output=True) run("git", "init", "-q") run("git", "config", "user.email", "t@t") run("git", "config", "user.name", "t") f = repo / "backend" / "thing.py" f.write_text("x = 1\n", encoding="utf-8") run("git", "add", "-A") run("git", "commit", "-qm", "first") f.write_text("x = 2\n", encoding="utf-8") import unittest.mock as m with m.patch.object(cr, "BACKEND", repo / "backend"): got = cr.git_tip() assert got["clean"] is False assert got["dirty_files"] == ["backend/thing.py"] for rel in got["dirty_files"]: assert (repo / rel).exists(), f"★없는 경로를 댄다: {rel}" def test_it_prints_the_file_it_actually_wrote(self): """★재개는 `canary_resume_*.json` 을 쓰는데 보고는 `canary_run.json` 이라 적었다 — 첫 판 증거를 **덮은 것처럼** 읽힌다.""" import ast import inspect import textwrap from tools.grounding_audit import canary_run as cr src = textwrap.dedent(inspect.getsource(cr.main)) lits = {n.value for n in ast.walk(ast.parse(src)) if isinstance(n, ast.Constant) and isinstance(n.value, str)} assert "canary_run.json" not in lits, ( "★적은 곳을 이름으로 박았다 — `got['output_file']` 을 쓴다") assert "output_file" in src class TestWhatWeBoughtIsSavedBeforeAnyGate: """★★★유료 주행의 pipeline 단계가 **산출 파일에서 사라졌다** (2026-09-02 실측). 부트스트랩 뒤로 `_save()` 가 **한 번도 없었다**. 파이프라인이 다 돈 뒤 `assert_client_retries_observed()` 가 서자 마지막 `_save()` 에 못 닿았고, 운반·문 셋·스텝별 수가 통째로 안 남았다. `pipeline_run.json` 이 따로 남아 살았을 뿐이다. """ def test_the_pipeline_stage_is_saved_before_the_next_gate(self): import ast import inspect import textwrap from tools.grounding_audit import canary_run as cr src = textwrap.dedent(inspect.getsource(cr.run)) tree = ast.parse(src) def _lines_of(name): return sorted(n.lineno for n in ast.walk(tree) if isinstance(n, ast.Call) and getattr(n.func, "attr", "") == name) def _line_of(name, after=0): # ★첫 번째를 집으면 안 된다 — `assert_router_locked` 은 부트스트랩 # 뒤에도 한 번 불린다. **유료 단계 뒤의 것**을 봐야 한다. return next((ln for ln in _lines_of(name) if ln > after), None) pipe = _line_of("run_pipeline") assert pipe, "★`run_pipeline` 을 못 찾았다 — 이름이 바뀌었다" saves = sorted(n.lineno for n in ast.walk(tree) if isinstance(n, ast.Call) and getattr(n.func, "id", "") == "_save") after = [ln for ln in saves if ln > pipe] assert after, "★유료 단계 뒤에 저장이 **하나도 없다**" for gate in ("assert_router_locked", "assert_client_retries_observed"): g = _line_of(gate, after=pipe) if g is None: continue assert any(pipe < ln < g for ln in saves), ( f"★{gate} 이 서면 그 앞에 산 것이 안 남는다") class TestTheReplayPassCannotBuyPictures: """★★★재판정 판은 **검색·받기 문이 0** 이다 (Codex 계약 2026-09-02). 운영자가 조심해서 0 을 적는 것이 아니라 **코드가 0 으로 만든다** — 그래야 이 판이 무엇을 못 하는지가 문에서 증명된다. """ def test_it_zeroes_the_two_outbound_doors(self): from tools.grounding_audit import canary_run as cr got = cr.replay_scope(cr.approved_for("v2_chunk")) assert got["approved"]["search"] == 0 assert got["approved"]["download"] == 0 # ★글/VLM 상한은 **그대로** — 판정은 해야 한다 assert got["approved"]["counted"] == \ cr.approved_for("v2_chunk")["counted"] assert tuple(got["reopen"]) == (cr.CENTRAL_STEP,) def test_it_reopens_only_the_central_step(self): from tools.grounding_audit import canary_run as cr got = cr.replay_scope(cr.approved_for("v2_chunk")) assert len(got["reopen"]) == 1, "★다른 스텝까지 다시 열면 또 산다" def test_the_run_passes_reopen_and_the_zeroed_doors_down(self): """★★값을 만들어 놓고 **안 넘기면** 아무것도 안 바뀐다.""" import ast import inspect import textwrap from tools.grounding_audit import canary_run as cr src = textwrap.dedent(inspect.getsource(cr.run)) call = next(n for n in ast.walk(ast.parse(src)) if isinstance(n, ast.Call) and getattr(n.func, "attr", "") == "run_pipeline") kw = {k.arg for k in call.keywords} assert "reopen" in kw, "★다시 열 스텝을 안 넘긴다" assert "approved_search" in kw and "approved_downloads" in kw assert "approved = _replay[\"approved\"]" in src, \ "★문 0 을 만들어 놓고 안 쓴다" def test_a_normal_pass_reopens_nothing(self): import ast import inspect import textwrap from tools.grounding_audit import canary_run as cr src = textwrap.dedent(inspect.getsource(cr.run)) assert "reopen = tuple(_replay[\"reopen\"]) if _replay else ()" in src class TestANarrowedRunDoesOnlyWhatItSays: """★`steps_scope` — 적은 스텝만, 검색·받기 문 0, 글 상한은 남은 것 그대로.""" def test_it_zeroes_the_outbound_doors_and_keeps_text(self): from tools.grounding_audit import canary_run as cr ap = cr.approved_for("v2_chunk") got = cr.steps_scope(ap, ["shot_selection", "shot_director"]) assert got["approved"]["search"] == 0 assert got["approved"]["download"] == 0 assert got["approved"]["counted"] == ap["counted"] assert tuple(got["reopen"]) == ("shot_selection", "shot_director") def test_an_empty_list_stops(self): import pytest from tools.grounding_audit import canary_run as cr with pytest.raises(cr.ScopeMismatch): cr.steps_scope(cr.approved_for("v2_chunk"), []) def test_replay_and_steps_together_stop(self): import inspect from tools.grounding_audit import canary_run as cr src = inspect.getsource(cr.run) assert "if replay and steps:" in src, "★한 판에 두 뜻을 주면 서야 한다" def test_main_parses_steps_into_the_run(self): import inspect from tools.grounding_audit import canary_run as cr src = inspect.getsource(cr.main) assert '"--steps"' in src and "steps=[" in src class TestSlotsFollowTheProviderNotTheAliasName: """★★`gpt-mini` 는 이름은 gpt 인데 **물리 모델이 gemini** 다 (실측 09-02). alias 앞글자로 가르면 openai 슬롯을 센다 — 키 수가 갈리면 상한이 틀린다. """ def test_provider_wins(self, monkeypatch): from tools.grounding_audit import canary_cost_table as ct import app.modules.llm.gemini_key_pool as gk import app.core.openai_keys as ok monkeypatch.setattr(gk, "key_count", lambda: 3) monkeypatch.setattr(ok, "slot_count", lambda: 2) assert ct.slots_for("gpt-mini", provider="gemini")["slots"] == 3 assert ct.slots_for("gpt", provider="openai")["slots"] == 2 def test_without_provider_the_physical_model_decides(self, monkeypatch): from tools.grounding_audit import canary_cost_table as ct import app.modules.llm.gemini_key_pool as gk import app.core.openai_keys as ok monkeypatch.setattr(gk, "key_count", lambda: 3) monkeypatch.setattr(ok, "slot_count", lambda: 2) assert ct.slots_for("gpt-mini")["how"] == "gemini_key_pool" def test_plan_rows_pass_the_manifest_provider(self): import inspect from tools.grounding_audit import canary_cost_table as ct assert 'provider=M[s].get("provider")' in inspect.getsource(ct.plan_rows)