"""중앙 참조 조사 **스텝** — CP 셋 → 조사 → CP 하나. ★유료 0 · 활성화 0. Codex D-inert 2/3 (2026-09-01) — 「기존 `reference_acquisition_step` 의 NotImplemented 자리를 중앙 acquisition 한 곳으로 연결. screen 장부 + phase3 뒤 outlook_binding + obligation planner 를 한 CP 로 합침. **한 buyer 만**.」 ★이 갈래는 아직 못 켠다 — `v2_chunk` 가 `PLANNED_MODES` 에만 있어 `resolve_grounding_mode` 가 거절한다. 그래서 시험은 **공개 함수를 직접** 부른다. """ from __future__ import annotations import pytest from app.modules.pipeline import coarse_type_pick as ctp from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import grounding_central_inputs as ci from app.modules.pipeline import grounding_reference_obligations as ro from app.modules.pipeline import grounding_screen as gs from app.modules.pipeline.grounding_entity_contract import ( FACET_BINDING, MATERIALIZABLE_OWNER_TYPES, PRODUCER_PAYLOAD) EV = {"source_quote": "원문", "occurrences": [{"source_span": {"segment_id": "scene-1"}}]} def _pl(word): return {"coarse_type_label": word, "visual_brief": f"{word} 찾기", "search_terms_native": [word], "language_lock_native": "ko"} def _row(rsid, owner, *, word="가", fb=None, screen=gs.SCREEN_OBLIGATION): r = {"research_subject_id": rsid, "owner_type": owner, "screen": screen, "source_evidence": dict(EV), PRODUCER_PAYLOAD: _pl(word)} if fb is not None: r[FACET_BINDING] = fb return r FB_LP = {"local_id": "c0#2", "final_id": "LP01", "owner_type": "location_part", "parent_local_id": "c0#1", "parent_owner_type": "location", "parent_final_id": "L01"} FB_O = {"local_id": "c0#3", "final_id": "O01", "owner_type": "outlook", "parent_local_id": "c0#1", "parent_owner_type": "character", "parent_final_id": "C01"} SCREEN_CP = {"data": {"rows": [ _row("rs_c", "character", word="사람"), _row("rs_l", "location", word="이발소"), _row("rs_p", "prop", word="됫박"), _row("rs_lp", "location_part", word="회전 간판", fb=FB_LP), _row("rs_o", "outlook", word="긴 겉옷", fb=FB_O), ]}} PLAN_CP = {"data": {"decided": [ {"research_subject_id": "rs_c", "_short_id": "C01"}, {"research_subject_id": "rs_l", "_short_id": "L01"}, {"research_subject_id": "rs_p", "_short_id": "P01"}, ]}} PHASE3_CP = {"data": { "outlooks": [{"outlook_id": "O01"}], "scene_assignments": [{"scene_index": 1, "assignments": [ {"character_id": "C01", "outlook_id": "O01"}]}]}} class _Step: """CP 를 손으로 주는 대역 — **스텝 함수는 진짜**를 쓴다.""" def __init__(self, cps): self._cps = cps def _load_prev_checkpoint(self, step_id): return self._cps.get(step_id) def _step(**over): from app.core.steps.reference_acquisition_step import ( ReferenceAcquisitionStep as R) cps = {"grounding_screen": SCREEN_CP, "grounding_plan": PLAN_CP, "outlook_phase3": PHASE3_CP, **over} obj = _Step(cps) obj.central_obligations = R.central_obligations.__get__(obj) obj.central_result = R.central_result.__get__(obj) obj.central_wrap = R.central_wrap return obj class _Spy: def __init__(self): self.searched = 0 def search(self, **kw): self.searched += 1 return {"queries": ["q"], "images": [{"image_url": f"https://x/{i}.jpg", "thumbnail_url": "", "source_website_url": "", "caption": ""} for i in range(3)]} def download(self, url, dest, fallback_url=""): from pathlib import Path Path(dest).write_bytes(b"x") return True def judge(self, got): return {"j1": {"verdicts": [ {"index": i, "object_type_match": ctp.TYPE_MATCH[0], "visible": True} for i in range(1, len(got) + 1)]}} class TestItIsReachableNow: """★★**뒤집은 시험** (2026-09-01 D 활성화) — 이제 그 mode 가 받아진다.""" def test_the_mode_is_accepted(self): from app.core import grounding_mode as gm assert gm.GROUNDING_MODE_V2_CHUNK in gm.GROUNDING_MODES assert gm.GROUNDING_MODE_V2_CHUNK not in gm.PLANNED_MODES def test_runtime_config_takes_it(self): from app.core import grounding_mode as gm got = gm.resolve_grounding_mode( {"grounding_mode": gm.GROUNDING_MODE_V2_CHUNK}) assert got == gm.GROUNDING_MODE_V2_CHUNK def test_it_is_still_one_step_not_a_new_one(self): """★새 스텝 id 를 안 만들었다 — 있던 자리를 채웠다.""" from app.core.step_manifest import STEP_MANIFEST assert "reference_acquisition" in STEP_MANIFEST assert "grounding_central_acquisition" not in STEP_MANIFEST def test_it_runs_after_outlook_and_before_the_scene(self): from app.core.step_catalog import STEP_CATALOG as C e = C["reference_acquisition"] assert C["outlook_phase3"].order < e.order < C["scene_detail"].order assert "outlook_phase3" in e.depends_on class TestTheThreeCheckpointsBecomeOneLedger: def test_every_owner_reaches_the_door(self): led = ci.build_ledger(SCREEN_CP, PLAN_CP, PHASE3_CP) assert set(ci.owners_reaching_the_door(led)) == set( MATERIALIZABLE_OWNER_TYPES) def test_the_final_ids_come_from_the_plan_checkpoint(self): got = ci.final_id_by_subject(PLAN_CP) assert got == {"rs_c": "C01", "rs_l": "L01", "rs_p": "P01"} def test_a_missing_screen_output_stops(self): with pytest.raises(ci.CentralInputsError, match="rows"): ci.build_ledger({}, PLAN_CP, PHASE3_CP) def test_a_missing_phase3_stops_when_outlooks_need_binding(self): """★★이 스텝은 phase3 **뒤**에 와야 한다 — 앞이면 아웃룩이 못 선다.""" with pytest.raises(ci.CentralInputsError, match="phase3"): ci.build_ledger(SCREEN_CP, PLAN_CP, None) def test_an_episode_with_no_outlook_does_not_need_phase3(self): """★★아웃룩이 없는 에피소드에서 **잘못 서면 안 된다** (Codex 09-01). 결속할 줄이 없으면 phase3 를 요구할 까닭이 없다. """ no_outlook = {"data": {"rows": [r for r in SCREEN_CP["data"]["rows"] if r["owner_type"] != "outlook"]}} led = ci.build_ledger(no_outlook, PLAN_CP, None) assert "outlook" not in ci.owners_reaching_the_door(led) assert len(led["rows"]) == 4 # ★그래도 나머지 넷은 그대로 간다 ob = ci.build_obligations(no_outlook, PLAN_CP, None) assert ro.purposes_of(ob)["L01"] == ["context"] def test_an_empty_phase3_is_not_the_same_as_a_missing_one(self): """★음성 대조 — 아웃룩 줄이 **있는데** 빈 phase3 면 못 이은 채 남는다.""" from app.modules.pipeline import grounding_acquisition_ledger as gl led = ci.build_ledger(SCREEN_CP, PLAN_CP, {"data": {"outlooks": [], "scene_assignments": []}}) row = [r for r in led["rows"] if r["owner_type"] == "outlook"][0] assert row["status"] == gl.UNRESOLVED assert row["final_id"] is None def test_the_obligations_split_context_and_detail(self): ob = ci.build_obligations(SCREEN_CP, PLAN_CP, PHASE3_CP) by = ro.purposes_of(ob) assert by["LP01"] == ["detail"] assert by["L01"] == ["context"] class TestTheStepRunsTheWholeThing: @pytest.fixture def done(self, tmp_path): from app.modules.pipeline import grounding_chunk_journal as cj st = _step() ob = st.central_obligations() spy = _Spy() res = st.central_result( ob, journal=cj.ChunkJournal(tmp_path / "j.json", contract={"v": 1}), cap=99, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge) return {"ob": ob, "res": res, "cp": st.central_wrap(ob, res), "spy": spy} def test_it_writes_one_checkpoint_with_every_row(self, done): data = done["cp"]["data"] assert data["target_count"] == len(done["ob"]["rows"]) assert set(data["owners_present"]) == set(MATERIALIZABLE_OWNER_TYPES) def test_nothing_is_lost(self, done): assert ca.ledger_coverage(done["ob"], done["res"])["ok"] def test_nobody_waits_for_a_person(self, done): assert ca.unfinished_rows(done["res"]) == [] assert all(r["downstream_blocked"] is False for r in done["cp"]["data"]["rows"]) def test_there_is_exactly_one_buyer(self, done): """★조사 호출은 **중앙 한 곳**에서만 나간다.""" pur = done["cp"]["data"]["purchases"] assert pur["dispatch_attempted_this_run"] == done["spy"].searched def test_the_contracts_are_stamped(self, done): data = done["cp"]["data"] assert data["obligation_contract"] == ro.OBLIGATION_CONTRACT_VERSION assert data["inputs_contract"] == ci.CENTRAL_INPUTS_CONTRACT_VERSION assert data["wiring_contract"] == ca.CONTRACT_VERSION def test_what_was_not_found_is_named(self, done): data = done["cp"]["data"] assert data["selected_count"] + len(data["unresolved"]) == \ data["target_count"] # ★대상 수는 **참조가 필요했던 줄**이지 장부 줄 전체가 아니다 assert data["target_count"] + data["not_applicable_count"] == \ data["ledger_rows"] class TestNotApplicableIsNotAFailedSearch: """★★Codex 재현 (09-01) — 「애초에 필요 없었다」를 「못 구했다」로 세면 앞에서 갈라 놓은 구분을 **이 자리에서 다시 뭉갠다**. """ def _wrap(self, rows_in): from app.core.steps.reference_acquisition_step import ( ReferenceAcquisitionStep as R) ob = {"rows": rows_in} res = {"rows": [{"research_subject_id": r["research_subject_id"], "identity": None, "disposition": (ca.DISP_NOT_APPLICABLE if r["screen"] != gs.SCREEN_OBLIGATION else ca.DISP_AUTO_DONE), "status": None, "why_unbought": None, "why": "", "outcome": (None if r["screen"] != gs.SCREEN_OBLIGATION else "reference_unavailable"), "downstream_blocked": False, "acquisition": None, "source_evidence": {}, "ledger_row": r} for r in rows_in]} return R.central_wrap(ob, res)["data"] def _ledger_row(self, rsid, screen): from app.modules.pipeline import grounding_acquisition_ledger as gl return {**gl.row(owner_type="prop", research_subject_id=rsid, screen=screen, status=gl.RESOLVED, final_id="P01"), "purpose": None, "covers": ["P01"]} def test_a_not_target_row_is_not_counted_as_missing(self): d = self._wrap([self._ledger_row("not", gs.SCREEN_NOT_TARGET)]) assert d["target_count"] == 0 assert d["unresolved"] == [], "★비대상이 「못 구함」으로 세어졌다" assert d["not_applicable_count"] == 1 assert d["ledger_rows"] == 1 def test_a_needed_one_that_was_not_found_is_counted(self): """★음성 대조 — 세지 **않기만** 하면 진짜 실패도 사라진다.""" d = self._wrap([self._ledger_row("need", gs.SCREEN_OBLIGATION)]) assert d["target_count"] == 1 assert d["unresolved"] == ["need"] assert d["not_applicable_count"] == 0 def test_the_two_are_told_apart_in_one_run(self): d = self._wrap([self._ledger_row("not", gs.SCREEN_NOT_TARGET), self._ledger_row("need", gs.SCREEN_OBLIGATION)]) assert d["target_count"] == 1 and d["not_applicable_count"] == 1 assert d["unresolved"] == ["need"] assert d["selected_count"] + len(d["unresolved"]) == d["target_count"] def test_the_verdict_comes_from_the_one_place(self): """★판단을 여기서 다시 적지 않는다.""" import ast import inspect from app.core.steps.reference_acquisition_step import ( ReferenceAcquisitionStep as R) # ★`@staticmethod` 는 소스가 들여쓰인 채 나온다 — 그대로 파싱하면 # IndentationError 다. 들여쓰기를 벗겨서 본다. import textwrap tree = ast.parse(textwrap.dedent(inspect.getsource(R.central_wrap))) calls = {ast.unparse(n.func) for n in ast.walk(tree) if isinstance(n, ast.Call)} assert "ca.acquisition_outcome_of" in calls class TestItRefusesToWriteABrokenCheckpoint: def test_a_lost_row_stops(self, tmp_path): st = _step() ob = st.central_obligations() with pytest.raises(RuntimeError, match="사라졌다"): st.central_wrap(ob, {"rows": []}) def test_a_row_that_waits_for_a_person_stops(self, tmp_path): st = _step() ob = st.central_obligations() rows = [{"research_subject_id": r["research_subject_id"], "disposition": ca.DISP_ACQUIRED, "outcome": None, "downstream_blocked": True} for r in ob["rows"]] with pytest.raises(RuntimeError, match="HITL"): st.central_wrap(ob, {"rows": rows}) class TestTheLegacyBranchIsUntouched: def test_it_still_stands_when_targets_exist(self): """★옛 갈래는 **한 글자도 안 바뀌었다** — 여전히 승인 앞에서 선다.""" import inspect from app.core.steps import reference_acquisition_step as step src = inspect.getsource(step.ReferenceAcquisitionStep._execute) assert "NotImplementedError" in src assert "_targets" in src