"""`location_part` 를 받는 sync 계약 — **순수 helper**. ★inert. 유료 0. Codex 2026-08-31: 「기존 query 목록을 지금 무조건 넓히면 **inert 가 아니다**」. 그래서 **표식이 있을 때만** 새 길로 가고 legacy CP 는 옛 길 그대로다. """ from __future__ import annotations import pytest from app.modules.pipeline import grounding_entity_sync_ext as x from app.modules.pipeline.grounding_entity_contract import OWNER_PREFIX LEGACY = ["characters", "locations", "props"] class TestLegacyCheckpointsAreUntouched: """★표식이 없으면 **옛 셋 그대로** — `location_parts` 를 아예 안 본다.""" @pytest.mark.parametrize("cp", [{}, None, "글자", 7, {"other": 1}]) def test_no_marker_means_the_old_three(self, cp): """★표식이 **없으면** 옛 CP 다 — 옛 길이 맞다.""" assert x.owner_keys(cp) == LEGACY assert x.is_chunk_marked(cp) is False @pytest.mark.parametrize("ver", ["", "0", "2", "abc", "1.1"]) def test_a_present_but_unknown_version_stops(self, ver): """★★★**뒤집은 시험** — 앞 판은 모르는 판도 legacy 로 떨어뜨렸다. 그러면 새 CP 인데 `location_part` 가 **조용히 버려진다** — 오류도 안 나고 행만 사라진다 (Codex 재현 2026-08-31). 내가 그 동작을 **의도로 시험에 적기까지 했다**. 가르는 자리는 「표식이 **있느냐**」다. 없으면 옛 CP, 있는데 모르면 **모르는 것**이다 — 모르는 것을 옛 것으로 읽지 않는다. """ cp = {x.CHUNK_SCHEMA_MARKER: ver} with pytest.raises(x.UnknownChunkSchema): x.is_chunk_marked(cp) with pytest.raises(x.UnknownChunkSchema): x.owner_keys(cp) def test_the_marker_key_alone_is_enough_to_stop(self): """★값이 `None` 이어도 **키가 있으면** 새 CP 를 뜻한다.""" with pytest.raises(x.UnknownChunkSchema): x.is_chunk_marked({x.CHUNK_SCHEMA_MARKER: None}) def test_the_marker_opens_the_new_key(self): cp = {x.CHUNK_SCHEMA_MARKER: x.CHUNK_SCHEMA_VERSION} assert x.is_chunk_marked(cp) is True assert x.owner_keys(cp) == LEGACY + ["location_parts"] def test_outlook_is_never_in_this_path(self): """★`OutlookSyncService` 관할이다 — 여기 넣으면 중복이 난다.""" assert "outlook" not in x.CHUNK_OWNER_KEYS cp = {x.CHUNK_SCHEMA_MARKER: x.CHUNK_SCHEMA_VERSION} assert not any("outlook" in k for k in x.owner_keys(cp)) class TestTheCounterSeesEveryOwnerApart: """★★옛 셈은 `LP20` 을 **조용히 건너뛰어** `L` counter 를 작게 만들었다.""" def test_location_and_its_part_are_counted_apart(self): got = x.seed_counters(["L05", "LP20"]) assert got["L"] == 5 and got["LP"] == 20 def test_the_next_id_does_not_collide(self): """★★이것이 조용한 결함의 **끝점**이다 — 번호를 다시 주면 안 된다.""" c = x.seed_counters(["L05", "LP20"]) assert x.next_short_id(c, "location") == "L06" assert x.next_short_id(c, "location_part") == "LP21" def test_the_old_way_would_have_reissued(self): """★견줌 — 옛 셈은 `LP20` 을 못 봐서 `L` 이 5 에 머문다.""" def old(sid): try: return sid[0], int(sid[1:]) except (ValueError, IndexError): return None assert old("LP20") is None, "★옛 셈이 LP20 을 본다" assert old("L05") == ("L", 5) def test_junk_is_not_counted(self): got = x.seed_counters(["Pfoo", "", "L", "xyz"]) assert all(v == 0 for v in got.values()) def test_every_owner_has_a_slot(self): """★갈래를 늘려도 따라간다 — 목록을 손으로 안 적는다.""" got = x.seed_counters([]) assert set(got) == set(OWNER_PREFIX.values()) def test_an_unknown_owner_stops(self): with pytest.raises(KeyError): x.next_short_id({}, "지어낸갈래") class TestCanonLookupPrefersTheCanonicalId: """★이름 단독이면 **중복**과 **조용한 충돌**이 난다.""" class Row: def __init__(self, tag): self.tag = tag def test_the_short_id_wins(self): a, b = self.Row("a"), self.Row("b") got = x.resolve_canon(short_id="L01", entity_type="location", name="가게", by_short_id={"L01": a}, by_type_name={}) assert got is a def test_the_name_is_the_fallback(self): b = self.Row("b") got = x.resolve_canon(short_id="L99", entity_type="location", name="가게", by_short_id={}, by_type_name={("location", "가게"): b}) assert got is b def test_a_renamed_entity_is_still_found_by_id(self): """★★정본 ID 가 있는데 이름이 바뀌면 — 옛 것을 **찾아야** 한다.""" a = self.Row("a") got = x.resolve_canon(short_id="L01", entity_type="location", name="새 이름", by_short_id={"L01": a}, by_type_name={}) assert got is a, "★이름이 바뀌었다고 새로 만든다" def test_the_same_name_in_two_owners_does_not_collide(self): """★★`location` 과 `location_part` 의 **같은 이름**.""" loc, part = self.Row("loc"), self.Row("part") by_name = {("location", "이발소"): loc, ("location_part", "이발소"): part} assert x.resolve_canon(short_id="", entity_type="location", name="이발소", by_short_id={}, by_type_name=by_name) is loc assert x.resolve_canon(short_id="", entity_type="location_part", name="이발소", by_short_id={}, by_type_name=by_name) is part def test_two_keys_pointing_at_different_canons_is_the_id(self): """★★2026-09-03 뒤집음(Codex 계약 06:55 · 실측 f7cc45c576c0): short_id 가 **정본**이다. 같은 이름·같은 타입인데 short_id 가 다른 행은 **남의 행**이라 충돌이 아니다(「흙바닥」 LP05→L03 · LP07→L02). 옛 판은 두 열쇠가 다른 행을 가리키면 `SyncKeyConflict` 로 섰다 — 그러면 합법인 동명 행이 막힌다.""" a, b = self.Row("a"), self.Row("b") b.short_id = "L02" got = x.resolve_canon(short_id="L01", entity_type="location", name="가게", by_short_id={"L01": a}, by_type_name={("location", "가게"): b}) assert got is a, "★short_id 가 맞는 행이 있으면 이름은 보지 않는다" def test_the_same_canon_from_both_keys_is_fine(self): a = self.Row("a") got = x.resolve_canon(short_id="L01", entity_type="location", name="가게", by_short_id={"L01": a}, by_type_name={("location", "가게"): a}) assert got is a def test_nothing_found_is_none(self): assert x.resolve_canon(short_id="L01", entity_type="location", name="가게", by_short_id={}, by_type_name={}) is None class TestItIsStillInert: def test_the_live_sync_actually_consumes_it(self): """★★**뒤집힌 시험**이다 (§2-6.5, 2026-09-01). 앞에는 「활성 경로가 아직 이것을 안 부른다」를 잠그고 있었다 — 그것이 inert 라는 증거였다. 이제는 **부르는 것이 계약**이다. Codex 가 잡은 결함이 정확히 이것이었다: adapter 는 `location_parts` 를 내는데 `EntitySyncService` 가 한 줄도 안 읽었다. """ import ast from pathlib import Path as _P me = "grounding_entity_sync_ext" root = _P(__file__).resolve().parents[2] / "app" hits = [] for f in root.rglob("*.py"): if f.name == f"{me}.py": continue try: tree = ast.parse(f.read_text(encoding="utf-8")) except SyntaxError: # noqa: PERF203 continue for n in ast.walk(tree): if isinstance(n, ast.ImportFrom) and me in str(n.module or ""): hits.append(f.name) assert "entity_sync_service.py" in hits, ( "★EntitySync 가 이 계약을 안 쓴다 — `location_parts` 가 버려진다") def test_the_live_sync_asks_the_checkpoint_which_kinds_to_read(self): """★갈래 목록을 **손으로 안 적는다** — CP 표식이 정한다. ★★조회·loop·stale cleanup 이 **같은 한 벌**을 써야 한다. 조회만 넓히면 LP link 가 `_existing_links` 에 들어오고 `_seen_canon_ids` 에는 안 들어와 stale cleanup 이 **그것을 지운다** (Codex 데이터 손실 BLOCK). """ import inspect from app.services.checkpoint_sync import entity_sync_service as es src = inspect.getsource(es.EntitySyncService.sync_from_checkpoint) assert '["characters", "locations", "props"]' not in src, ( "★갈래 목록을 손으로 적었다") assert "_sync_keys(t2i_cp)" in src # ★읽는 자리와 지우는 자리가 **같은 변수**를 본다 assert src.count("_keys") >= 3 and "_types" in src