"""하류가 갈래 표를 **계약에서 파생**하나. ★유료 0. Codex (2026-09-01) — 「일곱 군데 손목록을 각각 늘리지 말고 `ENTITY_KEY_TO_OWNER`/owner→reference-kind·role 중앙 투영에서 파생시키십시오. `ctx.entities` 와 `detail_steps` 사본도 같은 계약을 소비하게 해야 합니다.」 """ from __future__ import annotations import ast import inspect from app.core.steps import detail_steps as ds from app.modules.pipeline.grounding_carry import ENTITY_KEY_TO_OWNER def _fn(name): for n in ast.walk(ast.parse(inspect.getsource(ds))): if isinstance(n, ast.FunctionDef) and n.name == name: return n raise AssertionError(f"★{name} 을 못 찾았다") class TestTheVisibleEntityLoopUsesTheContract: NAME = "_build_visible_entity_details" def _the_loop(self): """visible_entity_details 를 만드는 loop 의 **반복 대상**을 찾는다.""" tree = ast.parse(inspect.getsource(ds)) for n in ast.walk(tree): if not isinstance(n, ast.For): continue tgt = ast.unparse(n.target) if "_etype_key" in tgt and "_etype_val" in tgt: return ast.unparse(n.iter) raise AssertionError("★갈래 loop 를 못 찾았다") def test_it_iterates_the_contract_not_a_hand_list(self): it = self._the_loop() assert it == "ENTITY_KEY_TO_OWNER", \ f"★갈래 표를 손으로 적었다: {it}" def test_the_contract_already_has_the_fourth_lane(self): assert ("location_parts", "location_part") in ENTITY_KEY_TO_OWNER assert len(ENTITY_KEY_TO_OWNER) == 4 def test_no_lane_name_is_written_by_hand_in_that_loop(self): """★`"locations"` 같은 글자가 loop 안에 남아 있으면 사본이다.""" tree = ast.parse(inspect.getsource(ds)) names = {k for k, _o in ENTITY_KEY_TO_OWNER} for n in ast.walk(tree): if not (isinstance(n, ast.For) and "_etype_key" in ast.unparse(n.target)): continue lits = {c.value for c in ast.walk(n) if isinstance(c, ast.Constant) and isinstance(c.value, str)} assert not (lits & names), f"★갈래 이름을 손으로 적었다: {lits & names}" class TestTheEntityChainCarriesTheFourthLane: """★★D-inert (Codex 09-01) — 네 번째 칸이 엔티티 사슬을 **지나가되**, 갈래가 셋뿐인 주행은 **한 글자도 안 달라야** 한다. ★앞 판은 `entity_detail` 과 `entity_t2i` 가 세 갈래를 **손으로 적어** 둬서 v2_chunk producer 가 낸 `location_parts` 가 거기서 통째로 사라졌다. """ def test_the_lane_table_comes_from_the_contract(self): from app.core.steps.entity_steps import _ENTITY_KEYS assert list(_ENTITY_KEYS()) == list(ENTITY_KEY_TO_OWNER) def test_three_lanes_look_exactly_like_before(self): """★★무변 — 칸 이름도 **차례도** 같아야 JSON 이 같다.""" from app.core.steps.entity_steps import _by_lane done = {"가": {"entity_type": "character"}, "나": {"entity_type": "location"}, "다": {"entity_type": "prop"}} assert list(_by_lane(done)) == ["characters", "locations", "props"] assert list(_by_lane({})) == ["characters", "locations", "props"], \ "★빈 주행에서 기존 칸이 사라지면 옛 소비자가 깨진다" def test_the_fourth_key_appears_only_when_it_has_something(self): from app.core.steps.entity_steps import _by_lane done = {"가": {"entity_type": "character"}, "라": {"entity_type": "location_part"}} got = _by_lane(done) assert list(got)[-1] == "location_parts" assert [v["entity_type"] for v in got["location_parts"]] == \ ["location_part"] def test_the_detail_queue_takes_the_fourth_lane_too(self): """★큐에 안 들어가면 그 갈래는 상세도 T2I 도 못 받는다.""" import ast import inspect from app.core.steps import entity_steps as es tree = ast.parse(inspect.getsource(es.EntityDetailStep)) loops = [ast.unparse(n.iter) for n in ast.walk(tree) if isinstance(n, ast.For) and "etype_key" in ast.unparse(n.target)] assert loops == ["_ENTITY_KEYS()"], f"★손으로 적었다: {loops}" def test_the_t2i_step_writes_no_lane_name_by_hand(self): import ast import inspect from app.core.steps import entity_steps as es names = {k for k, _o in ENTITY_KEY_TO_OWNER} lits = {n.value for n in ast.walk( ast.parse(inspect.getsource(es.EntityT2iStep))) if isinstance(n, ast.Constant) and isinstance(n.value, str)} assert not (lits & names), f"★갈래 이름을 적었다: {lits & names}" def test_the_legacy_extract_fallback_is_three_steps_not_a_lane_table(self): """★★래칫을 **지적 범위 밖으로** 넓히지 않는다. `entity_detail` 에 남은 세 이름은 **갈래 표가 아니라** 세 개의 다른 스텝 체크포인트(`entity_extract_character`/`_location`/`_prop`)를 읽는 자리다. 네 번째 추출 스텝은 **없다** — 그것을 계약에서 파생시키려면 없는 스텝을 지어내야 한다. 그래서 그대로 둔다. """ import ast import inspect from app.core.steps import entity_steps as es src = inspect.getsource(es.EntityDetailStep) cps = {n.args[0].value for n in ast.walk(ast.parse(src)) if isinstance(n, ast.Call) and ast.unparse(n.func).endswith("_load_prev_checkpoint") and n.args and isinstance(n.args[0], ast.Constant)} assert {"entity_extract_character", "entity_extract_location", "entity_extract_prop"} <= cps assert "entity_extract_location_part" not in cps, \ "★네 번째 추출 스텝이 생겼다 — 이 시험을 뒤집어라" def test_the_v2_projection_already_emits_it(self): """★내는 쪽 — chunk producer 의 투영은 이미 네 갈래다.""" from app.modules.pipeline import grounding_chunk_adapter as ad from app.modules.pipeline import grounding_chunk_merge as cm rows = [{"local_id": "c0#0", "owner_type": "location_part", "surface_form": "나", "occurrences": [{"source_span": {"segment_id": "s1", "start": 1, "end": 2}, "source_quote": "나"}], "shot_binding_status": "bound_complete", "shot_appearance_ids": ["a", "b"], "hard_to_generate": True, "viewers_would_notice": True}] red = cm.reduce_episode(rows, [], segments={"s1": "가나"}) assert "location_parts" in ad.to_entity_rows(red)["rows"] class TestAddingTheLaneChangesNothingYet: """★★더하기만 한다 — `ctx.entities` 에 그 칸이 없으면 무동작이다.""" def test_the_gate_is_open_now(self): """★★**뒤집었다** (2026-09-01 D 활성화) — 이제 그 갈래가 참조를 만든다. 앞에는 「아직 안 열렸다」를 잠갔는데 그것이 이 커밋이 바꾸는 것이다. """ from app.modules.pipeline import grounding_entity_contract as ec assert "location_part" in ec.REFERENCE_SUPPORTED_OWNERS assert ec.reference_owner_of("LP01") == "location_part" def test_a_missing_key_is_skipped_not_an_error(self): """★`ctx.entities.get(...) or []` 라 칸이 없으면 그냥 지나간다.""" tree = ast.parse(inspect.getsource(ds)) found = False for n in ast.walk(tree): if not (isinstance(n, ast.For) and "_etype_key" in ast.unparse(n.target)): continue for inner in ast.walk(n): if (isinstance(inner, ast.For) and ".get(_etype_key)" in ast.unparse(inner.iter)): assert "or []" in ast.unparse(inner.iter) found = True assert found, "★칸이 없을 때를 어떻게 다루는지 못 봤다"