"""중앙 입구 **갈래 중립 계약**. ★유료 0. Codex BLOCK (2026-09-01) — 「중앙 참조 조사기의 입구·회계를 owner-neutral 계약으로 한 번만 일반화하십시오. owner 별 결속은 resolver 로 남기되, 공통 ledger 하나가 중앙으로 갑니다.」 """ from __future__ import annotations import pytest from app.modules.pipeline import grounding_acquisition_ledger as gl 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"}}]} PL = {"coarse_type_label": "가", "visual_brief": "나", "search_terms_native": ["다"], "language_lock_native": "ko"} def _screen_row(rsid, owner, screen=gs.SCREEN_OBLIGATION, fb=None): r = {"research_subject_id": rsid, "owner_type": owner, "screen": screen, "source_evidence": dict(EV), PRODUCER_PAYLOAD: dict(PL)} if fb is not None: r[FACET_BINDING] = fb return r class TestTheCommonRowShape: def test_it_names_the_fields_codex_asked_for(self): for f in ("owner_type", "research_subject_id", "screen", "status", "final_id", "parent_final_id", "source_evidence"): assert f in gl.ROW_FIELDS def test_a_resolved_row_without_a_final_id_stops(self): """★「무엇에 붙일지 안다」가 해소의 뜻인데 그 값이 없으면 아닌 것이다.""" with pytest.raises(gl.LedgerContractError, match="final_id"): gl.row(owner_type="prop", research_subject_id="x", screen=gs.SCREEN_OBLIGATION, status=gl.RESOLVED) @pytest.mark.parametrize("bad", [ {"owner_type": "무엇인가"}, {"status": "corrupt"}, {"screen": "모르는 판별"}]) def test_an_unknown_value_stops(self, bad): kw = {"owner_type": "prop", "research_subject_id": "x", "screen": gs.SCREEN_NOT_TARGET, "status": gl.UNRESOLVED} kw.update(bad) with pytest.raises(gl.LedgerContractError): gl.row(**kw) def test_the_status_value_did_not_change(self): """★값을 바꾸면 이미 적힌 장부를 못 읽는다 — 이름만 중립으로 바꿨다.""" from app.modules.pipeline import grounding_outlook_binding as ob assert gl.RESOLVED == "bound" == ob.BOUND class TestEntityBackedOwnersResolveDirectly: def test_the_three_base_owners_resolve_from_the_registered_row(self): rows = [_screen_row("a", "character"), _screen_row("b", "location"), _screen_row("c", "prop")] got = gl.resolve_entity_backed( rows, final_id_by_subject={"a": "C01", "b": "L01", "c": "P01"}) assert [r["status"] for r in got["rows"]] == [gl.RESOLVED] * 3 assert [r["final_id"] for r in got["rows"]] == ["C01", "L01", "P01"] assert all(r["source_evidence"]["source_quote"] == "원문" for r in got["rows"]) assert all(r[PRODUCER_PAYLOAD]["coarse_type_label"] == "가" for r in got["rows"]) def test_no_registered_row_is_unresolved_not_invented(self): got = gl.resolve_entity_backed([_screen_row("a", "location")], final_id_by_subject={}) assert got["rows"][0]["status"] == gl.UNRESOLVED assert got["rows"][0]["why"] == gl.WHY_NO_ENTITY_ROW assert got["rows"][0]["final_id"] is None def test_a_final_id_of_the_wrong_lane_is_refused(self): """★접두를 손으로 안 견준다 — 계약에 묻는다.""" got = gl.resolve_entity_backed([_screen_row("a", "location")], final_id_by_subject={"a": "LP01"}) assert got["rows"][0]["status"] == gl.UNRESOLVED assert got["rows"][0]["why"] == gl.WHY_OWNER_MISMATCH assert got["rows"][0]["seen_final_id"] == "LP01" def test_a_facet_owner_may_not_come_here(self): for owner in ("location_part", "outlook"): with pytest.raises(gl.LedgerContractError): gl.resolve_entity_backed([_screen_row("a", owner)], final_id_by_subject={"a": "LP01"}) class TestFacetRowsResolveByCoordinate: FB = {"local_id": "c0#2", "final_id": "LP01", "owner_type": "location_part", "parent_local_id": "c0#1", "parent_owner_type": "location", "parent_final_id": "L01"} def test_it_binds_by_the_coordinate_the_producer_left(self): got = gl.resolve_facet_rows( [_screen_row("a", "location_part", fb=self.FB)]) r = got["rows"][0] assert r["status"] == gl.RESOLVED assert r["final_id"] == "LP01" and r["parent_final_id"] == "L01" def test_no_coordinate_is_unresolved(self): got = gl.resolve_facet_rows([_screen_row("a", "location_part")]) assert got["rows"][0]["why"] == gl.WHY_NO_FACET_COORDINATE def test_a_coordinate_of_the_wrong_lane_is_refused(self): got = gl.resolve_facet_rows([_screen_row( "a", "location_part", fb={**self.FB, "final_id": "L01"})]) assert got["rows"][0]["why"] == gl.WHY_OWNER_MISMATCH class TestTheMergedLedgerCarriesEveryLane: def test_all_five_owners_can_reach_the_central_door(self): """★★여기가 BLOCK 의 핵심 — 앞 판은 outlook 만 들어올 수 있었다.""" from app.modules.pipeline import grounding_outlook_binding as ob base = gl.resolve_entity_backed( [_screen_row("a", "character"), _screen_row("b", "location"), _screen_row("c", "prop")], final_id_by_subject={"a": "C01", "b": "L01", "c": "P01"}) facet = gl.resolve_facet_rows([_screen_row( "d", "location_part", fb=TestFacetRowsResolveByCoordinate.FB)]) out = ob.bind( [_screen_row("e", "outlook", fb={"local_id": "c0#3", "final_id": "O01", "owner_type": "outlook", "parent_local_id": "c0#1", "parent_owner_type": "character", "parent_final_id": "C01"})], {"outlooks": [{"outlook_id": "O01"}], "scene_assignments": [{"scene_index": 1, "assignments": [ {"character_id": "C01", "outlook_id": "O01"}]}]}) led = gl.merge(base, facet, out) assert set(gl.owners_present(led)) == set(MATERIALIZABLE_OWNER_TYPES) assert gl.accounting(led)["rows"] == 5 assert len(gl.acquisition_targets(led)) == 5 # ★★공통 칸이 **모든 갈래에** 있어야 소비하는 쪽이 안 깨진다 for r in led["rows"]: for f in gl.ROW_FIELDS: assert f in r, f"{r['owner_type']} 에 {f} 칸이 없다" by = {r["owner_type"]: r for r in led["rows"]} assert by["outlook"]["final_id"] == "O01" assert by["outlook"]["parent_final_id"] == "C01" assert by["location_part"]["parent_final_id"] == "L01" assert by["prop"]["parent_final_id"] is None def test_an_unresolved_row_still_has_the_common_fields(self): """★「없다」와 「칸이 없다」를 안 섞는다.""" from app.modules.pipeline import grounding_outlook_binding as ob out = ob.bind([_screen_row("x", "outlook")], {"outlooks": []}) r = out["rows"][0] assert r["status"] == gl.UNRESOLVED for f in gl.ROW_FIELDS: assert f in r, f"{f} 칸이 없다" assert r["final_id"] is None and r["parent_final_id"] is None def test_the_same_subject_from_two_resolvers_stops(self): """★소유가 겹치면 **같은 대상을 두 번 조사한다**.""" a = gl.resolve_entity_backed([_screen_row("dup", "prop")], final_id_by_subject={"dup": "P01"}) b = gl.resolve_facet_rows([_screen_row( "dup", "location_part", fb=TestFacetRowsResolveByCoordinate.FB)]) with pytest.raises(gl.LedgerContractError, match="둘 다"): gl.merge(a, b) def test_splitting_refuses_an_unknown_lane(self): with pytest.raises(gl.LedgerContractError): gl.split_by_owner([_screen_row("a", "무엇인가")]) def test_splitting_covers_every_known_lane(self): got = gl.split_by_owner([_screen_row("a", o) for o in MATERIALIZABLE_OWNER_TYPES]) assert set(got) == set(MATERIALIZABLE_OWNER_TYPES) assert all(len(v) == 1 for v in got.values())