# pyright: reportMissingImports=false """IncrementalUpdater 테스트 — TDD Phase 3""" from __future__ import annotations import pytest from kakao_knowledge.vector_store import VectorStore from kakao_knowledge.knowledge_graph import KnowledgeGraph from kakao_knowledge.incremental_updater import IncrementalUpdater # --------------------------------------------------------------------------- # Shared sample data # --------------------------------------------------------------------------- @pytest.fixture def sample_insights() -> list[dict]: """기존 DB에 들어갈 InsightV2 형식 인사이트 3개.""" return [ { "id": "insight-001", "title": "광응고술 보상 기준", "type": "qa", "category": "보상/일반", "summary": "광응고술은 실손보험에서 수술비로 인정된다.", "key_points": ["광응고술은 수술비 대상", "입원 없이도 수술비 청구 가능"], "expert": "김전문", "confidence": "low", "related_topics": ["광응고술", "실손보험"], "tags": ["#보상", "#실손"], "source_date": "2025-12-01", "source_chat": "보험전문가방", "participants": ["김전문", "이설계"], }, { "id": "insight-002", "title": "고지의무 위반 시 계약 해지", "type": "regulation_interpretation", "category": "고지의무", "summary": "고지의무 위반이 있더라도 보험사가 2년 내 발견하지 못하면 해지 불가.", "key_points": ["2년 경과 후 해지 불가", "고의 위반은 예외"], "expert": "박약관", "confidence": "high", "related_topics": ["고지의무", "계약해지"], "tags": ["#고지의무", "#약관"], "source_date": "2025-12-02", "source_chat": "보험전문가방", "participants": ["박약관", "최신입"], }, { "id": "insight-003", "title": "자동차보험 렌트비 청구 방법", "type": "practical_tip", "category": "보상/자동차", "summary": "사고 후 렌트비는 수리 기간 동안 인정된다.", "key_points": ["수리 기간 렌트비 인정", "사전 보험사 승인 필수"], "expert": "홍자동차", "confidence": "medium", "related_topics": ["자동차보험", "렌트비"], "tags": ["#자동차", "#렌트"], "source_date": "2025-12-03", "source_chat": "보험전문가방", "participants": ["홍자동차", "정설계"], }, ] @pytest.fixture def setup(tmp_path, sample_insights): """VectorStore + KnowledgeGraph + IncrementalUpdater 설정.""" db_path = str(tmp_path / "chroma_db") insights_dir = str(tmp_path / "insights") graph_path = str(tmp_path / "graph.json") index_path = str(tmp_path / "index.json") store = VectorStore(db_path=db_path, collection_name="test_insights") store.add_insights(sample_insights) graph = KnowledgeGraph( insights_dir=insights_dir, graph_path=graph_path, index_path=index_path, ) graph.build_from_insights(sample_insights) updater = IncrementalUpdater(vector_store=store, graph=graph) return { "store": store, "graph": graph, "updater": updater, "existing_insights": sample_insights, "insights_dir": insights_dir, "graph_path": graph_path, "index_path": index_path, } # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- class TestIncrementalUpdater: def test_empty_new_insights(self, setup) -> None: """빈 new_insights 리스트 처리 시 올바른 stats를 반환해야 한다.""" updater: IncrementalUpdater = setup["updater"] stats = updater.update([]) assert stats["new"] == 0 assert stats["augmented"] == 0 assert stats["total_processed"] == 0 def test_new_insight_added(self, setup) -> None: """완전히 새로운 인사이트는 'new'로 처리되어야 한다.""" updater: IncrementalUpdater = setup["updater"] store: VectorStore = setup["store"] count_before = store.count() # 기존 데이터와 완전히 다른 주제의 인사이트 new_insights = [ { "id": "insight-010", "title": "세금 절세 전략 연금보험", "type": "expert_opinion", "category": "세금/절세", "summary": "연금보험을 통한 절세 효과와 연말정산 세액공제 방법.", "key_points": ["연금보험 세액공제 가능", "납입 한도 내 절세"], "expert": "세금전문가", "confidence": "high", "related_topics": ["세금", "절세", "연금"], "tags": ["#세금", "#절세"], "source_date": "2025-12-10", "source_chat": "보험전문가방", "participants": ["세금전문가"], } ] stats = updater.update(new_insights, similarity_threshold=0.85) assert stats["new"] == 1 assert stats["augmented"] == 0 assert stats["total_processed"] == 1 assert "insight-010" in stats["new_ids"] # 스토어에 추가되어야 한다 assert store.count() == count_before + 1 def test_augment_existing(self, setup) -> None: """유사한 인사이트(같은 카테고리+타입)는 'augmented'로 처리되어야 한다.""" updater: IncrementalUpdater = setup["updater"] # insight-001(qa, 보상/일반, 광응고술)과 유사한 인사이트 similar_insights = [ { "id": "insight-020", "title": "광응고술 실손보험 청구 안내", "type": "qa", "category": "보상/일반", "summary": "광응고술은 실손보험 수술비 항목으로 청구 가능하며 입원 기준이 적용된다.", "key_points": ["수술비로 청구 가능", "병원 영수증 필요"], "expert": "이전문", "confidence": "medium", "related_topics": ["광응고술", "실손보험", "수술비"], "tags": ["#실손", "#수술"], "source_date": "2025-12-10", "source_chat": "보험전문가방", "participants": ["이전문", "김신입"], } ] stats = updater.update(similar_insights, similarity_threshold=0.85) assert stats["augmented"] == 1 assert stats["total_processed"] == 1 assert len(stats["augmented_ids"]) == 1 def test_augment_merges_key_points(self, setup) -> None: """보강 시 key_points가 병합되어야 한다.""" updater: IncrementalUpdater = setup["updater"] store: VectorStore = setup["store"] # insight-001과 유사, 새로운 key_point 추가 similar_insights = [ { "id": "insight-021", "title": "광응고술 보험금 청구 절차", "type": "qa", "category": "보상/일반", "summary": "광응고술 수술비 청구 시 필요한 서류와 절차.", "key_points": ["의사 소견서 필수", "청구 기한 3년"], "expert": "박전문", "confidence": "high", "related_topics": ["광응고술", "실손보험", "청구서류"], "tags": ["#실손", "#청구"], "source_date": "2025-12-11", "source_chat": "보험전문가방", "participants": ["박전문"], } ] stats = updater.update(similar_insights, similarity_threshold=0.85) # 보강이 되었다면 augmented_ids에 insight-001이 있어야 함 if stats["augmented"] == 1: augmented_id = stats["augmented_ids"][0] result = store.get_by_id(augmented_id) # 메타데이터에서 확인 (VectorStore는 메타데이터만 저장) assert result is not None def test_update_returns_stats(self, setup) -> None: """update()는 올바른 통계 dict를 반환해야 한다.""" updater: IncrementalUpdater = setup["updater"] new_insights = [ { "id": "insight-030", "title": "손해사정사 역할과 활용법", "type": "practical_tip", "category": "손해사정", "summary": "손해사정사는 보험금 청구 분쟁 시 전문가 도움을 제공한다.", "key_points": ["무료 상담 가능", "복잡한 사고 필수"], "expert": "손해사정가", "confidence": "medium", "related_topics": ["손해사정", "보험분쟁"], "tags": ["#손해사정"], "source_date": "2025-12-12", "source_chat": "보험전문가방", "participants": ["손해사정가"], } ] stats = updater.update(new_insights) # 필수 키 확인 assert "new" in stats assert "augmented" in stats assert "total_processed" in stats assert "new_ids" in stats assert "augmented_ids" in stats assert isinstance(stats["new_ids"], list) assert isinstance(stats["augmented_ids"], list) def test_confidence_upgrade_on_augment(self, setup) -> None: """보강 시 confidence가 상향되어야 한다 (low → medium → high).""" updater: IncrementalUpdater = setup["updater"] # insight-001의 confidence는 "low" # 보강 후 "medium"이 되어야 함 existing = setup["existing_insights"][0].copy() assert existing["confidence"] == "low" new_insight = { "id": "augment-test", "title": "광응고술 추가 정보", "type": "qa", "category": "보상/일반", "summary": "광응고술 실손 청구 추가 정보 및 주의 사항.", "key_points": ["추가 서류 필요"], "expert": "추가전문가", "confidence": "medium", "related_topics": ["광응고술", "실손보험"], "tags": ["#실손"], "source_date": "2025-12-13", "source_chat": "보험전문가방", "participants": ["추가전문가"], } # _augment_insight 직접 테스트 augmented = updater._augment_insight(existing, new_insight) assert augmented["confidence"] in ("medium", "high") # low → medium 이상 def test_augment_merges_participants(self, setup) -> None: """보강 시 participants가 합집합으로 병합되어야 한다.""" updater: IncrementalUpdater = setup["updater"] existing = setup["existing_insights"][0].copy() # 기존: ["김전문", "이설계"] new_insight = { "id": "participant-test", "title": "광응고술 추가 논의", "type": "qa", "category": "보상/일반", "summary": "광응고술 관련 추가 전문가 의견.", "key_points": ["추가 포인트"], "expert": "신규전문가", "confidence": "medium", "related_topics": ["광응고술"], "tags": ["#실손"], "source_date": "2025-12-14", "source_chat": "보험전문가방", "participants": ["신규전문가", "이설계"], # 이설계는 중복 } augmented = updater._augment_insight(existing, new_insight) participants = set(augmented["participants"]) assert "김전문" in participants assert "이설계" in participants assert "신규전문가" in participants