"""C 주행기 — ★유료 0. 실제 에피소드로 계획을 짓고 **구조만** 잰다. Codex: C 에서 자동으로 재는 것은 schema/runtime enum · 씬 제약 · 격리 · 장부/재개 · Opik/provider 대조뿐이다. 의미는 사람이 본다. """ from __future__ import annotations import json import pytest from tools.grounding_audit import cc_c_runner as cr from tools.grounding_audit import cc_runner as rr def _plan(): try: return cr.build_plan() except LookupError: pytest.skip("후보 에피소드가 없다") @pytest.fixture(autouse=True) def _shot_images_present(monkeypatch): """★선정기가 **샷 이미지 있는 후보만** 고른다 (2026-08-31). 이 파일의 시험들은 **승인 문·좌표·재생**을 재는 것이지 에피소드 고르기가 아니라, 시험 DB 가 비면 선정기가 서서 재려던 축이 통째로 안 돈다. ★그 조건 자체는 `test_shot_catalog.py` 가 따로 잰다 — 여기서 대역을 다는 것이 그 축을 무디게 만들지 않는다. """ from tools.grounding_audit import cc_c_preflight as pf real = pf.candidates monkeypatch.setattr( pf, "shot_image_counts", lambda: {c["episode_id"]: 1 for c in real()}) class TestThePlanComesFromTheRealEpisode: def test_every_chunk_carries_its_own_catalog(self): pick, plan, segs, extra = _plan() assert len(plan) == pick["chunks"] >= 2 for c in plan: cat = extra["catalogs"][c["chunk_id"]] node = (c["payload"]["schema"]["properties"]["rows"]["items"] ["properties"]["shot_appearance_ids"]) if cat: assert node["items"]["enum"] == [x["id"] for x in cat] else: assert node.get("maxItems") == 0 def test_the_world_facts_come_from_the_production_builder(self): _pick, plan, _segs, extra = _plan() assert extra["world"], "★세계 사실이 비었다" # ★장소 전용 builder 가 버리는 것까지 실려야 한다 assert "[" in extra["world"], "★규칙 종류 표시가 없다" assert extra["world"] in plan[0]["payload"]["parts"][0]["text"] def test_the_shot_descriptions_are_whole(self): """★자르면 구별점이 사라져 잘못된 샷에 붙는다.""" _pick, plan, _segs, extra = _plan() text = plan[0]["payload"]["parts"][0]["text"] for c in extra["catalogs"][plan[0]["chunk_id"]][:5]: if c["description"]: assert c["description"] in text class TestTheLockCarriesWhatDecidesThisRun: def test_it_pins_the_episode_and_the_shot_ids(self, tmp_path): pick, plan, segs, extra = _plan() got = rr.run(tmp_path / "j.json", rr._dry_send, world_facts=extra["world"], cap=len(plan) + 1, dispatch_budget=len(plan) + 1, approved_slots=None, plan=plan, segments=segs, shot_catalogs=extra["catalogs"], lock_extra={"project_id": pick["project_id"], "episode_id": pick["episode_id"], "chunks": len(plan), "shot_ids": sorted( c["id"] for cat in extra["catalogs"].values() for c in cat)}) lock = got["lock"] assert lock["episode_id"] == pick["episode_id"] assert lock["chunks"] == len(plan) assert lock["shot_ids"], "★샷 ID 가 잠금에 없다" def test_a_processing_stamp_per_chunk(self, tmp_path): pick, plan, segs, extra = _plan() got = rr.run(tmp_path / "j.json", rr._dry_send, world_facts=extra["world"], cap=len(plan) + 1, dispatch_budget=len(plan) + 1, approved_slots=None, plan=plan, segments=segs, shot_catalogs=extra["catalogs"]) assert set(got["processing_stamps"]) == {c["chunk_id"] for c in plan} # ★해석 지문은 **획득 신원에 안 접힌다** from app.modules.pipeline import grounding_chunk as gc acq = gc.acquisition_identity( plan[0]["payload"], model_alias=rr.MODEL_ALIAS, model_physical=rr.physical_model(), request_contract=rr.PINNED) assert got["processing_stamps"][plan[0]["chunk_id"]] != acq class TestTheAutomaticChecksSeeOnlyStructure: CATS = {"c0": [{"id": "s1#1", "scene_id": "scene-1", "description": ""}]} def _got(self, rows): return {"reduced": {"rows": rows}} def _row(self, ids, scene="scene-1", quote="가방"): from tests.grounding.fixtures import synthetic_episode as ep return {"local_id": "c0#0", "shot_appearance_ids": list(ids), "occurrences": [{"source_span": ep.span_of(1, quote, 1), "source_quote": quote}]} def _segs(self): from tests.grounding.fixtures import synthetic_episode as ep return ep.segment_texts() def test_a_catalog_outsider_is_caught(self): bad = cr.automatic_checks(self._got([self._row(["s9#9"])]), self.CATS, self._segs()) assert any("catalog 밖" in x for x in bad) def test_a_bad_quote_is_caught(self): r = self._row(["s1#1"]) r["occurrences"][0]["source_quote"] = "원고에 없는 말" bad = cr.automatic_checks(self._got([r]), self.CATS, self._segs()) assert any("인용이 그 자리에 없다" in x for x in bad) def test_a_clean_row_passes(self): """★positive control — 막기만 하고 정상까지 잡으면 못 쓴다.""" assert cr.automatic_checks(self._got([self._row(["s1#1"])]), self.CATS, self._segs()) == [] class TestItNeverInventsTargets: """★★실제 에피소드에는 심어 둔 표적이 **없다**. 있는 척하면 그것이 지어낸 정답이다.""" def test_the_c_runner_does_not_use_the_target_scorer(self): """★**코드 줄만** 본다 — 「왜 안 쓰나」를 적은 설명까지 잡으면, 까닭을 지워야 통과하는 시험이 된다.""" import ast import inspect tree = ast.parse(inspect.getsource(cr)) if (tree.body and isinstance(tree.body[0], ast.Expr) and isinstance(tree.body[0].value, ast.Constant)): tree.body = tree.body[1:] code = ast.unparse(tree) for banned in ("cc_scorer", "EXPECTED_TARGETS", "score_run", "mechanical_candidate"): assert banned not in code, f"★표적 채점을 쓴다: {banned}" def test_the_output_says_what_it_is_not(self, tmp_path): import sys old = sys.argv try: sys.argv = ["x", "--dry", str(tmp_path / "j.json")] cr.main() finally: sys.argv = old d = json.loads((tmp_path / "j_run.json").read_text(encoding="utf-8")) assert "production PASS 도 아니다" in d["note"] assert "사람만" in d["note"] class TestTheApprovedCapIsADoorNotAComputation: """★★★계획에서 다시 계산하면 승인 뒤 구간이 늘 때 **조용히 넓어진다**.""" def test_a_plan_of_a_different_size_stops(self): with pytest.raises(cr.ApprovedScopeMismatch): cr.assert_approved([{"chunk_id": "c0"}] * 5) # 논리 6 def test_the_approved_numbers_are_written_by_hand(self): assert cr.C_APPROVED_LOGICAL == 5 and cr.C_APPROVED_SLOTS == 2 def test_it_stops_before_the_journal_and_the_provider(self, tmp_path, monkeypatch): """★장부도 안 열고 trace 도 안 연다.""" import sys monkeypatch.setattr(cr, "C_APPROVED_LOGICAL", 99) old = sys.argv try: sys.argv = ["x", "--dry", str(tmp_path / "j.json")] with pytest.raises(cr.ApprovedScopeMismatch): cr.main() finally: sys.argv = old assert not (tmp_path / "j.json").exists(), "★장부를 열었다" class TestTheLockPinsThePlannedIdentities: """★★잠금이 project/episode/chunks/shot_ids 뿐이면, **원문이나 샷 전문이 바뀌어도 잠금은 같고 획득 신원만 달라진다** — 부분 장부에 옛 것과 새 것이 섞인다 (Codex 2026-08-31).""" def test_the_lock_carries_every_chunk_identity(self): pick, plan, _segs, extra = _plan() lk = cr._lock_extra(pick, plan, extra["catalogs"]) assert len(lk["chunk_acquisition_ids"]) == len(plan) assert len(set(lk["chunk_acquisition_ids"])) == len(plan) assert "merge_acquisition" in lk def test_a_changed_manuscript_drifts_the_lock(self, tmp_path): pick, plan, segs, extra = _plan() a = cr._lock_extra(pick, plan, extra["catalogs"]) # ★원문 한 글자를 바꾸면 payload 가 바뀌고 신원도 바뀐다 from app.modules.pipeline import grounding_chunk as gc moved = list(plan) p0 = dict(moved[0]) p0["payload"] = gc.build_chunk_payload( p0["segment_ids"], {**segs, p0["segment_ids"][0]: segs[p0["segment_ids"][0]] + "덧붙임"}, extra["world"], shot_catalog=extra["catalogs"][p0["chunk_id"]]) moved[0] = p0 b = cr._lock_extra(pick, moved, extra["catalogs"]) assert a["chunk_acquisition_ids"] != b["chunk_acquisition_ids"] assert a["shot_ids"] == b["shot_ids"], "★샷은 그대로여야 한다" def test_the_identities_use_the_same_callable_as_preflight(self): from app.modules.pipeline import grounding_chunk as gc pick, plan, _segs, extra = _plan() lk = cr._lock_extra(pick, plan, extra["catalogs"]) want = gc.acquisition_identity( plan[0]["payload"], model_alias=rr.MODEL_ALIAS, model_physical=rr.physical_model(), request_contract=rr.PINNED) assert lk["chunk_acquisition_ids"][0] == want class TestCHasItsOwnTraceCoordinates: """★실제 에피소드를 **합성 B 이름으로** 기록하면 감사를 오독한다.""" def test_the_coordinates_differ_from_b(self): assert cr.C_TAG != rr.LIVE_TAG assert cr.C_TRACE_NAME != rr.LIVE_TRACE_NAME assert cr.C_THREAD != rr.LIVE_THREAD assert cr.C_STEP != rr.LIVE_STEP def test_the_sender_is_not_copied_but_parameterised(self): """★복사하면 두 벌이 되어 한쪽만 고쳐진다.""" import inspect sig = inspect.signature(rr.live_send) for k in ("trace_name", "tag", "thread", "step"): assert k in sig.parameters, f"★{k} 를 인자로 못 연다" src = inspect.getsource(cr) assert "open_trace" not in src, "★sender 를 복사했다" def test_the_c_send_passes_its_own_coordinates(self, monkeypatch): seen = {} def fake(payload, ident, rid="", **kw): seen.update(kw) return {"rows": []} monkeypatch.setattr(rr, "live_send", fake) import sys # main 안의 `_send` 를 그대로 태운다 _pick, plan, segs, extra = _plan() old = sys.argv try: sys.argv = ["x", "--dry", "/dev/null"] finally: sys.argv = old # ★직접 부르기 — main 을 안 태우고 좌표만 본다 rr.live_send({"system": "", "parts": [{"text": ""}], "schema": {}}, "i", "", trace_name=cr.C_TRACE_NAME, tag=cr.C_TAG, thread=cr.C_THREAD, step=cr.C_STEP) assert seen["tag"] == cr.C_TAG and seen["step"] == cr.C_STEP class TestReplayBuysNothingAndRedoesTheProcessing: def test_replay_of_a_c_journal_reuses_everything(self, tmp_path): import sys old = sys.argv try: sys.argv = ["x", "--dry", str(tmp_path / "j.json")] cr.main() sys.argv = ["x", "--replay", str(tmp_path / "j.json")] assert cr.main() in (0, 1) finally: sys.argv = old d = json.loads((tmp_path / "j_run.json").read_text(encoding="utf-8")) assert d["bought"] == 0 and d["dispatched"] == 0 assert d["reused"] == cr.C_APPROVED_LOGICAL assert d["processing_stamps"], "★해석 지문이 없다" def test_a_missing_response_stops_instead_of_buying(self, tmp_path): import sys old = sys.argv try: sys.argv = ["x", "--dry", str(tmp_path / "j.json")] cr.main() finally: sys.argv = old d = json.loads((tmp_path / "j.json").read_text(encoding="utf-8")) d["calls"] = d["calls"][:1] (tmp_path / "j.json").write_text(json.dumps(d, ensure_ascii=False), encoding="utf-8") _pick, plan, segs, extra = _plan() with pytest.raises(LookupError, match="사지 않는다"): rr.replay(tmp_path / "j.json", world_facts=extra["world"], plan=plan, segments=segs, shot_catalogs=extra["catalogs"])