"""저작기가 **실제 중앙 사슬**에서 도나. ★유료 0 — 모델 호출만 대역. Codex BLOCK (2026-09-02): > 새 시험은 TARGET 에 네 칸을 손으로 넣어 writer 만 불렀기 때문에 못 잡았다. > ledger 원행 → inputs_from_ledger → ca.run → 실제 make_writer → acquire_one > 공개 사슬을 태우십시오. ★손으로 enriched target 을 만들지 않는다 — `producer_payload` 가 실제 writer target 까지 **닿아야** 한다. """ from __future__ import annotations import pytest from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import grounding_outlook_binding as ob from app.modules.pipeline import grounding_search_brief as gsb from tests.grounding.test_central_acquisition import P3, _row, _Spy @pytest.fixture def journal(tmp_path): from app.modules.pipeline import grounding_chunk_journal as cj return cj.ChunkJournal(tmp_path / "j.json", contract={"v": 1}) def _writer(seen, *, era="1960년대", region="대한민국"): """**진짜** `make_writer` — 모델 경계 하나만 대역.""" def _call(tag, system, user, schema, **rest): seen.append({"tag": tag, "user": user}) # ★★검색어도 **선언된 좌표를 다 실어야** 한다 (2026-09-02). 팩이 # 「선언된 말 그대로 모든 질의에 실어라」라고 시키고, provider 앞 # 문이 그것을 본다. 앞 판의 대역은 지역만 싣고 시대를 빠뜨려 # `CoordinatesMissing` 으로 섰다 — **대역이 계약을 안 지킨 것**이다. return {"search_directive_native": f"{era} {region} 저작한 지시문", "search_terms_native": [f"{era} {region} 말"], "language_lock_native": "그 나라 말"} return gsb.make_writer(world_facts=f"- Region: {region}", source_text="원문 표본", era=era, region=region, call=_call) def _run(journal, tmp_path, *, writer, rows=("rs1",)): spy = _Spy() led = ob.bind([_row(r) for r in rows], P3) got = ca.run(led, journal=journal, cap=9, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge, write_brief=writer) return got, spy class TestTheProducerPayloadReachesTheWriter: """★★★첫 대상이 **검색 전에 죽지 않는다**.""" def test_the_writer_is_called_with_real_target_fields(self, journal, tmp_path): seen = [] got, _spy = _run(journal, tmp_path, writer=_writer(seen)) assert seen, "★저작기가 한 번도 안 불렸다" assert got["rows"], "★줄이 없다" def test_it_does_not_die_on_missing_material(self, journal, tmp_path): """★앞 판은 여기서 `BriefInputsMissing` 으로 죽었다.""" seen = [] got, _spy = _run(journal, tmp_path, writer=_writer(seen)) why = " ".join(str(r.get("why") or "") for r in got["rows"]) assert "BriefInputsMissing" not in why, why def test_the_adapter_carries_what_the_writer_reads(self): """★조립 자리 — 어댑터가 저작기가 읽는 칸을 **낸다**.""" from app.modules.pipeline import grounding_acquisition_adapter as ad from app.modules.pipeline.grounding_entity_contract import ( PRODUCER_PAYLOAD) row = {"research_subject_id": "rs", "owner_type": "prop", PRODUCER_PAYLOAD: {"visual_brief": "생김새", "coarse_type_label": "부류", "surface_form": "표기", "search_terms_native": ["말"]}} got = ad.search_target(row) for k in ("owner_type", "coarse_type_label", "surface_form", "visual_brief"): assert got.get(k), f"★{k} 가 저작기까지 안 간다" def test_the_written_directive_is_what_goes_to_search(self, journal, tmp_path): """★★저작기가 낸 지시문이 **검색으로 나간다**.""" seen = [] spy_holder = {} class _Watching(_Spy): def search(self_inner, **kw): spy_holder.setdefault("directives", []).append( kw.get("directive_native")) return _Spy.search(self_inner, **kw) led = ob.bind([_row("rs1")], P3) w = _writer(seen) spy = _Watching() ca.run(led, journal=journal, cap=9, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge, write_brief=w) got = spy_holder.get("directives") or [] assert got, "★검색이 한 번도 안 불렸다" assert "1960년대" in got[0] and "대한민국" in got[0], got[0] class TestTheIdentityFoldsEveryWriterInput: """★②**뼈대**(owner/coarse/terms/언어 잠금/시대/지역)가 바뀌면 다시 산다. ★★뒤집음 2026-09-03 (Codex BLOCK 15c15d7d ①): 문장(surface_form · visual_brief · 원문 표본)은 신원이 **아니다** — 실측으로 문장이 조금 달라 열 대상을 다시 샀다.""" BASE = dict(subject_id="s", owner_type="location", coarse_type_label="부류", surface_form="표기", visual_brief="생김새") KW = dict(world_facts="w", source_text="원문", era="e", region="r") @pytest.mark.parametrize("field,other", [ ("owner_type", "prop"), ("coarse_type_label", "다름"), ("terms_native", ["다른 낱말"]), ("language_lock_native", "다름")]) def test_one_changed_skeleton_field_changes_it(self, field, other): base = gsb.target_identity(dict(self.BASE), **self.KW) got = gsb.target_identity({**self.BASE, field: other}, **self.KW) assert got != base, f"★{field} 가 신원에 안 실린다" @pytest.mark.parametrize("field,other", [ ("surface_form", "다름"), ("visual_brief", "다름")]) def test_a_changed_sentence_does_not_change_it(self, field, other): base = gsb.target_identity(dict(self.BASE), **self.KW) got = gsb.target_identity({**self.BASE, field: other}, **self.KW) assert got == base, ( f"★{field} 는 뼈대가 아닌데 신원에 실린다 — 문장을 고칠 때마다 다시 산다") def test_a_different_source_language_sample_does_not_change_it(self): base = gsb.target_identity(dict(self.BASE), **self.KW) got = gsb.target_identity(dict(self.BASE), **{**self.KW, "source_text": "다른 언어", "world_facts": "다른 사실"}) assert got == base, "★원문 표본·world facts 는 뼈대가 아니다" @pytest.mark.parametrize("axis", ["era", "region"]) def test_a_declared_coordinate_changes_it(self, axis): base = gsb.target_identity(dict(self.BASE), **self.KW) got = gsb.target_identity(dict(self.BASE), **{**self.KW, axis: "다름"}) assert got != base, f"★{axis} 가 신원에 안 실린다" def test_the_same_inputs_reuse(self): assert gsb.target_identity(dict(self.BASE), **self.KW) == \ gsb.target_identity(dict(self.BASE), **self.KW) def test_the_central_run_folds_it(self, journal, tmp_path): """★★★끝점 — 같은 좌표면 재개가 **다시 안 산다**.""" seen = [] w = _writer(seen) _run(journal, tmp_path, writer=w) n = len(seen) _run(journal, tmp_path, writer=w) assert len(seen) == n, "★같은 것을 다시 샀다" def test_a_changed_region_buys_again(self, journal, tmp_path): seen = [] _run(journal, tmp_path, writer=_writer(seen, region="가")) n = len(seen) _run(journal, tmp_path, writer=_writer(seen, region="나")) assert len(seen) > n, "★지역이 바뀌었는데 옛 구매를 되썼다" def test_the_field_list_lives_in_one_place(self): """★목록을 부르는 쪽에 다시 안 적는다.""" import inspect src = inspect.getsource(ca.run) for banned in ("coarse_type_label", "surface_form", "visual_brief"): assert banned not in src, f"★{banned} 를 중앙이 다시 적는다"