"""GROUNDING-V2 §2-3.5 — 후보 승격 overlay + 완전성 gate. ★A0 를 sidecar 로만 두면 **결속할 엔티티가 없다.** `list_entities_from_shots` 는 shot description 만 읽어서 샷 구조에 안 들어온 대상은 **볼 기회조차 없다.** """ from unittest.mock import patch import pytest from app.modules.pipeline.grounding_overlay import ( build_overlay_lines, candidates_for, completeness_report, missing_candidates, ) _SHOTS = [{"scene_index": 1, "scene_heading": "S#1", "shots": [{"shot_index": 1, "description": "버스 안", "characters": ["정임"]}]}] #: ★한 시험 안에서 후보마다 **다른 id**. 프로덕션 A0 는 `build_subject` 로 #: 언제나 id 를 붙이므로, id 없는 fixture 는 실제와 다른 모양이었다. _SEQ = [0] def _cand(surface, owner="prop", **over): _SEQ[0] += 1 base = {"research_subject_id": f"rs_{_SEQ[0]:03d}", "surface_form": surface, "owner_type": owner, "source_anchor": "S1", "source_quote": "…", "why_candidate": "시대 규격품", "planned_occurrences": 1} base.update(over) return base class TestLegacyIsByteIdentical: """★후보를 안 주면 **한 글자도** 안 달라져야 한다.""" def _prompt(self, **kw): seen = {} def _call(**kwargs): seen["user"] = kwargs["user_prompt"] return {"props": []} import app.modules.pipeline.entity_lister as el with patch.object(el, "call_structured", _call): el.list_entities_from_shots(_SHOTS, "prop", visual_rules="규칙", **kw) return seen["user"] def test_no_candidates_is_identical_to_not_passing_the_arg(self): assert self._prompt() == self._prompt(a0_candidates=None) assert self._prompt() == self._prompt(a0_candidates=[]) def test_candidates_of_another_type_do_not_change_the_prompt(self): """★prop 호출에 character 후보만 주면 그 호출은 그대로여야 한다.""" assert self._prompt() == self._prompt( a0_candidates=[_cand("감색 차장 제복", owner="character")]) def test_candidates_of_this_type_do_change_it(self): """positive control — 붙일 때는 실제로 붙는다.""" with_c = self._prompt(a0_candidates=[_cand("회수권 뭉치")]) assert with_c != self._prompt() assert "회수권 뭉치" in with_c def test_the_overlay_tells_it_not_to_re_filter(self): """★후보만 나열하면 모델이 기존 제외 규칙으로 **다시 지운다.**""" text = self._prompt(a0_candidates=[_cand("회수권 뭉치")]) assert "빼지 마세요" in text def test_the_exception_comes_after_the_exclusion_rules(self): """★조립 순서가 뜻을 뒤집는다 — 「빼지 마라 … 빼라」면 뒤가 이긴다. `type_prompt` 안의 제외 기준 뒤에 overlay 가 와야 한다. 앞에 두면 프롬프트가 자기모순이고 나중에 오는 목록이 이긴다 (Codex 지적). 여기서 보는 것은 **내가 넣은 표식의 위치**이지 문안의 뜻이 아니다. """ from app.modules.prompt_loader import load_prompt text = self._prompt(a0_candidates=[_cand("회수권 뭉치")]) type_prompt = load_prompt("entity_all", "prop") # 제외 기준은 type_prompt 의 **마지막 줄**에 가깝다 — 그 줄을 좌표로 쓴다. tail = [ln for ln in type_prompt.strip().split("\n") if ln.strip()][-1] assert tail in text, "type_prompt 끝줄을 못 찾았다 — 좌표가 틀렸다" assert text.index(tail) < text.index("이 목록이 우선합니다"), \ "overlay 가 제외 기준 **앞**에 있다 — 뒤에 오는 제외가 이긴다" class TestOwnerIsNotForced: """★착용물을 prop 으로 억지 통과시키지 않는다 (계약 §6).""" def test_outlook_goes_to_no_entity_type(self): c = [_cand("한복", owner="outlook")] for etype in ("prop", "character", "location"): assert candidates_for(c, etype) == [] assert build_overlay_lines(c, etype) == [] def test_outlook_is_deferred_not_lost(self): """★「없다」와 「여기서 못 받는다」를 갈라 센다.""" r = completeness_report([_cand("한복", owner="outlook")], {"prop": [], "character": [], "location": []}) assert r["complete"] is True, "받을 갈래가 없는 것을 실패로 셌다" assert len(r["deferred"]) == 1 and r["checked_count"] == 0 @pytest.mark.parametrize("owner,etype", [ ("prop", "prop"), ("character", "character"), ("location", "location")]) def test_each_owner_goes_to_its_own_type(self, owner, etype): assert len(candidates_for([_cand("x", owner=owner)], etype)) == 1 class TestCompletenessCatchesDrops: """★넣기만 하고 끝나면 뒤 단계가 또 거른다.""" def test_a_survivor_with_a_shortened_name_still_counts(self): """「고무줄로 묶인 회수권 뭉치」가 「회수권 뭉치」로 줄어도 잡는다.""" gone = missing_candidates([_cand("고무줄로 묶인 회수권 뭉치")], [{"name": "회수권 뭉치"}], "prop") assert gone == [] def test_a_dropped_candidate_is_reported(self): gone = missing_candidates([_cand("회수권 뭉치")], [{"name": "가방"}], "prop") assert len(gone) == 1 and gone[0]["surface_form"] == "회수권 뭉치" def test_report_is_not_complete_when_something_is_gone(self): r = completeness_report([_cand("회수권 뭉치")], {"prop": [{"name": "가방"}]}) assert r["complete"] is False and r["missing_count"] == 1 def test_report_is_complete_when_all_survive(self): r = completeness_report([_cand("회수권 뭉치")], {"prop": [{"name": "회수권 뭉치"}]}) assert r["complete"] is True and r["missing_count"] == 0 def test_empty_surface_form_is_skipped_not_counted_missing(self): assert missing_candidates([_cand("")], [{"name": "가방"}], "prop") == [] def test_normalization_is_letters_only(self): """공백·대소문자만 접는다. ★뜻으로 묶지 않는다.""" assert missing_candidates([_cand("Paper Ticket")], [{"name": "paper ticket"}], "prop") == [] assert len(missing_candidates([_cand("승차권")], [{"name": "표"}], "prop")) == 1 class TestProtectedUnion: """★새 필터 기구를 만들지 않는다 — 기존 `protected_short_ids` 통로에 union 한다. 계획 §1.8. """ def _ent(self): return { "props": [{"short_id": "P01", "name": "회수권 뭉치"}, {"short_id": "P02", "name": "가방"}], "characters": [{"short_id": "C01", "name": "정임"}], "locations": [], } def _d(self, surface, route): return {"_surface_form": surface, "route": route} def test_research_is_protected(self): from app.modules.pipeline.grounding_overlay import protected_short_ids assert protected_short_ids( [self._d("회수권 뭉치", "research")], self._ent()) == {"P01"} def test_design_is_protected(self): """★`fictional` → design 도 하류가 덮지 못하게 보호한다 (계약 §13).""" from app.modules.pipeline.grounding_overlay import protected_short_ids assert protected_short_ids( [self._d("정임", "design")], self._ent()) == {"C01"} def test_unresolved_is_protected(self): """★「모른다」를 「빼도 된다」로 읽으면 계약 §3 을 어긴다.""" from app.modules.pipeline.grounding_overlay import protected_short_ids assert protected_short_ids( [self._d("회수권 뭉치", "unresolved")], self._ent()) == {"P01"} def test_skip_is_not_protected(self): """★skip 까지 보호하면 저빈도 필터가 아무것도 못 지운다 — 필터를 없앤 것과 같다.""" from app.modules.pipeline.grounding_overlay import protected_short_ids assert protected_short_ids([self._d("가방", "skip")], self._ent()) == set() def test_nothing_decided_protects_nothing(self): from app.modules.pipeline.grounding_overlay import protected_short_ids assert protected_short_ids([], self._ent()) == set() def test_entities_without_short_id_are_skipped(self): from app.modules.pipeline.grounding_overlay import protected_short_ids ents = {"props": [{"short_id": "", "name": "회수권 뭉치"}]} assert protected_short_ids([self._d("회수권 뭉치", "research")], ents) == set() def test_a_shortened_survivor_name_still_matches(self): from app.modules.pipeline.grounding_overlay import protected_short_ids assert protected_short_ids( [self._d("고무줄로 묶인 회수권 뭉치", "research")], self._ent()) == {"P01"} class TestFilterStepUnionIsLegacySafe: """★체크포인트가 없으면 **legacy 가 그대로 돈다.** ★소스 문자열 검사를 걷어냈다 — 「그 줄이 있다」는 **그 줄이 도는 것**과 다르다. 여기서는 `EntityFilterStep._execute` 를 태워 `filter_low_frequency_entities` 에 **실제로 넘어간 보호집합**을 본다. """ _ENTS = {"props": [{"short_id": "P01", "name": "고무줄로 묶인 회수권 뭉치"}, {"short_id": "P02", "name": "빈 병"}], "characters": [], "locations": []} @staticmethod def _plan_cp(mode, decided): """★`run()` 이 저장하는 모양 그대로 — `{"status": ..., **result}`.""" return {"status": "completed", "data": {"mode": mode, "decided": decided, "counts": {}, "subject_count": len(decided), "search_calls": 0}} def _step(self, *, mode, plan_cp=None, relations=None): from app.core.steps.entity_steps import EntityFilterStep s = EntityFilterStep.__new__(EntityFilterStep) s.project_config = {"grounding_mode": mode} if mode else {} s.project_id, s.episode_id = "p", "e" s.build_opik_metadata = lambda: {} s._load_cleaned_text = lambda: "본문" cps = {"entity_merge": {"data": self._ENTS}, "grounding_plan": plan_cp} if relations: cps["entity_relation"] = {"data": {"relations": relations}} s._load_prev_checkpoint = lambda sid: cps.get(sid) return s @staticmethod def _protected_passed_to_filter(step): from unittest.mock import patch with patch("app.modules.pipeline.entity_filter" ".filter_low_frequency_entities") as f: f.return_value = {"filtered": {}} step._execute() assert f.call_count == 1 return f.call_args.kwargs["protected_short_ids"] def test_v2_plan_actually_reaches_the_filter(self): """★`data` 로 안 싸면 여기가 빨개진다 — 실제로 그랬다.""" d = [{"route": "research", "_surface_form": "회수권 뭉치", "_short_id": "P01", "_owner_type": "prop"}] got = self._protected_passed_to_filter( self._step(mode="v2", plan_cp=self._plan_cp("v2", d))) assert got == {"P01"} def test_shadow_plan_does_not_reach_the_filter(self): """관측이 하류를 바꾸면 안 된다.""" cp = {"status": "completed", "data": {"mode": "shadow_plan", "decided": [], "shadow_decided": [{"route": "research", "_surface_form": "회수권 뭉치", "_short_id": "P01"}]}} got = self._protected_passed_to_filter( self._step(mode="shadow_plan", plan_cp=cp)) assert got is None def test_legacy_without_a_checkpoint_is_unchanged(self): got = self._protected_passed_to_filter(self._step(mode="legacy")) assert got is None def test_it_unions_rather_than_replacing(self): """★기존 변형 관계 보호를 덮어쓰면 안 된다.""" d = [{"route": "research", "_surface_form": "회수권 뭉치", "_short_id": "P01", "_owner_type": "prop"}] got = self._protected_passed_to_filter(self._step( mode="v2", plan_cp=self._plan_cp("v2", d), relations=[{"visual_similarity": True, "base_short_id": "P02", "variant_short_id": ""}])) assert got == {"P01", "P02"} def test_skip_route_is_not_protected(self): """★`skip` 까지 보호하면 저빈도 필터가 아무것도 못 지운다.""" d = [{"route": "skip", "_surface_form": "빈 병", "_short_id": "P02", "_owner_type": "prop"}] got = self._protected_passed_to_filter( self._step(mode="v2", plan_cp=self._plan_cp("v2", d))) assert got is None