"""union coverage (#3), 참조 특징 캐시 (#5), 점수 의미 (#9), 법적 맥락 (#10). scipy/sklearn 이 없으면 인덱스 관련 테스트는 skip 된다. """ from __future__ import annotations import pytest from app.engine.legal_risk import LegalRiskEngine, Precedent from app.engine.persistent_index import _covered_length, _merge_intervals from app.engine.provenance import CorpusStore, DocumentRecord, SegmentRecord scipy = pytest.importorskip("scipy", reason="scipy 미설치") from app.engine.persistent_index import PersistentCorpusIndex # noqa: E402 def _segment(segment_id: str, text: str, **kw) -> SegmentRecord: return SegmentRecord( segment_id=segment_id, document_id=kw.pop("document_id", "doc-1"), text=text, ordinal=segment_id, char_start=0, char_end=len(text), source_locator=f"book.json#{segment_id}", **kw, ) def _index_with(tmp_path, segments: list[SegmentRecord], title="원본"): db = tmp_path / "corpus.sqlite3" store = CorpusStore(db) for doc_id in {s.document_id for s in segments}: store.upsert_document(DocumentRecord(document_id=doc_id, title=title)) store.add_segments(segments) index = PersistentCorpusIndex(db, tmp_path / "index") index.sync() return store, index # --------------------------------------------------------------------------- # 구간 병합 (순수 함수) # --------------------------------------------------------------------------- def test_merge_intervals_deduplicates_overlap(): assert _merge_intervals([(0, 10), (5, 20), (30, 40)]) == [(0, 20), (30, 40)] assert _covered_length([(0, 10), (5, 20)]) == 20 assert _covered_length([(0, 10), (0, 10)]) == 10, "중복 구간을 두 번 세면 안 된다" assert _covered_length([]) == 0 def test_merge_intervals_handles_adjacent_and_nested(): assert _merge_intervals([(0, 10), (10, 20)]) == [(0, 20)] assert _merge_intervals([(0, 100), (10, 20)]) == [(0, 100)] # --------------------------------------------------------------------------- # #3 union coverage # --------------------------------------------------------------------------- def test_union_coverage_sums_multiple_segments(tmp_path): """서로 다른 세그먼트가 질의의 다른 부분과 일치하면 coverage 가 합산된다.""" part_a = "바닷가 마을에서 파도 소리를 들으며 자란 기억이 아직도 선명하게 남아 있다. " * 3 part_b = "군에 입대하던 날 아버지는 아무 말 없이 내 어깨를 두드려 주셨던 기억이 난다. " * 3 _, index = _index_with(tmp_path, [ _segment("seg-a", part_a), _segment("seg-b", part_b), ]) query = part_a + "완전히 무관한 중간 문단입니다. " * 20 + part_b result = index.search(query, top_k=10) assert result.union_coverage > 0.4, "두 구간이 모두 반영되어야 한다" assert result.covered_chars >= len(part_a) assert result.query_chars == len(query) # 개별 세그먼트 coverage 는 각자 union 보다 작다 per_hit = [h.coverage for h in result.hits if h.reranked] assert max(per_hit) < result.union_coverage def test_document_coverage_does_not_leak_between_sources(tmp_path): part_a = "바닷가 마을에서 파도 소리를 들으며 자랐다. " * 3 part_b = "군에 입대하던 날 아버지가 내 어깨를 두드렸다. " * 3 _, index = _index_with(tmp_path, [ _segment("seg-a", part_a, document_id="doc-a"), _segment("seg-b", part_b, document_id="doc-b"), ]) query = part_a + ("서로 무관한 중간 문장입니다. " * 20) + part_b result = index.search(query, top_k=10) assert result.union_coverage > result.document_coverage["doc-a"] assert result.union_coverage > result.document_coverage["doc-b"] assert result.document_coverage["doc-a"] < 0.30 assert result.document_coverage["doc-b"] < 0.30 def test_union_coverage_low_for_long_unrelated_document(tmp_path): _, index = _index_with(tmp_path, [_segment("seg-1", "바닷가 마을의 파도 소리를 기억한다.")]) query = "전혀 다른 주제의 글입니다. 오늘 회의에서 분기 실적을 논의했습니다. " * 100 result = index.search(query, top_k=5) assert result.union_coverage < 0.1 def test_short_full_copy_reaches_high_coverage(tmp_path): text = "나는 어린 시절 바닷가 마을에서 살았고 매일 파도 소리를 들으며 잠들었다." _, index = _index_with(tmp_path, [_segment("seg-1", text)]) result = index.search(text, top_k=5) assert result.union_coverage > 0.8 def test_long_document_partial_copy_is_not_structurally_zero(tmp_path): """긴 원고 안의 부분 복사가 coverage 에 실제로 잡히는지 (#3 회귀).""" copied = "바닷가 마을에서 파도 소리를 들으며 자란 기억이 선명하다. " * 10 _, index = _index_with(tmp_path, [_segment("seg-1", copied)]) filler = "무관한 문장입니다. " * 300 query = filler + copied + filler result = index.search(query, top_k=5) assert result.covered_chars >= len(copied) * 0.5 assert result.union_coverage > 0.0 # --------------------------------------------------------------------------- # CPU 상한 # --------------------------------------------------------------------------- def test_evidence_limit_caps_precise_comparison(tmp_path): segments = [ _segment(f"seg-{i}", f"바닷가 마을 이야기 {i}번 문단입니다. 파도 소리를 들었다. " * 3) for i in range(8) ] _, index = _index_with(tmp_path, segments) result = index.search("바닷가 마을 이야기 파도 소리를 들었다.", top_k=8, evidence_limit=3) reranked = [h for h in result.hits if h.reranked] assert len(reranked) == 3 assert result.evidence_truncated is True for hit in result.hits: if not hit.reranked: assert hit.evidence == [] and hit.coverage == 0.0 and hit.longest_span == 0 def test_query_wrapper_still_returns_hits(tmp_path): """기존 호출부 호환 — query() 는 여전히 list[PersistentHit].""" _, index = _index_with(tmp_path, [_segment("seg-1", "바닷가 마을의 파도 소리.")]) hits = index.query("바닷가 마을의 파도 소리.", top_k=1) assert isinstance(hits, list) and hits[0].segment_id == "seg-1" # --------------------------------------------------------------------------- # #5 참조 특징 캐시 + 마이그레이션 # --------------------------------------------------------------------------- def test_v1_database_migrates_without_data_loss(tmp_path): """lemmas_json/elements_json 없는 기존 DB 도 그대로 열려야 한다.""" import sqlite3 db = tmp_path / "old.sqlite3" con = sqlite3.connect(db) con.executescript( """ CREATE TABLE corpus_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL); CREATE TABLE documents ( document_id TEXT PRIMARY KEY, title TEXT NOT NULL, source_path TEXT, source_sha256 TEXT, metadata_json TEXT NOT NULL DEFAULT '{}', created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP); CREATE TABLE segments ( segment_id TEXT PRIMARY KEY, document_id TEXT NOT NULL, ordinal TEXT NOT NULL, text TEXT NOT NULL, text_sha256 TEXT NOT NULL, coordinate_scope TEXT NOT NULL, page_number INTEGER, paragraph_number INTEGER, char_start INTEGER, char_end INTEGER, source_locator TEXT, metadata_json TEXT NOT NULL DEFAULT '{}', created_at TEXT DEFAULT CURRENT_TIMESTAMP, UNIQUE(document_id, text_sha256)); INSERT INTO documents(document_id,title) VALUES('doc-1','옛 책'); INSERT INTO segments VALUES('seg-1','doc-1','1','옛 본문입니다.','h','episode', NULL,NULL,0,7,NULL,'{}',CURRENT_TIMESTAMP); """ ) con.commit() con.close() store = CorpusStore(db) store.initialize() # 마이그레이션 segments = list(store.iter_segments()) assert len(segments) == 1 assert segments[0].text == "옛 본문입니다." assert segments[0].lemmas is None assert store.count_missing_features() == 1 def test_feature_roundtrip_and_backfill(tmp_path): store = CorpusStore(tmp_path / "corpus.sqlite3") store.upsert_document(DocumentRecord(document_id="doc-1", title="책")) store.add_segments([_segment("seg-1", "홍길동은 활빈당을 만들었다.")]) assert store.count_missing_features() == 1 store.update_segment_features([("seg-1", ["홍길동", "활빈당"], {"characters": ["홍길동"]})]) assert store.count_missing_features() == 0 loaded = store.get_segments(["seg-1"])["seg-1"] assert loaded.lemmas == ["홍길동", "활빈당"] assert loaded.elements == {"characters": ["홍길동"]} def test_precomputed_features_reach_the_hit(tmp_path): store = CorpusStore(tmp_path / "corpus.sqlite3") store.upsert_document(DocumentRecord(document_id="doc-1", title="책")) store.add_segments([_segment( "seg-1", "나는 어린 시절 바닷가 마을에서 살았다.", lemmas=["바닷가", "마을", "살다"], elements={"characters": [], "motifs": [], "genre": None, "keywords": ["바닷가"]}, )]) index = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index") index.sync() hit = index.query("나는 어린 시절 바닷가 마을에서 살았다.", top_k=1)[0] assert hit.reference_lemmas == ["바닷가", "마을", "살다"] assert hit.reference_elements["keywords"] == ["바닷가"] def test_detector_uses_cache_instead_of_recomputing(monkeypatch, tmp_path): """캐시가 있으면 형태소 분석을 호출하지 않아야 한다 (#5 호출횟수 테스트).""" from app.api.schemas import ExtractedElements from app.engine import detector as det_module from app.engine.persistent_index import PersistentHit calls = {"lemmas": 0, "extract": 0} def counting_lemmas(text, *a, **kw): calls["lemmas"] += 1 return ["x"] monkeypatch.setattr(det_module, "extract_lemmas", counting_lemmas) detector = det_module.PlagiarismDetector.__new__(det_module.PlagiarismDetector) detector._feature_cache = {} detector._persistent = None class _Extractor: def extract(self, text): calls["extract"] += 1 return ExtractedElements() detector._extractor = _Extractor() cached_hit = PersistentHit( segment_id="seg-1", document_id="doc-1", title="책", score=0.9, evidence=[], coverage=0.0, longest_span=0, source_locator=None, coordinate_scope="episode", page_number=None, paragraph_number=None, source_char_start=None, source_char_end=None, reference_text="본문", reference_lemmas=["미리", "계산"], reference_elements={"characters": [], "motifs": [], "genre": None, "keywords": []}, ) lemmas, _ = detector.reference_features(cached_hit) assert lemmas == ["미리", "계산"] assert calls == {"lemmas": 0, "extract": 0}, "캐시가 있는데 재계산했다" uncached = PersistentHit( segment_id="seg-2", document_id="doc-1", title="책", score=0.9, evidence=[], coverage=0.0, longest_span=0, source_locator=None, coordinate_scope="episode", page_number=None, paragraph_number=None, source_char_start=None, source_char_end=None, reference_text="본문", ) detector.reference_features(uncached) assert calls == {"lemmas": 1, "extract": 1} # 두 번째 호출은 프로세스 캐시로 처리 detector.reference_features(uncached) assert calls == {"lemmas": 1, "extract": 1} # --------------------------------------------------------------------------- # #10 법적 맥락 + 판례 랭킹 # --------------------------------------------------------------------------- def _precedents() -> list[Precedent]: return [ Precedent("2020다1", "가", "https://x/1", ("literary",), ("reproduction",), (), "요지1"), Precedent("2019다2", "나", "https://x/2", ("literary",), ("reproduction", "derivative_work"), (), "요지2"), Precedent("2018다3", "다", "https://x/3", ("musical",), ("reproduction",), (), "요지3"), ] def test_precedent_ranking_prefers_more_tag_overlap(): engine = LegalRiskEngine(_precedents()) out = engine.assess( max_similarity=0.9, coverage=0.5, longest_span=200, legal_tags=["reproduction", "derivative_work"], work_type="literary", ) assert out.precedent_ids[0] == "2019다2", "태그 교집합이 큰 판례가 먼저" assert "2018다3" not in out.precedent_ids, "work_type 이 다른 판례는 제외" def test_only_registered_precedents_are_returned(): engine = LegalRiskEngine(_precedents()) out = engine.assess( max_similarity=0.9, coverage=0.5, longest_span=200, legal_tags=["reproduction"], work_type="literary", ) registered = {p.case_id for p in _precedents()} assert set(out.precedent_ids) <= registered def test_empty_precedent_db_reports_insufficient(): out = LegalRiskEngine([]).assess( max_similarity=0.9, coverage=0.9, longest_span=500, legal_tags=["reproduction"], ) assert out.status == "insufficient_precedent_data" assert out.risk_level is None assert out.precedent_ids == () def test_legal_context_clears_missing_factors(): engine = LegalRiskEngine(_precedents()) default = engine.assess(max_similarity=0.5, coverage=0.1, longest_span=20, legal_tags=["reproduction"]) assert len(default.missing_factors) == 3 assert default.access_evidence == "not_provided" supplied = engine.assess( max_similarity=0.5, coverage=0.1, longest_span=20, legal_tags=["reproduction"], access_evidence=True, protected_expression_reviewed=True, rights_verified=True, ) assert supplied.missing_factors == () assert supplied.access_evidence == "provided" assert supplied.protected_expression == "reviewed" def test_legal_context_flows_through_detect_request(): from app.api.schemas import DetectRequest, LegalContext req = DetectRequest( doc_id="d", text="본문", legal_context=LegalContext(work_type="musical", access_evidence=False, protected_expression_reviewed=True), ) assert req.legal_context.work_type == "musical" assert req.legal_context.access_evidence is False assert req.legal_context.rights_verified is False # 미제공이 기본 assert DetectRequest(doc_id="d", text="본문").legal_context is None