338 lines
14 KiB
Python
338 lines
14 KiB
Python
"""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
|