"""C(c) → `entity_merge` 모양 adapter. ★유료 0 · 아직 **안 켠 것**이다. `entity_merge` 체크포인트를 읽는 모듈이 12개다. 그래서 새 producer 는 모양을 바꾸지 않고 **기존 모양으로 낸다**. 여기서 재는 것은 「옮길 것만 옮기고, 안 옮긴 것은 **왜 안 옮겼는지** 남기나」다. """ from __future__ import annotations import json from pathlib import Path import pytest from app.modules.pipeline import grounding_chunk_adapter as ad from app.modules.pipeline.grounding_entity_contract import OWNER_PREFIX def _reduced(rows, registered): return {"rows": rows, "registered": registered} def _row(lid, owner, form, **over): d = {"local_id": lid, "owner_type": owner, "surface_form": form, "visual_brief": "겉모습 한 문장", "shot_binding_status": "bound_complete", "shot_appearance_ids": ["s1#0"], "hard_to_generate": True, "viewers_would_notice": True, "occurrences": [{"source_span": {"segment_id": "scene-1", "start": 0, "end": 2}, "source_quote": form[:2]}], "evidence_quotes": [f"{form} 의 겉모습 근거"], "search_terms_native": ["낱말"], "language_lock_native": "ko"} d.update(over) return d def _reg(registered=True, reason="shot_appearances=3", final_id=None, owner="prop", n=1): """★`final_id` 는 **장부가 정한 신원**이다 — 갈래 접두와 맞아야 한다.""" if final_id is None and registered is True: final_id = f"{OWNER_PREFIX[owner]}{n:02d}" return {"registered": registered, "reason": reason, "final_id": final_id, "disposition": ("registered" if registered is True else "not_registered" if registered is False else "registration_unresolved")} class TestItDoesNotInventADescription: """★★`visual_brief` 를 `description` 자리에 **안 넣는다**. 둘 다 「겉모습」을 말하지만 `description` 에는 **단일 시각 형태 원칙**이 붙어 있고(「A이거나 B」 금지) 그 글이 **T2I 프롬프트에 그대로** 들어간다. `visual_brief` 는 그 계약을 받은 적이 없다. """ def test_description_is_empty_and_the_brief_is_kept_elsewhere(self): got = ad.to_entity_rows(_reduced( [_row("c0#1", "prop", "가방", visual_brief="낡은 가죽 가방")], {"c0#1": _reg()})) row = got["rows"]["props"][0] assert row["description"] == "", \ "★겉모습 글이 단일 시각 형태 계약 없이 T2I 로 들어간다" assert row["visual_traits"] == [] assert row["grounding_provenance"]["visual_brief"] == "낡은 가죽 가방", \ "★버리지는 않는다 — 자리가 달라서 T2I 정본으로 안 읽힐 뿐이다" def test_the_brief_never_lands_in_a_t2i_bound_field(self): brief = "겉모습 한 문장" got = ad.to_entity_rows(_reduced( [_row("c0#1", "prop", "가방")], {"c0#1": _reg()})) row = got["rows"]["props"][0] for k, v in row.items(): if k == "grounding_provenance": continue assert brief not in str(v), f"★{k} 에 겉모습 글이 샜다" class TestFacetsDoNotSneakIntoBaseTypes: """★facet 을 base 갈래로 **우회 등록하지 않는다**. ★★2026-09-01 (§2-6.5) 에 **반쪽이 뒤집혔다.** `location_part` 는 이제 제 갈래로 서고 `location_parts` 라는 **1급 sibling 키**를 갖는다 (Codex ⓐ: 「`locations` 안에 넣으면 owner/LP 신원과 별도 `part_of` SOT 를 잃고 location 강등이 된다」). `outlook` 은 여전히 자리가 없다 — `OutlookSyncService` 가 `character_outlook` 으로 따로 쓴다. """ def test_an_outlook_still_becomes_a_skip_not_a_base_row(self): got = ad.to_entity_rows(_reduced( [_row("c0#1", "outlook", "무엇")], {"c0#1": _reg(owner="outlook")})) assert got["rows"] == {}, "★outlook 이 base 행이 됐다" assert got["skipped"][0]["skip"] == ad.SKIP_NO_MERGE_KEY assert got["skipped"][0]["owner_type"] == "outlook" def test_a_location_part_becomes_its_own_row_not_a_location(self): """★★**우회 등록 0건** — `locations` 가 아니라 `location_parts` 다.""" got = ad.to_entity_rows(_reduced( [_row("c0#1", "location_part", "무엇")], {"c0#1": _reg(owner="location_part")})) assert set(got["rows"]) == {"location_parts"}, got["rows"] assert got["skipped"] == [] assert got["rows"]["location_parts"][0]["short_id"].startswith("LP") def test_the_key_map_gives_the_part_its_own_slot(self): """★`outlook` 은 여전히 자리가 없다 — 주인이 다르다.""" assert "outlook" not in ad.OWNER_TO_MERGE_KEY assert ad.OWNER_TO_MERGE_KEY["location_part"] == "location_parts" assert ad.OWNER_TO_MERGE_KEY["location"] == "locations" class TestUnresolvedIsNotFolded: """★★미확정을 「미등록」으로 접으면 사람이 볼 것이 사라진다.""" def test_unresolved_and_not_registered_are_told_apart(self): got = ad.to_entity_rows(_reduced( [_row("c0#1", "prop", "가"), _row("c0#2", "prop", "나")], {"c0#1": _reg(None, "exception_unresolved"), "c0#2": _reg(False, "shot_appearances=1")})) skips = {s["local_id"]: s["skip"] for s in got["skipped"]} assert skips == {"c0#1": ad.SKIP_UNRESOLVED, "c0#2": ad.SKIP_NOT_REGISTERED} assert got["rows"] == {} def test_nothing_disappears_without_a_reason(self): rows = [_row(f"c0#{i}", "prop", f"x{i}") for i in range(5)] reg = {"c0#0": _reg(n=1), "c0#1": _reg(False), "c0#2": _reg(None), "c0#3": _reg(n=2), "c0#4": _reg(False)} got = ad.to_entity_rows(_reduced(rows, reg)) made = sum(len(v) for v in got["rows"].values()) assert made + len(got["skipped"]) == len(rows), \ "★조용히 사라진 행이 있다 — 다섯 갈래 끝점을 못 센다" class TestTheLedgerOwnsTheIdentity: """★★신원은 **장부가 이미 정했다** (Codex BLOCK 2026-08-31). 앞 판은 번호를 **새로 발급**했다. 그러면 같은 대상의 신원이 둘이 되고, `final_id` 로 걸린 참조 의무·하류 결속이 엉뚱한 행을 가리킨다. """ def test_short_id_is_the_ledger_final_id(self): got = ad.to_entity_rows(_reduced( [_row("c0#1", "prop", "가")], {"c0#1": _reg(final_id="P07")})) assert got["rows"]["props"][0]["short_id"] == "P07", \ "★장부가 준 신원을 안 쓰고 새로 매겼다" def test_a_collision_stops_instead_of_renumbering(self): """★앞 판은 이 자리에서 **다음 번호로 비켜 갔다** — 그것이 결함이다. 비켜 가면 장부와 갈라진다. 두 벌이 선 상태이니 사람이 봐야 한다. """ with pytest.raises(ad.ShortIdCollision, match="다시 매기지"): ad.to_entity_rows( _reduced([_row("c0#1", "prop", "가")], {"c0#1": _reg(final_id="P01")}), existing={"props": [{"short_id": "P01"}]}) def test_registered_without_a_final_id_stops(self): """★앞 판은 **건너뛰었다** — 그러면 등록된 엔티티가 조용히 사라진다. Codex BLOCK: durable CP·재생 경계에서 어긋난 것이 오면 흘려보내지 말고 **선다**. 건너뛰기는 「없던 일」로 만드는 것이라 아무도 못 본다. """ with pytest.raises(ad.LedgerContractViolation, match="지어내지 않는다"): ad.to_entity_rows(_reduced( [_row("c0#1", "prop", "가")], {"c0#1": {"registered": True, "reason": "x", "final_id": None}})) def test_existing_rows_are_not_touched(self): olds = [{"short_id": "P01", "name": "옛것"}] ad.to_entity_rows( _reduced([_row("c0#1", "prop", "가")], {"c0#1": _reg(final_id="P09")}), existing={"props": olds}) assert olds == [{"short_id": "P01", "name": "옛것"}] class TestSameInputSameBytes: """★지문이 서려면 **바이트가 같아야** 한다.""" def test_two_runs_are_byte_identical(self): rows = [_row(f"c{i%3}#{9-i}", "prop", f"x{i}") for i in range(9)] reg = {r["local_id"]: _reg(n=i) for i, r in enumerate(rows, 1)} a = ad.to_entity_rows(_reduced(rows, reg)) b = ad.to_entity_rows(_reduced(list(reversed(rows)), reg)) assert json.dumps(a, ensure_ascii=False, sort_keys=True) == \ json.dumps(b, ensure_ascii=False, sort_keys=True), \ "★행 순서가 바뀌면 산출이 달라진다 — 지문이 안 선다" class TestTheNameKeepsItsCase: def test_normalization_does_not_lowercase(self): got = ad.to_entity_rows(_reduced( [_row("c0#1", "prop", " iPhone  상자 ")], {"c0#1": _reg()})) assert got["rows"]["props"][0]["name"] == "iPhone 상자", \ "★비교용 정규화를 이름에 쓰면 대소문자가 죽는다" class TestAgainstTheRealPaidOutput: """★★합성만으로 끝내지 않는다 — **실제로 산 산출**에 물려 본다.""" FROZEN = Path(__file__).resolve().parents[3] / "artifact" \ / "20260831_cc_preflight" / "c_live_j_run.json" def _reduced(self): if not self.FROZEN.exists(): pytest.skip("얼어붙은 유료 산출이 없다") return json.loads(self.FROZEN.read_text(encoding="utf-8"))["reduced"] def test_every_row_is_either_carried_or_explained(self): red = self._reduced() got = ad.to_entity_rows(red) made = sum(len(v) for v in got["rows"].values()) assert made + len(got["skipped"]) == len(red["rows"]) def test_no_facet_reached_a_base_type(self): red = self._reduced() got = ad.to_entity_rows(red) by_lid = {str(r["local_id"]): r for r in red["rows"]} for key, rows in got["rows"].items(): for r in rows: owner = by_lid[r["grounding_provenance"]["local_id"]][ "owner_type"] assert ad.OWNER_TO_MERGE_KEY[owner] == key, \ f"★{owner} 가 {key} 에 들어갔다" def test_short_ids_are_unique_within_each_type(self): got = ad.to_entity_rows(self._reduced()) for key, rows in got["rows"].items(): ids = [r["short_id"] for r in rows] assert len(ids) == len(set(ids)), f"★{key} 에 겹치는 번호가 있다" def test_the_carried_count_matches_the_registered_base_owners(self): """★수를 **손으로 다시 세어** 맞댄다 — adapter 가 세 준 수를 안 믿는다.""" red = self._reduced() by_lid = {str(r["local_id"]): r for r in red["rows"]} want = sum(1 for lid, rec in red["registered"].items() if rec.get("registered") is True and by_lid.get(lid, {}).get("owner_type") in ad.OWNER_TO_MERGE_KEY) got = ad.to_entity_rows(red) assert sum(len(v) for v in got["rows"].values()) == want class TestProvenanceSurvivesToTheRow: """★★D 에서 `entity_merge` 호환 CP 가 하류 정본이 되면, 「왜 이 엔티티· 의무가 생겼나」를 **그 행에서** 되짚을 수 있어야 한다 (Codex BLOCK). 앞 판은 겉모습 글·샷·두 축·사유만 실어서 **정본 인용(span+quote)과 겉모습 근거, 처분이 통째로 빠졌다.** """ FROZEN = Path(__file__).resolve().parents[3] / "artifact" \ / "20260831_cc_preflight" / "c_live_j_run.json" def _red(self): if not self.FROZEN.exists(): pytest.skip("얼어붙은 유료 산출이 없다") return json.loads(self.FROZEN.read_text(encoding="utf-8"))["reduced"] def test_every_carried_row_can_be_traced_back(self): red = self._red() by = {str(r["local_id"]): r for r in red["rows"]} got = ad.to_entity_rows(red) n = 0 for rows in got["rows"].values(): for row in rows: pv = row["grounding_provenance"] src = by[pv["local_id"]] assert pv["final_id"] == row["short_id"] assert pv["owner_type"] == src["owner_type"] assert pv["registration"] == red["registered"][pv["local_id"]] n += 1 # ★★69 → **89** (§2-6.5, 2026-09-01). `location_part` 20 행이 # 「건너뜀」에서 「옮김」으로 갔다 — 그 갈래가 durable debt 이던 것이 # 풀린 자리다. 나머지 9 행은 등록 문턱 미달로 남는데, 그것은 다른 # 갈래와 **같은 이유**지 갈래 때문이 아니다. assert n == 89, f"★실제 유료 산출에서 옮긴 행이 89 가 아니라 {n}" def test_the_original_quotes_and_spans_are_still_there(self): """★인용만 있고 **자리(span)** 가 없으면 원문에서 되찾을 수 없다.""" red = self._red() by = {str(r["local_id"]): r for r in red["rows"]} got = ad.to_entity_rows(red) checked = 0 for rows in got["rows"].values(): for row in rows: pv = row["grounding_provenance"] want = by[pv["local_id"]]["occurrences"] assert len(pv["occurrences"]) == len(want) for a, b in zip(pv["occurrences"], want): assert a["source_quote"] == b["source_quote"] assert a["source_span"] == b["source_span"] assert a["source_span"].get("segment_id") assert isinstance(a["source_span"].get("start"), int) checked += 1 assert checked > 0, "★인용이 하나도 안 실렸다 — 빈손은 모든 축을 지난다" def test_the_evidence_quotes_are_still_there(self): red = self._red() by = {str(r["local_id"]): r for r in red["rows"]} got = ad.to_entity_rows(red) kept = 0 for rows in got["rows"].values(): for row in rows: pv = row["grounding_provenance"] assert pv["evidence_quotes"] == list( by[pv["local_id"]].get("evidence_quotes") or ()) kept += len(pv["evidence_quotes"]) assert kept > 0, "★겉모습 근거가 한 줄도 안 실렸다" def test_provenance_never_bleeds_into_the_t2i_fields(self): """★`description`·`visual_traits` 에 새면 T2I 프롬프트로 나간다.""" red = self._red() got = ad.to_entity_rows(red) for rows in got["rows"].values(): for row in rows: assert row["description"] == "" assert row["visual_traits"] == [] # ★행에 남은 `grounding_*` 칸은 **한 벌**뿐이어야 한다 keys = {k for rows in got["rows"].values() for r in rows for k in r if k.startswith("grounding")} assert keys == {"grounding_provenance"}, f"★흩어진 칸: {keys}" class TestTheInputContractIsCheckedBeforeConsuming: """★★`reduce_episode` 밖(durable CP·재생)에서 어긋난 것이 오면 **선다**. 조용히 흘리면 ①장부에만 있는 등록 행이 사라지고 ②`character` 인데 `P01` 같은 어긋난 신원이 그대로 나가 참조 의무가 엉뚱한 행에 걸린다. """ def test_a_row_without_a_ledger_entry_stops(self): with pytest.raises(ad.LedgerContractViolation, match="1:1"): ad.to_entity_rows(_reduced([_row("c0#1", "prop", "가")], {})) def test_a_ledger_entry_without_a_row_stops(self): """★이것이 **조용히 사라지던** 쪽이다.""" with pytest.raises(ad.LedgerContractViolation, match="장부에만"): ad.to_entity_rows(_reduced([], {"c0#1": _reg()})) def test_a_prefix_that_disagrees_with_the_owner_stops(self): with pytest.raises(ad.LedgerContractViolation, match="엉뚱한 행"): ad.to_entity_rows(_reduced( [_row("c0#1", "character", "누구")], {"c0#1": _reg(final_id="P01")})) def test_a_duplicated_final_id_stops(self): with pytest.raises(ad.LedgerContractViolation, match="전역 중복"): ad.to_entity_rows(_reduced( [_row("c0#1", "prop", "가"), _row("c0#2", "prop", "나")], {"c0#1": _reg(final_id="P01"), "c0#2": _reg(final_id="P01")})) def test_a_duplicated_local_id_stops(self): with pytest.raises(ad.LedgerContractViolation, match="두 번"): ad.to_entity_rows(_reduced( [_row("c0#1", "prop", "가"), _row("c0#1", "prop", "나")], {"c0#1": _reg()})) def test_the_check_uses_the_central_prefix_contract(self): """★이름·regex 로 안 본다 — 중앙 계약이 바뀌면 검사도 따라간다. ★★**설명 문장은 빼고 본다.** 앞 판은 글자로 찾아서 이 함수의 주석 「이름·regex 안 쓴다」를 **위반으로 읽었다** — 그러면 왜 그렇게 했는지 설명을 지워야 시험이 통과한다. 오늘 네 번째로 같은 부류다. """ import ast import inspect tree = ast.parse(inspect.getsource(ad.validate_ledger)) for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Module)): body = getattr(node, "body", []) if (body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant) and isinstance(body[0].value.value, str)): body.pop(0) # ★설명을 벗긴다 code = ast.unparse(tree) assert "OWNER_PREFIX" in code, "★중앙 계약을 안 쓴다" for banned in ("re.match", "re.search", "re.compile", "fullmatch"): assert banned not in code, f"★정규식으로 판정한다: {banned}" def test_the_real_paid_ledger_passes(self): """★막는 조건을 넓혔으면 **걸릴 정상 데이터**를 하나 대 본다.""" f = Path(__file__).resolve().parents[3] / "artifact" \ / "20260831_cc_preflight" / "c_live_j_run.json" if not f.exists(): pytest.skip("얼어붙은 유료 산출이 없다") ad.validate_ledger(json.loads(f.read_text(encoding="utf-8"))["reduced"]) class TestThePrefixTableIsNotPrefixFree: """★★`L` ⊂ `LP` — `startswith` 로 보면 `location` 행에 `LP01` 이 통과한다. Codex 가 잡았다 (2026-08-31). 접두표를 지을 때 `location_part` 를 `LP` 로 둔 것이 원인이고, **표를 고치는 대신** 가르는 함수를 계약 모듈 한 곳에 뒀다 — 표를 바꾸면 이미 저장된 신원이 다 어긋난다. """ def test_the_contract_resolves_by_longest_prefix(self): from app.modules.pipeline.grounding_entity_contract import ( owner_of_final_id) assert owner_of_final_id("LP01") == "location_part" assert owner_of_final_id("L01") == "location" assert owner_of_final_id("X9") is None def test_every_prefix_resolves_to_its_own_owner(self): """★표가 바뀌어도 이 시험이 따라간다 — 갈래를 손으로 안 적는다.""" from app.modules.pipeline.grounding_entity_contract import ( OWNER_PREFIX, owner_of_final_id) for owner, pre in OWNER_PREFIX.items(): assert owner_of_final_id(f"{pre}01") == owner, \ f"★{pre!r} 가 {owner!r} 로 안 돌아온다" def test_a_facet_id_on_a_base_row_stops(self): """★막던 자리 — 여기로 새면 참조 의무가 엉뚱한 행에 걸린다.""" with pytest.raises(ad.LedgerContractViolation, match="location_part.*의 것이다"): ad.to_entity_rows(_reduced( [_row("c0#1", "location", "방")], {"c0#1": _reg(final_id="LP01")})) def test_a_base_id_on_a_facet_row_stops(self): with pytest.raises(ad.LedgerContractViolation, match="의 것이다"): ad.to_entity_rows(_reduced( [_row("c0#1", "location_part", "선반")], {"c0#1": _reg(final_id="L01")})) def test_the_adapter_does_not_compare_prefixes_itself(self): """★계약 함수 한 곳이 가른다 — adapter 가 제 것을 만들면 두 벌이다.""" import ast import inspect tree = ast.parse(inspect.getsource(ad.validate_ledger)) for n in ast.walk(tree): if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute): assert n.func.attr != "startswith", \ "★adapter 가 접두를 제 손으로 비교한다" class TestSkippedRowsKeepTheirEvidence: """★★안 옮긴 행에도 근거가 붙어야 **다섯 갈래 감사**가 안 끊긴다. Codex 가 잡았다 (2026-08-31). 앞 판은 facet 23개와 미확정 30개가 「이름과 사유」만 남아, 어느 인용에서 온 무엇인지 adapter 출력에서 못 되짚었다. """ FROZEN = Path(__file__).resolve().parents[3] / "artifact" \ / "20260831_cc_preflight" / "c_live_j_run.json" def test_every_skipped_row_carries_its_provenance(self): if not self.FROZEN.exists(): pytest.skip("얼어붙은 유료 산출이 없다") red = json.loads(self.FROZEN.read_text(encoding="utf-8"))["reduced"] by = {str(r["local_id"]): r for r in red["rows"]} got = ad.to_entity_rows(red) # ★★72 → **52** — 같은 사실의 반대쪽이다 (LP 20 행이 옮겨졌다) assert len(got["skipped"]) == 52, \ f"★안 옮긴 행이 72 가 아니라 {len(got['skipped'])}" for sk in got["skipped"]: pv = sk["grounding_provenance"] src = by[pv["local_id"]] assert pv["occurrences"] and pv["occurrences"][0]["source_span"], \ f"★{sk['local_id']} 의 부른 자리가 없다" assert pv["evidence_quotes"] == list( src.get("evidence_quotes") or ()) assert pv["registration"] == red["registered"][pv["local_id"]] def test_the_facet_owners_can_still_be_audited_from_the_output(self): """★`location_part` 는 옮길 자리가 없다 — 그래도 **셀 수는 있어야** 한다.""" if not self.FROZEN.exists(): pytest.skip("얼어붙은 유료 산출이 없다") red = json.loads(self.FROZEN.read_text(encoding="utf-8"))["reduced"] got = ad.to_entity_rows(red) facet = [s for s in got["skipped"] if s["skip"] == ad.SKIP_NO_MERGE_KEY] owners = {s["owner_type"] for s in facet} # ★★`location_part` 는 이제 **자리가 있어서** 이 사유로 안 걸린다. # 남은 9 행은 `SKIP_NOT_REGISTERED` 로 다른 갈래와 같이 세어진다. assert owners == {"outlook"} assert all(s["grounding_provenance"]["occurrences"] for s in facet), \ "★facet 이 어느 인용에서 왔는지 산출에서 못 본다" class TestAllOneHundredFortyOneRowsCanBeTracedBack: """★★**옮긴 것만** 되짚을 수 있으면 가장 중요한 **빚의 근거**가 끊긴다. Codex 가 못박았다 (2026-08-31): 사용자는 다섯 갈래 **전부**를 요구했고, 특히 facet/debt·미확정이 사람이 볼 대상이다. 그래서 `made + skipped = 141` 로 그치지 않고 **각 `local_id` 의 인용·자리·근거·장부 기록이 원본과 같은지** 141행 전부에서 맞댄다. """ FROZEN = Path(__file__).resolve().parents[3] / "artifact" \ / "20260831_cc_preflight" / "c_live_j_run.json" def _both(self): if not self.FROZEN.exists(): pytest.skip("얼어붙은 유료 산출이 없다") red = json.loads(self.FROZEN.read_text(encoding="utf-8"))["reduced"] got = ad.to_entity_rows(red) out = {} for rows in got["rows"].values(): for r in rows: out[r["grounding_provenance"]["local_id"]] = \ r["grounding_provenance"] for sk in got["skipped"]: out[sk["grounding_provenance"]["local_id"]] = \ sk["grounding_provenance"] return red, out def test_every_single_row_appears_exactly_once(self): red, seen = self._both() want = {str(r["local_id"]) for r in red["rows"]} assert seen.keys() == want, \ f"★빠진 것 {sorted(want - seen.keys())[:5]} · " \ f"더 있는 것 {sorted(seen.keys() - want)[:5]}" assert len(seen) == 141 def test_every_row_keeps_its_quotes_spans_evidence_and_ledger(self): red, seen = self._both() by = {str(r["local_id"]): r for r in red["rows"]} for lid, pv in sorted(seen.items()): src = by[lid] assert pv["owner_type"] == src["owner_type"] assert pv["evidence_quotes"] == list( src.get("evidence_quotes") or ()) assert pv["registration"] == red["registered"][lid] want = src.get("occurrences") or [] assert len(pv["occurrences"]) == len(want) for a, b in zip(pv["occurrences"], want): assert a["source_quote"] == b["source_quote"] assert a["source_span"] == b["source_span"] assert pv["shot_appearance_ids"] == list( src.get("shot_appearance_ids") or ()) assert pv["hard_to_generate"] == src.get("hard_to_generate") assert pv["viewers_would_notice"] == src.get( "viewers_would_notice") def test_unregistered_rows_keep_an_empty_id_not_an_invented_one(self): """★없으면 **없는 채로** 적는다 — 지어내지 않는다.""" red, seen = self._both() n = 0 for lid, pv in seen.items(): rec = red["registered"][lid] if rec.get("registered") is True: assert pv["final_id"] == rec["final_id"] else: assert pv["final_id"] == "", f"★{lid} 에 없는 신원이 붙었다" n += 1 assert n == 49, f"★등록 안 된 행이 49 가 아니라 {n}" def test_all_five_owners_are_present_in_the_output(self): red, seen = self._both() by = {str(r["local_id"]): r for r in red["rows"]} owners = {by[lid]["owner_type"] for lid in seen} assert owners == {"prop", "character", "location", "location_part", "outlook"}, f"★갈래가 빠졌다: {sorted(owners)}" class TestTheIdGrammarIsOneContractAndItIsStrict: """★신원 문법은 **계약 모듈 한 벌**이고, 접두만으로는 안 닫힌다.""" @pytest.mark.parametrize("fid,want", [ ("LP01", "location_part"), ("L01", "location"), ("C07", "character"), ("P0", "prop"), ("O99", "outlook"), ("Pfoo", None), # ★접두는 맞는데 뒤가 숫자가 아니다 ("LPfoo", None), ("L", None), ("", None), ("X9", None), ("P١٢", None), # ★`isdigit()` 은 아랍 숫자도 참이다 ]) def test_the_parser(self, fid, want): from app.modules.pipeline.grounding_entity_contract import ( owner_of_final_id) assert owner_of_final_id(fid) == want @pytest.mark.parametrize("owner,fid", [ ("location", "LP01"), ("location_part", "L01"), ("prop", "Pfoo"), ("character", "P01"), ("prop", "P"), ]) def test_a_mismatched_id_stops_at_the_endpoint(self, owner, fid): with pytest.raises(ad.LedgerContractViolation, match="의 것이다"): ad.to_entity_rows(_reduced( [_row("c0#1", owner, "무엇")], {"c0#1": _reg(final_id=fid)})) class TestQuietFoldsAtTheInputBoundary: """★★조용히 접으면 **등록된 것이 미등록**이 되고, **빈 신원끼리 같아 보인다**.""" @pytest.mark.parametrize("bad", ["true", 1, "", 0, [], "True"]) def test_registered_must_be_exactly_true_false_or_none(self, bad): with pytest.raises(ad.LedgerContractViolation, match="True/False/None"): ad.to_entity_rows(_reduced( [_row("c0#1", "prop", "가")], {"c0#1": {"registered": bad, "reason": "x", "final_id": "P01"}})) @pytest.mark.parametrize("ok", [True, False, None]) def test_the_three_real_values_pass(self, ok): """★막는 조건을 넓혔으면 **걸릴 정상 데이터**를 대 본다.""" ad.to_entity_rows(_reduced( [_row("c0#1", "prop", "가")], {"c0#1": {"registered": ok, "reason": "x", "final_id": "P01" if ok is True else None}})) @pytest.mark.parametrize("blank", ["", " "]) def test_a_blank_local_id_stops(self, blank): with pytest.raises(ad.LedgerContractViolation, match="빈 신원|빈 key"): ad.to_entity_rows(_reduced( [_row(blank, "prop", "가")], {blank: _reg(final_id="P01")}))