o2o-plagiarism-ai/tests/test_legal_risk.py

import json
from dataclasses import replace
from types import SimpleNamespace

from app.engine.legal_risk import (
    LegalJudgeDecision,
    LegalRiskEngine,
    Precedent,
    load_precedents,
)
from app.engine.openai_legal_judge import OpenAILegalJudge


def _case() -> Precedent:
    return Precedent(
        case_id="case-1",
        title="테스트 판례",
        source_url="https://example.test/case-1",
        work_types=("literary",),
        legal_tags=("reproduction",),
        criteria=("실질적 유사성",),
        holding_summary="테스트 요약",
    )


def test_empty_precedents_never_claim_legal_conclusion():
    result = LegalRiskEngine([]).assess(
        max_similarity=0.95, coverage=0.8, longest_span=300,
        legal_tags=["reproduction"],
    )
    assert result.status == "insufficient_precedent_data"
    assert result.risk_level is None


def test_risk_requires_missing_facts_and_only_registered_case_ids():
    result = LegalRiskEngine([_case()]).assess(
        max_similarity=0.9, coverage=0.5, longest_span=200,
        legal_tags=["reproduction"],
    )
    assert result.status == "review_required"
    assert result.risk_level == "high"
    assert result.precedent_ids == ("case-1",)
    assert len(result.missing_factors) == 3


def test_repository_seed_is_valid():
    cases = load_precedents("data/precedents/precedents.jsonl")
    assert cases
    assert any(case.case_id == "2012다73493" for case in cases)
    assert len(cases) == len({case.case_id for case in cases})


class _FakeJudge:
    def __init__(self, case_ids=("case-1",), review_required=True):
        self.case_ids = case_ids
        self.review_required = review_required
        self.request = None

    def judge(self, request):
        self.request = request
        return LegalJudgeDecision(
            verdict="likely",
            confidence=0.82,
            matched_precedent_ids=self.case_ids,
            supporting_reasons=("표현 일치 구간이 길다",),
            counter_reasons=("접근 가능성은 확인되지 않았다",),
            missing_factors=("시장 영향",),
            review_required=self.review_required,
            model="test-gpt",
            prompt_version="test-v1",
        )


def test_llm_judge_augments_rule_result_and_keeps_deterministic_missing_factors():
    judge = _FakeJudge()
    result = LegalRiskEngine([_case()], judge=judge).assess(
        max_similarity=0.9,
        coverage=0.5,
        longest_span=200,
        legal_tags=["reproduction"],
        evidence=[{"source_document_id": "doc-1", "excerpts": ["일치 문장"]}],
    )
    assert result.judgment_method == "llm"
    assert result.llm_verdict == "likely"
    assert result.llm_confidence == 0.82
    assert result.llm_matched_precedent_ids == ("case-1",)
    assert "case-1 판례" in result.judgment_summary
    assert "저작권 침해가 의심" in result.judgment_summary
    assert "GPT" not in (result.judge_note or "")
    assert "보호되는 창작적 표현인지에 대한 사람 검토" in result.missing_factors
    assert "시장 영향" in result.missing_factors
    assert judge.request.evidence[0]["source_document_id"] == "doc-1"


def test_llm_hallucinated_precedent_id_falls_back_to_rules():
    result = LegalRiskEngine([_case()], judge=_FakeJudge(("invented-case",))).assess(
        max_similarity=0.9,
        coverage=0.5,
        longest_span=200,
        legal_tags=["reproduction"],
    )
    assert result.judgment_method == "rule_fallback"
    assert result.llm_verdict is None
    assert result.precedent_ids == ("case-1",)
    assert "GPT" not in (result.judge_note or "")


def test_server_keeps_human_review_even_if_judge_returns_false():
    result = LegalRiskEngine([_case()], judge=_FakeJudge(review_required=False)).assess(
        max_similarity=0.9,
        coverage=0.5,
        longest_span=200,
        legal_tags=["reproduction"],
    )
    assert result.judgment_method == "llm"
    assert result.llm_review_required is True


def test_likely_judgment_without_precedent_id_falls_back():
    result = LegalRiskEngine([_case()], judge=_FakeJudge(())).assess(
        max_similarity=0.9,
        coverage=0.5,
        longest_span=200,
        legal_tags=["reproduction"],
    )
    assert result.judgment_method == "rule_fallback"


def test_openai_judge_uses_strict_schema_and_disables_response_storage():
    captured = {}

    class _Responses:
        def create(self, **kwargs):
            captured.update(kwargs)
            return SimpleNamespace(output_text=json.dumps({
                "verdict": "insufficient_evidence",
                "confidence": 0.35,
                "matched_precedent_ids": ["case-1"],
                "supporting_reasons": [],
                "counter_reasons": ["법적 맥락 부족"],
                "missing_factors": ["의거관계"],
                "review_required": True,
            }, ensure_ascii=False))

    client = SimpleNamespace(responses=_Responses())
    fake = _FakeJudge()
    engine = LegalRiskEngine([_case()], judge=fake)
    engine.assess(
        max_similarity=0.9, coverage=0.5, longest_span=200,
        legal_tags=["reproduction"],
    )
    judge = OpenAILegalJudge(api_key="unused", model="test-gpt", client=client)
    decision = judge.judge(fake.request)

    assert decision.verdict == "insufficient_evidence"
    assert captured["store"] is False
    assert captured["text"]["format"]["type"] == "json_schema"
    assert captured["text"]["format"]["strict"] is True


