"""승인한 시나리오가 **live 사슬 끝까지** 가나. ★유료 0. Codex BLOCK (2026-09-02): > `build_plan()` 은 fixture 인자를 안 받아 기본값을 씁니다 … > 즉 지금 live 하면 period_episode/v2_chunk/reference_acquisition/70·70 이 > 아니라 **기본 legacy 시나리오가 실행될 수 있습니다.** ★★★「만들었다」와 「불린다」는 다르다 — `fixture_config(mode)`· `fixture_dimensions(fixture)`·`manuscript_pdf(fixture=)`·`approved_for(mode)` 를 다 지어 놓고 `run()` 이 **하나도 안 넘기고 있었다**. """ from __future__ import annotations import ast import inspect import textwrap import pytest from tools.grounding_audit import canary_bootstrap as cbs from tools.grounding_audit import canary_run as cr APPROVED = dict(mode="v2_chunk", fixture="period_episode", target="reference_acquisition") def _calls(fn): tree = ast.parse(textwrap.dedent(inspect.getsource(fn))) return [n for n in ast.walk(tree) if isinstance(n, ast.Call)] def _kwargs_of(fn, name): for c in _calls(fn): if ast.unparse(c.func).endswith(name): return {k.arg for k in c.keywords}, [ast.unparse(a) for a in c.args] return None, None class TestTheScenarioIsOneValue: def test_it_refuses_an_unknown_axis(self): for bad in ({"mode": "없는모드"}, {"fixture": "없는원고"}, {"target": "없는스텝"}): with pytest.raises(cr.ScopeMismatch): cr.scenario(**{**APPROVED, **bad}) def test_it_carries_the_approval_of_that_mode(self): sc = cr.scenario(**APPROVED) assert sc["approved"] == cr.approved_for("v2_chunk") == { "counted": 480, "raw": 480, "search": 120, "download": 240} def test_the_legacy_approval_is_not_inherited(self): """★모드마다 따로다 — 옛 승인을 새 판이 물려받지 않는다.""" assert cr.approved_for("legacy") != cr.approved_for("v2_chunk") with pytest.raises(KeyError): cr.approved_for("v2") class TestThePlanIsBuiltFromIt: def test_all_three_axes_reach_the_plan(self): built = cr.build_plan(cr.scenario(**APPROVED)) assert built["scenario"]["mode"] == "v2_chunk" assert built["plan"]["target"] == "reference_acquisition" assert built["dimensions"] == cr.fixture_dimensions("period_episode") assert built["dimensions"]["shots"] == 10, "★원고가 안 바뀌었다" def test_the_numbers_come_from_the_central_target_count(self): """★2026-09-02: 중앙 조사가 **일반 1** 이 아니라 대상 수에서 온다. 앞 판은 그 스텝에 non-fanout 기본 1 을 줘서 계획이 65 였다. 그러면 `cap=2` 로 첫 대상 언저리에서 끝나고 나머지는 **예산 때문에** `reference_unavailable` 로 접힌다 — 다섯 갈래를 못 잰다(Codex BLOCK). """ built = cr.build_plan(cr.scenario(**APPROVED)) assert built["caps"]["reference_acquisition"] == 25 t = built["totals"] # 2026-09-02 저녁: 앞단 18 스텝을 읽어서 적자 씬 단위 넷(scene_summary·beat·shot·validator)이 # fan_out 자리표시 10 이 아니라 씬 4 로 세어져 94 → 73 (entity 20 · 중앙 20 · 씬 4×5 · 단발 13) # 2026-09-02 밤: 중앙 조사에 대상당 재판정 1 → 73 + 5 = 78 assert t["logical_cap_total"] == 78 assert t["counted_cap_total"] == t["hard_cap_total"] * 2 # 2026-09-02: counted 는 hard cap × 슬롯 2 and t["raw_upper_total"] == 168 def test_the_planning_number_is_inside_the_stop_line(self): """★승인선과 견주는 것은 **논리 수**다 — 최악치가 아니다. ★2026-09-02: 84 는 한때 승인선 70 **밖**이었고 그때 `run()` 이 섰다. Codex 가 93 으로 올려(파생: 15 + 38 + 40) 이제 안이다. """ t = cr.build_plan(cr.scenario(**APPROVED))["totals"] # 2026-09-02 저녁: 앞단 18 스텝을 읽어서 적자 씬 단위 넷(scene_summary·beat·shot·validator)이 # fan_out 자리표시 10 이 아니라 씬 4 로 세어져 94 → 73 (entity 20 · 중앙 20 · 씬 4×5 · 단발 13) assert t["logical_cap_total"] == 78 assert t["logical_cap_total"] <= cr.approved_for("v2_chunk")["counted"] def test_no_argument_still_means_legacy(self): """★음성 대조 — 옛 판은 **그대로** 돈다.""" built = cr.build_plan() assert built["scenario"]["mode"] == cr.DEFAULT_CANARY_MODE assert built["plan"]["target"] == "scene_detail" class TestOneChangedAxisStopsBeforeTheProvider: """★조건 ⑥ — 계획과 실행이 다르면 **provider 0 으로** 선다.""" @pytest.mark.parametrize("axis,other", [ ("mode", "legacy"), ("fixture", "canary_one_scene"), ("target", "scene_detail")]) def test_it_refuses(self, axis, other): sc = cr.scenario(**APPROVED) built = cr.build_plan(sc) bad = cr.scenario(**{**APPROVED, axis: other}) with pytest.raises(cr.ScopeMismatch): cr.assert_scope(built, bad) def test_the_same_scenario_passes(self): sc = cr.scenario(**APPROVED) assert cr.assert_scope(cr.build_plan(sc), sc)["scope"] == "일치" class TestTheLiveChainActuallyThreadsIt: """★★★AST 로 **부르는 자리**를 본다 — 「만들었다」가 아니라 「넘긴다」.""" def test_run_builds_the_plan_from_the_scenario(self): assert "build_plan(sc, run_id=run_id)" in inspect.getsource(cr.run) def test_run_checks_the_scope_before_paying(self): src = inspect.getsource(cr.run) i = src.index("assert_scope(") for later in ("ci.create_database", "cbs.bootstrap(", "cp.run_pipeline("): assert i < src.index(later), f"{later} 가 범위 문보다 앞이다" def test_the_env_gets_the_mode(self): kw, _a = _kwargs_of(cr.run, "prepare_env") assert kw and "grounding_mode" in kw def test_the_bootstrap_gets_the_fixture(self): kw, _a = _kwargs_of(cr.run, "bootstrap") assert kw and "fixture" in kw def test_the_pipeline_gets_this_modes_approval(self): """★legacy 전역이 아니라 **그 모드의 승인**을 넘긴다.""" src = inspect.getsource(cr.run) assert "emergency_counted=approved[\"counted\"]" in src assert "APPROVED_EMERGENCY_COUNTED" not in src, ( "★legacy 전역이 아직 live 사슬에 남아 있다") def test_the_manuscript_comes_from_the_fixture(self): kw, _a = _kwargs_of(cbs.bootstrap, "manuscript_pdf") assert kw and "fixture" in kw class TestTheCliDoesNotFallBackQuietly: def test_it_takes_all_three(self): src = inspect.getsource(cr.main) for flag in ("--mode", "--fixture", "--target"): assert flag in src, f"{flag} 를 안 받는다" def test_it_builds_a_scenario_and_passes_it(self): src = inspect.getsource(cr.main) assert "scenario(mode=" in src and "sc=sc" in src class TestTheOutputNamesTheNumberHonestly: """★조건 ⑦ — `expected_counted` 가 실은 **논리 수**였다.""" def test_the_field_is_renamed(self): src = inspect.getsource(cr.run) assert '"planning_logical"' in src assert '"expected_counted"' not in src, "★옛 이름이 남아 있다" def test_the_run_output_carries_the_identity(self): src = inspect.getsource(cr.run) assert '"scenario": dict(sc)' in src and '"code": git_tip()' in src class TestTheCodeTransitionCannotBeMadeUp: """★★★앞 tip 을 **손으로 넣지 못한다** — 실측으로 지어낸 적이 있다. 2026-09-02: `from_tip` 을 인자로 받게 해 놓고, 짧은 해시만 알고 **뒤를 지어냈다**. 저장소에 없는 SHA 가 장부에 적혔다. 값이 이미 산출에 있는데 사람이 다시 타이핑하게 두면 그런 일이 난다. """ def _run_dir(self, tmp_path, monkeypatch, tip="a" * 40): import json monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path)) d = tmp_path / "canary_abc123abc123" d.mkdir(parents=True, exist_ok=True) (d / "canary_run.json").write_text( json.dumps({"code": {"tip": tip, "clean": True}}), encoding="utf-8") return "abc123abc123" def test_it_reads_the_tip_from_the_run(self, tmp_path, monkeypatch): rid = self._run_dir(tmp_path, monkeypatch, tip="b" * 40) assert cr.recorded_tip(rid) == "b" * 40 def test_a_wrong_from_tip_stops(self, tmp_path, monkeypatch): rid = self._run_dir(tmp_path, monkeypatch, tip="b" * 40) with pytest.raises(cr.ScopeMismatch) as e: cr.record_code_transition(rid, from_tip="c" * 8, why="x", sc=cr.scenario(**APPROVED)) assert "모르는 채 안 잇는다" in str(e.value) def test_a_matching_prefix_is_accepted(self, tmp_path, monkeypatch): """★짧은 해시로 대조하는 것은 된다 — **지어내는 것**만 막는다.""" rid = self._run_dir(tmp_path, monkeypatch, tip="b" * 40) assert cr.recorded_tip(rid).startswith("b" * 8) def test_a_run_without_a_recorded_tip_stops(self, tmp_path, monkeypatch): import json monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path)) d = tmp_path / "canary_abc123abc123" d.mkdir(parents=True, exist_ok=True) (d / "canary_run.json").write_text(json.dumps({}), encoding="utf-8") with pytest.raises(cr.ScopeMismatch): cr.recorded_tip("abc123abc123")