# ---------------------------------------------------------------------------
# 사람 검토 등급이 인용 순서를 좌우해야 한다
#
# 등급이 없던 시절 운영 스모크 테스트에서 95가합11403(폰트·프로그램)이 근거로
# 인용됐다. 리스트업에서 "저작물 유형이 다르다"며 제외한 사건이다. 수집 단계
# 라벨만으로 순위를 정했기 때문이다.
# ---------------------------------------------------------------------------


def _graded(case_id: str, grade: str | None) -> Precedent:
    return Precedent(
        case_id=case_id,
        title=f"판례 {case_id}",
        source_url=f"https://example.test/{case_id}",
        work_types=("literary",),
        legal_tags=("reproduction",),
        criteria=("실질적 유사성",),
        holding_summary="요약",
        grade=grade,
    )


def _assess(precedents):
    return LegalRiskEngine(precedents).assess(
        max_similarity=0.9, coverage=0.7, longest_span=300,
        legal_tags=["reproduction"], work_type="literary",
    )


def test_reviewed_precedents_are_cited_before_unreviewed():
    result = _assess([_graded("unreviewed", None), _graded("grade-a", "A")])
    assert result.precedent_ids[0] == "grade-a"


def test_unreviewed_remains_searchable_when_reviewed_listup_exists():
    """등급 없음은 미검토 표시일 뿐 검색 제외 조건이 아니다."""
    result = _assess([_graded("grade-c", "C"), _graded("unreviewed", None)])
    assert result.precedent_ids == ("grade-c", "unreviewed")


def test_full_grade_order():
    cases = [_graded("c", "C"), _graded("none", None), _graded("b", "B"), _graded("a", "A")]
    assert _assess(cases).precedent_ids == ("a", "b", "c", "none")


def test_text_relevance_can_outrank_grade_boost():
    direct = replace(
        _graded("direct-unreviewed", None),
        holding_summary="독특한 문장 표현을 그대로 복제한 사안",
    )
    generic = replace(_graded("generic-a", "A"), holding_summary="일반적인 법률 원칙")
    result = LegalRiskEngine([generic, direct]).assess(
        max_similarity=0.9,
        coverage=0.7,
        longest_span=300,
        legal_tags=["reproduction"],
        work_type="literary",
        query_text="독특한 문장 표현을 그대로 복제",
    )
    assert result.precedent_ids[0] == "direct-unreviewed"


def test_c_grade_precedents_remain_reference_candidates():
    """C도 사람이 선정한 참고 판례이므로 후보에 남는다."""
    result = _assess([_graded("only-c", "C")])
    assert result.precedent_ids == ("only-c",)


def test_grades_are_reported_for_cited_precedents():
    result = _assess([_graded("a", "A"), _graded("none", None)])
    assert dict(result.precedent_grades) == {"a": "A"}
    assert "none" not in dict(result.precedent_grades)


def test_loader_reads_grade_from_corpus(tmp_path):
    path = tmp_path / "p.jsonl"
    path.write_text("\n".join([
        json.dumps({"case_id": "g1", "title": "t", "source_url": "https://x/1",
                    "holding_summary": "h", "grade": "a"}, ensure_ascii=False),
        json.dumps({"case_id": "g2", "title": "t", "source_url": "https://x/2",
                    "holding_summary": "h"}, ensure_ascii=False),
    ]), encoding="utf-8")
    loaded = {p.case_id: p.grade for p in load_precedents(path)}
    assert loaded == {"g1": "A", "g2": None}  # 소문자도 정규화된다


def test_operational_corpus_carries_grades():
    """실제 적재본에 등급이 반영돼 있어야 한다(apply_precedent_grades.py 결과)."""
    cases = load_precedents("data/precedents/precedents.jsonl")
    graded = [p for p in cases if p.grade]
    assert len(graded) == 57, f"등급 부여 57건이어야 하는데 {len(graded)}건"
    assert {p.grade for p in graded} == {"A", "B", "C"}


def test_operational_corpus_does_not_cite_excluded_font_program_case():
    cases = load_precedents("data/precedents/precedents.jsonl")
    result = LegalRiskEngine(cases).assess(
        max_similarity=0.9,
        coverage=0.7,
        longest_span=300,
        legal_tags=["reproduction", "derivative_work", "distribution"],
        work_type="literary",
    )
    assert "95가합11403" not in result.precedent_ids


def test_font_program_case_is_available_for_software_search():
    cases = load_precedents("data/precedents/precedents.jsonl")
    target = next(p for p in cases if p.case_id == "95가합11403")
    assert target.work_types == ("software",)
    result = LegalRiskEngine(cases).assess(
        max_similarity=0.9,
        coverage=0.7,
        longest_span=300,
        legal_tags=["reproduction", "derivative_work"],
        work_type="software",
        query_text="폰트파일 컴퓨터프로그램 복제 전환행위",
    )
    assert "95가합11403" in result.precedent_ids